# OpenAI Agents SDK

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

---

## Core Concepts

### Agent Definition
```python
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
| Component | Purpose |
|---|---|
| **Agent** | Model + instructions + tools + handoffs |
| **Runner** | Execution engine (streaming, tracing, guardrails) |
| **Tools** | `@function_tool` decorated functions |
| **Handoffs** | Agent-to-agent delegation |
| **Guardrails** | Input/output validation |
| **Tracing** | Built-in observability |
| **MCP** | Model Context Protocol support |

---

## Tools

### Function Tools
```python
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
```python
# 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
```python
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
```python
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)
```python
# LLM chooses when to handoff based on instructions
agent = Agent(
    instructions="If billing question, handoff to billing_agent."
)
```

#### 2. Tool-Based Handoff (Structured)
```python
@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)
```python
# 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)
```python
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)
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
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)

```python
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
```python
triage = Agent(handoffs=[billing, technical, sales])
billing = Agent(handoffs=[refund_specialist, subscription_specialist])
# Chain: triage → billing → refund_specialist
```

### 2. Parallel Specialists
```python
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
```python
supervisor = Agent(
    name="Supervisor",
    handoffs=[researcher, coder, reviewer, deployer],
    instructions="Coordinate the team. Delegate to specialists."
)
```

---

## Deployment

### FastAPI
```python
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
```dockerfile
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
```python
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
```python
@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
3. **Trace everything** — Built-in tracing is free
4. **Stream for UX** — `run_streamed` for responsive UI
5. **Test with TestModel** — Deterministic, fast unit tests

---

## Related Notes

- [[Frameworks/Framework-Comparison.md]] — vs LangGraph, PydanticAI
- [[Frameworks/LangGraph.md]] — Alternative for graphs
- [[Frameworks/PydanticAI.md]] — Alternative for type-safety
- [[MultiAgent/Coordination-Patterns.md]] — Handoff patterns
- [[Architectures/Human-in-the-Loop.md]] — Guardrails as HITL
- [[ContextEngineering/MCP-Protocol.md]] — MCP integration
- [[Domains/Customer-Facing-Agents.md]] — Triage, handoffs
- [[Production/Observability.md]] — Tracing integration
- [[Frontiers/Agent-Computer-Interface.md]] — 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.*