OpenAI Agents SDK

Official OpenAI framework for building agents. Built on Responses API with native handoffs, guardrails, tracing.

Core Concepts

Agent Definition

from agents import Agent, Runner, function_tool, handoff

# Simple agent
agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant.",
    model="gpt-4o",
    tools=[search_web, calculate],
    handoffs=[billing_agent, technical_agent]
)

# Run
result = await Runner.run(agent, "Help me with my bill")
print(result.final_output)

Key Components

Tools

Function Tools

from agents import function_tool
from pydantic import BaseModel

class SearchInput(BaseModel):
    query: str
    max_results: int = 10

@function_tool
async def search_web(input: SearchInput) -> list[SearchResult]:
    """Search the web for current information."""
    return await search_client.search(input.query, input.max_results)

@function_tool
def calculate(expression: str) -> float:
    """Evaluate a mathematical expression."""
    return safe_eval(expression)

# Agent with tools
agent = Agent(
    name="Assistant",
    tools=[search_web, calculate]
)

Tool Patterns

# Tool with context (dependencies)
@function_tool
async def get_customer(ctx: RunContext[Deps], customer_id: str) -> Customer:
    return await ctx.deps.db.get(customer_id)

# Tool returning structured data
@function_tool
async def analyze_sentiment(text: str) -> SentimentAnalysis:
    return await sentiment_model.analyze(text)

# Tool that can handoff
@function_tool
async def escalate_to_human(reason: str) -> HandoffResult:
    return await human_queue.escalate(reason)

Handoffs (Agent Delegation)

Defining Handoffs

billing_agent = Agent(
    name="Billing Specialist",
    instructions="Handle billing inquiries, refunds, subscriptions.",
    tools=[refund_tool, subscription_tool]
)

technical_agent = Agent(
    name="Technical Support",
    instructions="Handle technical issues, bugs, integrations.",
    tools=[diagnose_tool, logs_tool]
)

# Main agent with handoffs
triage_agent = Agent(
    name="Triage",
    instructions="Route to appropriate specialist.",
    handoffs=[billing_agent, technical_agent]
)

Handoff Configuration

from agents import handoff

# Custom handoff with input transformation
billing_handoff = handoff(
    agent=billing_agent,
    tool_name_override="transfer_to_billing",
    tool_description_override="Transfer to billing for refunds, charges, subscriptions",
    on_handoff=lambda ctx: log_handoff("billing", ctx)
)

agent = Agent(handoffs=[billing_handoff, technical_agent])

Handoff Patterns

1. Explicit Handoff (Agent Decides)

# LLM chooses when to handoff based on instructions
agent = Agent(
    instructions="If billing question, handoff to billing_agent."
)

2. Tool-Based Handoff (Structured)

@function_tool
async def route_to_specialist(category: Literal["billing", "tech", "sales"], summary: str):
    """Route to appropriate specialist with context."""
    return {"category": category, "summary": summary}

agent = Agent(tools=[route_to_specialist])

3. Automatic Handoff (Guardrail)

# Input guardrail triggers handoff
async def detect_billing(ctx, agent, input):
    if "refund" in input.lower() or "charge" in input.lower():
        return GuardrailFunctionOutput(tripwire_triggered=True)

billing_guardrail = InputGuardrail(guardrail_function=detect_billing)
agent = Agent(input_guardrails=[billing_guardrail])

Guardrails

Input Guardrails (Validate Before Agent)

from agents import InputGuardrail, GuardrailFunctionOutput

async def no_pii(ctx, agent, input):
    if detect_pii(input):
        return GuardrailFunctionOutput(
            tripwire_triggered=True,
            output_info="PII detected in input"
        )
    return GuardrailFunctionOutput(tripwire_triggered=False)

agent = Agent(input_guardrails=[no_pii])

Output Guardrails (Validate After Agent)

from agents import OutputGuardrail

class SafeResponse(BaseModel):
    response: str
    contains_advice: bool

async def validate_output(ctx, agent, output):
    parsed = SafeResponse.model_validate_json(output)
    if parsed.contains_advice and "medical" in output.lower():
        return GuardrailFunctionOutput(tripwire_triggered=True)
    return GuardrailFunctionOutput(tripwire_triggered=False)

agent = Agent(output_guardrails=[validate_output])

Structured Outputs

Pydantic Models

from pydantic import BaseModel, Field
from typing import Literal

class SupportTicket(BaseModel):
    category: Literal["billing", "technical", "general"]
    priority: Literal["low", "medium", "high", "urgent"]
    summary: str = Field(max_length=200)
    suggested_response: str

agent = Agent(
    name="Classifier",
    instructions="Classify and draft response.",
    output_type=SupportTicket  # Enforces structured output
)

result = await Runner.run(agent, "My bill is wrong!")
print(result.final_output.category)  # "billing"
print(result.final_output.priority)  # "high"

Tracing & Observability

Built-in Tracing

from agents import trace, Runner

with trace("customer_support_session"):
    result = await Runner.run(agent, "Help me!")
    # Automatic: spans for LLM calls, tools, handoffs, guardrails

Custom Spans

from agents import trace, Span

with trace("custom_operation") as t:
    t.span("database_query", data={"query": "SELECT *"})
    result = await db.query(...)
    t.span("processing", data={"rows": len(result)})

OpenTelemetry Export

from agents.tracing import set_trace_processors
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace.export import BatchSpanProcessor

set_trace_processors([
    BatchSpanProcessor(OTLPSpanExporter(endpoint="http://otel:4317"))
])

Streaming

Text Streaming

result = Runner.run_streamed(agent, "Query")
async for event in result.stream_events():
    if event.type == "raw_response_event":
        print(event.data.delta, end="", flush=True)

Structured Streaming

async for event in Runner.run_streamed(agent, "Query").stream_events():
    if event.type == "run_item_stream_event":
        item = event.item
        if item.type == "tool_call":
            print(f"Calling {item.tool.name}...")
        elif item.type == "tool_output":
            print(f"Result: {item.output}")
        elif item.type == "handoff":
            print(f"Handoff to {item.target_agent.name}")

Model Context Protocol (MCP)

Client

from agents.mcp import MCPClient, MCPClientConfig

client = MCPClient(MCPClientConfig(
    url="http://localhost:8000/mcp",
    headers={"Authorization": "Bearer token"}
))

tools = await client.list_tools()
agent = Agent(tools=tools)

Server

from agents.mcp import MCPServer, tool

@tool
async def my_tool(param: str) -> str:
    return f"Result: {param}"

server = MCPServer(tools=[my_tool])
server.run(host="0.0.0.0", port=8000)

Computer Use (Desktop Control)

from agents import Agent, Runner
from agents.tools import computer_tool

# Agent with computer control
computer_agent = Agent(
    name="Computer Operator",
    instructions="Use the computer to complete tasks.",
    tools=[computer_tool],
    model="computer-use-preview"  # Special model
)

result = await Runner.run(computer_agent, "Open browser and search for AI agents")

Multi-Agent Patterns

1. Handoff Chain

triage = Agent(handoffs=[billing, technical, sales])
billing = Agent(handoffs=[refund_specialist, subscription_specialist])
# Chain: triage → billing → refund_specialist

2. Parallel Specialists

async def run_parallel(query):
    results = await asyncio.gather(
        Runner.run(research_agent, query),
        Runner.run(analysis_agent, query),
        Runner.run(writing_agent, query)
    )
    return synthesize(results)

3. Supervisor Pattern

supervisor = Agent(
    name="Supervisor",
    handoffs=[researcher, coder, reviewer, deployer],
    instructions="Coordinate the team. Delegate to specialists."
)

Deployment

FastAPI

from fastapi import FastAPI
from agents import Agent, Runner

app = FastAPI()
agent = Agent(...)

@app.post("/run")
async def run_agent(request: RunRequest):
    result = await Runner.run(agent, request.input)
    return {"output": result.final_output}

@app.post("/stream")
async def stream_agent(request: RunRequest):
    async def gen():
        async for event in Runner.run_streamed(agent, request.input).stream_events():
            if event.type == "raw_response_event":
                yield f"data: {event.data.delta}\n\n"
    return StreamingResponse(gen(), media_type="text/event-stream")

Docker

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

Testing

Unit Testing

from agents.testing import TestModel, FunctionModel

# TestModel for deterministic responses
agent = Agent(model=TestModel())

# FunctionModel for custom behavior
def mock_model(messages, tools):
    return {"tool_calls": [...], "text": "Mock response"}

agent = Agent(model=FunctionModel(mock_model))

# Run in tests
result = await Runner.run(agent, "Test input")
assert "expected" in result.final_output

Integration Testing

@pytest.mark.asyncio
async def test_full_flow():
    result = await Runner.run(
        agent,
        "Test query",
        run_config=RunConfig(
            tracing_disabled=True,
            max_turns=10
        )
    )
    assert result.final_output.category == "billing"

Best Practices

1. Use structured outputs — output_type for validation
2. Guardrail early — Input guardrails prevent bad state
3. Handoff with context — Pass summary, not full history
4. Trace everything — Built-in tracing is free
5. Stream for UX — run_streamed for responsive UI
6. Test with TestModel — Deterministic, fast unit tests

Related Notes

• Framework Comparison — vs LangGraph, PydanticAI
• LangGraph — Alternative for graphs
• PydanticAI — Alternative for type-safety
• Coordination Patterns — Handoff patterns
• Human in the Loop — Guardrails as HITL
• MCP Protocol — MCP integration
• Customer Facing Agents — Triage, handoffs
• Observability — Tracing integration
• Agent Computer Interface — Computer use tool

Sources: OpenAI Agents SDK docs (openai.github.io/openai-agents-python), GitHub repo, Responses API docs, Computer Use API, MCP spec, OpenAI cookbook examples.