← Back to DevBytes
AI August 16, 2026 14 min read

Building a Log Analysis Agent with OpenAI Agents SDK: Complete Guide

Building a Log Analysis Agent with OpenAI Agents SDK: Complete Guide

Log analysis is one of those tasks that every engineering team needs but few enjoy. You ship a service, it runs in production, and at 2 AM something breaks. Now you're staring at thousands of log lines trying to figure out what happened. The OpenAI Agents SDK gives you a structured way to build an autonomous agent that can read logs, reason about them, correlate events, and produce actionable insights — all without you babysitting every prompt.

In this guide, we'll build a complete log analysis agent from scratch. The agent will ingest log files, identify anomalies, trace error chains across services, and generate a human-readable incident report. By the end, you'll have a working system you can extend for your own infrastructure.

What Is the OpenAI Agents SDK?

The OpenAI Agents SDK is a Python framework for building agentic applications. It provides abstractions for Agent, Runner, tools, handoffs, guardrails, and tracing. Unlike raw chat completions where you manage conversation state yourself, the Agents SDK handles the agent loop — calling tools, observing results, and deciding when to stop — so you can focus on defining what the agent should do rather than how it iterates.

Key concepts we'll use:

Why Build a Log Analysis Agent?

Traditional log tools like grep, awk, or even ELK dashboards are powerful, but they require you to know what you're looking for. An agent can reason inductively: it can spot a pattern you didn't think to query for, correlate a spike in 5xx errors with a deployment that happened 12 minutes earlier, and explain the likely root cause in plain English.

Specific benefits of an agentic approach:

Project Setup

Let's start by setting up the project. Create a new directory and install the dependencies.

mkdir log-agent && cd log-agent
python -m venv .venv
source .venv/bin/activate
pip install openai-agents pydantic python-dotenv

Create a .env file with your API key:

OPENAI_API_KEY=sk-your-key-here

Now create the project structure:

log-agent/
├── .env
├── main.py
├── agents/
│   ├── __init__.py
│   ├── triage.py
│   ├── investigator.py
│   └── reporter.py
├── tools/
│   ├── __init__.py
│   ├── log_tools.py
│   └── deploy_tools.py
├── models.py
└── samples/
    └── app.log

Defining the Data Models

Structured outputs keep the agent honest. Instead of free-form text that might drift, we define exactly what an incident report looks like. Put this in models.py:

from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime


class LogEntry(BaseModel):
    timestamp: str
    level: str
    service: str
    message: str
    trace_id: Optional[str] = None


class Anomaly(BaseModel):
    description: str = Field(description="What the anomaly is")
    severity: str = Field(description="low, medium, high, or critical")
    affected_service: str
    first_seen: str
    last_seen: str
    sample_log_line: str


class RootCauseHypothesis(BaseModel):
    hypothesis: str
    confidence: float = Field(ge=0.0, le=1.0)
    supporting_evidence: list[str]
    contradicting_evidence: list[str]


class IncidentReport(BaseModel):
    title: str
    summary: str
    severity: str
    anomalies: list[Anomaly]
    root_cause: RootCauseHypothesis
    recommended_actions: list[str]
    timeline: list[str]

These models serve as a contract. The reporter agent will be configured to return an IncidentReport, which means downstream code can rely on the shape of the data.

Sample Log Data

For this tutorial we'll use a synthetic log file so you can run everything locally. Create samples/app.log:

2025-01-15T10:00:01Z INFO  auth-service  User login successful trace_id=abc123
2025-01-15T10:00:05Z INFO  api-gateway   GET /api/orders 200 trace_id=abc123
2025-01-15T10:01:12Z INFO  deploy-service  Deploying orders-service v2.4.1
2025-01-15T10:01:45Z INFO  deploy-service  Deployment orders-service v2.4.1 complete
2025-01-15T10:02:03Z WARN  orders-service  Database connection pool at 80% capacity trace_id=def456
2025-01-15T10:02:30Z ERROR orders-service  Failed to query inventory table: connection timeout trace_id=def456
2025-01-15T10:02:31Z ERROR api-gateway   POST /api/orders 500 trace_id=def456
2025-01-15T10:03:00Z ERROR orders-service  Failed to query inventory table: connection timeout trace_id=ghi789
2025-01-15T10:03:01Z ERROR api-gateway   POST /api/orders 500 trace_id=ghi789
2025-01-15T10:03:15Z WARN  orders-service  Database connection pool at 95% capacity trace_id=jkl012
2025-01-15T10:03:30Z ERROR orders-service  Circuit breaker opened for inventory-db trace_id=jkl012
2025-01-15T10:04:00Z ERROR api-gateway   POST /api/orders 503 trace_id=jkl012
2025-01-15T10:05:00Z INFO  alert-manager  PagerDuty incident triggered: orders-service errors

There's a clear story here: a deployment at 10:01 was followed by database connection pool exhaustion and a cascade of 500/503 errors. The agent should be able to discover this on its own.

Building the Tools

Tools are how the agent interacts with the outside world. We'll create tools for reading logs, filtering by trace ID, and querying deployment history. Put this in tools/log_tools.py:

from agents import function_tool
from pathlib import Path
from models import LogEntry
import re

LOG_PATH = Path(__file__).parent.parent / "samples" / "app.log"

LOG_PATTERN = re.compile(
    r'(?P\S+)\s+(?P\w+)\s+(?P\S+)\s+(?P.+?)(?:\s+trace_id=(?P\S+))?$'
)


def _parse_log_line(line: str) -> LogEntry | None:
    match = LOG_PATTERN.match(line.strip())
    if not match:
        return None
    return LogEntry(
        timestamp=match.group("timestamp"),
        level=match.group("level"),
        service=match.group("service"),
        message=match.group("message").strip(),
        trace_id=match.group("trace_id"),
    )


def _load_logs() -> list[LogEntry]:
    entries = []
    for line in LOG_PATH.read_text().splitlines():
        entry = _parse_log_line(line)
        if entry:
            entries.append(entry)
    return entries


@function_tool
def get_all_logs() -> list[dict]:
    """Return every log entry from the application log file."""
    return [entry.model_dump() for entry in _load_logs()]


@function_tool
def filter_logs_by_level(level: str) -> list[dict]:
    """Return log entries matching a specific level (INFO, WARN, ERROR)."""
    level = level.upper()
    return [
        entry.model_dump()
        for entry in _load_logs()
        if entry.level.upper() == level
    ]


@function_tool
def get_logs_by_trace_id(trace_id: str) -> list[dict]:
    """Return all log entries sharing a given trace ID, ordered by timestamp."""
    return [
        entry.model_dump()
        for entry in _load_logs()
        if entry.trace_id == trace_id
    ]


@function_tool
def get_logs_for_service(service: str) -> list[dict]:
    """Return all log entries produced by a specific service."""
    return [
        entry.model_dump()
        for entry in _load_logs()
        if entry.service == service
    ]

Now the deployment tools in tools/deploy_tools.py. In a real system these would hit your CI/CD API; here we simulate with a static record:

from agents import function_tool


DEPLOYMENTS = [
    {
        "service": "orders-service",
        "version": "v2.4.1",
        "timestamp": "2025-01-15T10:01:12Z",
        "commit": "a1b2c3d",
        "author": "jdoe",
        "changes": "Increased connection pool max size from 10 to 50",
    },
    {
        "service": "auth-service",
        "version": "v1.8.0",
        "timestamp": "2025-01-15T08:00:00Z",
        "commit": "e4f5g6h",
        "author": "asmith",
        "changes": "Minor dependency bump",
    },
]


@function_tool
def get_recent_deployments(hours: int = 24) -> list[dict]:
    """Return deployments from the last N hours. Defaults to 24."""
    return DEPLOYMENTS


@function_tool
def get_deployment_for_service(service: str) -> list[dict]:
    """Return deployment history for a specific service."""
    return [d for d in DEPLOYMENTS if d["service"] == service]

Each tool has a docstring. The SDK sends these docstrings to the model so it knows when and how to call each tool. Write them carefully — they are effectively your agent's API documentation.

Building the Agents

We'll use a multi-agent architecture with three specialized agents that hand off to each other:

This separation matters. A single mega-agent with twenty tools tends to call the wrong ones. Specialized agents with focused toolsets perform better and are easier to debug.

The Triage Agent

Create agents/triage.py:

from agents import Agent
from tools.log_tools import get_all_logs, filter_logs_by_level

triage_agent = Agent(
    name="Triage Agent",
    instructions=(
        "You are a log triage specialist. Your job is to scan application logs "
        "and identify anomalies: spikes in error rates, warnings that precede "
        "failures, unusual patterns, or anything that suggests an incident.\n\n"
        "Steps:\n"
        "1. Call get_all_logs to load the full log file.\n"
        "2. Look for ERROR and WARN entries and group them by service and time window.\n"
        "3. Identify any cascading patterns (e.g., one service's errors causing "
        "another service's failures).\n"
        "4. Produce a concise list of anomalies with severity ratings.\n\n"
        "Be precise. Quote actual log lines as evidence. Do not speculate about "
        "root causes yet — that is the investigator's job."
    ),
    tools=[get_all_logs, filter_logs_by_level],
)

The Investigator Agent

Create agents/investigator.py. This agent has access to trace-level tools and deployment history so it can form a root cause hypothesis:

from agents import Agent
from tools.log_tools import (
    get_logs_by_trace_id,
    get_logs_for_service,
    filter_logs_by_level,
)
from tools.deploy_tools import get_recent_deployments, get_deployment_for_service

investigator_agent = Agent(
    name="Investigator Agent",
    instructions=(
        "You are a senior incident investigator. You receive a list of anomalies "
        "from the triage agent and must determine the root cause.\n\n"
        "Steps:\n"
        "1. For each anomaly, pull the full log context for the affected service "
        "using get_logs_for_service.\n"
        "2. For any trace_id mentioned in errors, use get_logs_by_trace_id to "
        "follow the request across services.\n"
        "3. Check get_recent_deployments to see if a deployment coincides with "
        "the anomaly's first_seen timestamp.\n"
        "4. Form a root cause hypothesis with a confidence score between 0 and 1.\n"
        "5. List supporting and contradicting evidence honestly.\n\n"
        "Prevent confirmation bias: actively look for evidence that contradicts "
        "your leading hypothesis before committing to it."
    ),
    tools=[
        get_logs_by_trace_id,
        get_logs_for_service,
        filter_logs_by_level,
        get_recent_deployments,
        get_deployment_for_service,
    ],
)

The Reporter Agent

Create agents/reporter.py. This agent produces the final structured output:

from agents import Agent
from models import IncidentReport

reporter_agent = Agent(
    name="Reporter Agent",
    instructions=(
        "You are an incident reporter. You receive the triage findings and the "
        "investigator's root cause hypothesis. Your job is to synthesize them "
        "into a structured IncidentReport.\n\n"
        "Requirements:\n"
        "- The title should be a one-line description of the incident.\n"
        "- The summary should be 2-3 sentences a non-engineer could understand.\n"
        "- Severity must be one of: low, medium, high, critical.\n"
        "- The timeline should be a chronological list of key events as strings.\n"
        "- Recommended actions should be concrete and actionable, not generic.\n"
        "- Be honest about uncertainty. If confidence is below 0.7, say so in "
        "the summary."
    ),
    output_type=IncidentReport,
)

Setting output_type=IncidentReport tells the SDK to constrain the agent's final response to the Pydantic model. The agent will still reason in natural language internally, but its final output will be validated against the schema.

Wiring It Together with Handoffs

Now we connect the agents. The triage agent hands off to the investigator, who hands off to the reporter. Update agents/triage.py to include the handoff:

from agents import Agent
from tools.log_tools import get_all_logs, filter_logs_by_level
from agents.investigator import investigator_agent

triage_agent = Agent(
    name="Triage Agent",
    instructions=(
        "You are a log triage specialist. Your job is to scan application logs "
        "and identify anomalies: spikes in error rates, warnings that precede "
        "failures, unusual patterns, or anything that suggests an incident.\n\n"
        "Steps:\n"
        "1. Call get_all_logs to load the full log file.\n"
        "2. Look for ERROR and WARN entries and group them by service and time window.\n"
        "3. Identify any cascading patterns.\n"
        "4. Produce a concise list of anomalies with severity ratings.\n"
        "5. Once you have your anomaly list, hand off to the Investigator Agent "
        "and pass along your findings.\n\n"
        "Be precise. Quote actual log lines as evidence."
    ),
    tools=[get_all_logs, filter_logs_by_level],
    handoffs=[investigator_agent],
)

And update agents/investigator.py to hand off to the reporter:

from agents import Agent
from tools.log_tools import (
    get_logs_by_trace_id,
    get_logs_for_service,
    filter_logs_by_level,
)
from tools.deploy_tools import get_recent_deployments, get_deployment_for_service
from agents.reporter import reporter_agent

investigator_agent = Agent(
    name="Investigator Agent",
    instructions=(
        "You are a senior incident investigator. You receive anomalies from the "
        "triage agent and must determine the root cause.\n\n"
        "Steps:\n"
        "1. For each anomaly, pull full log context for the affected service.\n"
        "2. Follow trace_ids across services.\n"
        "3. Check recent deployments for temporal correlation.\n"
        "4. Form a root cause hypothesis with confidence 0-1.\n"
        "5. List supporting and contradicting evidence.\n"
        "6. Hand off to the Reporter Agent with your full findings.\n\n"
        "Prevent confirmation bias: actively seek contradicting evidence."
    ),
    tools=[
        get_logs_by_trace_id,
        get_logs_for_service,
        filter_logs_by_level,
        get_recent_deployments,
        get_deployment_for_service,
    ],
    handoffs=[reporter_agent],
)

The Main Entry Point

Now create main.py to run the whole pipeline:

import asyncio
import json
from dotenv import load_dotenv
from agents import Runner
from agents.triage import triage_agent
from models import IncidentReport

load_dotenv()


async def main():
    print("Starting log analysis agent...\n")

    result = await Runner.run(
        triage_agent,
        input=(
            "Please analyze the application logs and identify any anomalies "
            "or incidents. Investigate thoroughly and produce a full incident "
            "report."
        ),
    )

    # The final output is an IncidentReport because the reporter agent
    # has output_type=IncidentReport and is the last agent in the chain.
    report: IncidentReport = result.final_output

    print("=" * 60)
    print(f"INCIDENT: {report.title}")
    print(f"Severity: {report.severity.upper()}")
    print("=" * 60)
    print(f"\nSummary:\n{report.summary}\n")

    print("Anomalies Detected:")
    for anomaly in report.anomalies:
        print(f"  [{anomaly.severity.upper()}] {anomaly.description}")
        print(f"    Service: {anomaly.affected_service}")
        print(f"    Window:  {anomaly.first_seen} -> {anomaly.last_seen}")
        print(f"    Sample:  {anomaly.sample_log_line}")
        print()

    print(f"Root Cause Hypothesis (confidence: {report.root_cause.confidence:.0%}):")
    print(f"  {report.root_cause.hypothesis}")
    print("\n  Supporting evidence:")
    for ev in report.root_cause.supporting_evidence:
        print(f"    + {ev}")
    print("\n  Contradicting evidence:")
    for ev in report.root_cause.contradicting_evidence:
        print(f"    - {ev}")

    print("\nTimeline:")
    for event in report.timeline:
        print(f"  {event}")

    print("\nRecommended Actions:")
    for i, action in enumerate(report.recommended_actions, 1):
        print(f"  {i}. {action}")

    # Optionally save the full report as JSON
    with open("incident_report.json", "w") as f:
        f.write(report.model_dump_json(indent=2))
    print("\nFull report saved to incident_report.json")


if __name__ == "__main__":
    asyncio.run(main())

Run it:

python main.py

You should see the agent load the logs, identify the database connection pool exhaustion, correlate it with the v2.4.1 deployment, and produce a structured report. The exact wording will vary because the model is non-deterministic, but the structure will be consistent because of the Pydantic output type.

Adding Tracing for Debugging

One of the SDK's best features is built-in tracing. Every tool call, every handoff, every LLM invocation is recorded. Enable it to see exactly what your agents did:

from agents import trace

async def main():
    with trace("Log Analysis Run"):
        result = await Runner.run(
            triage_agent,
            input="Analyze the application logs and produce an incident report.",
        )
        # ... rest of the code

When you set OPENAI_API_KEY, traces are sent to the OpenAI dashboard where you can inspect each step. This is invaluable when the agent produces a surprising result and you need to understand its reasoning chain.

Best Practices

After building this system, here are the practices that make the biggest difference in production:

Extending the System

Once you have the foundation, there are several valuable extensions:

Conclusion

Building a log analysis agent with the OpenAI Agents SDK demonstrates the framework's core strength: you compose specialized agents, give each one a focused set of tools, and let handoffs carry context through an investigation pipeline. The triage agent scans broadly, the investigator digs deep, and the reporter synthesizes — mirroring how a real incident response team works. The structured output guarantees that what comes out the other end is machine-readable and consistent, while the tracing system gives you full visibility into how the agent reached its conclusions. Start with the sample logs in this guide, then swap in your real log sources and deployment APIs. The architecture scales: add more tools, add more agents, and the handoff pattern keeps everything organized. The result is an always-on analyst that turns raw log noise into clear, actionable incident reports.

— Ad —

Google AdSense will appear here after approval

← Back to all articles