Openai Agents Sdk

OpenAI Agents SDK (Python)

Use this skill when developing AI agents using OpenAI Agents SDK (openai-agents package).

Quick Reference

Installation

pip install openai-agents

Environment Variables

# OpenAI (direct)
OPENAI_API_KEY=sk-...
LLM_PROVIDER=openai

# Azure OpenAI (via LiteLLM)
LLM_PROVIDER=azure
AZURE_API_KEY=...
AZURE_API_BASE=https://your-resource.openai.azure.com
AZURE_API_VERSION=2024-12-01-preview

Basic Agent

from agents import Agent, Runner

agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant.",
    model="gpt-5.4",  # or "gpt-5.4-mini", "gpt-5.4-nano"
)

# Synchronous
result = Runner.run_sync(agent, "Tell me a joke")
print(result.final_output)

# Asynchronous
result = await Runner.run(agent, "Tell me a joke")

Key Patterns

Preferred: Live Docs via MCP

Model names and API details change frequently. When available, consult the OpenAI Developer Docs MCP server (openaiDeveloperDocs) before relying on the static references below.

Setup (Codex CLI):

codex mcp add openaiDeveloperDocs --url https://developers.openai.com/mcp

Or config (~/.codex/config.toml, VS Code .vscode/mcp.json, Cursor ~/.cursor/mcp.json):

[mcp_servers.openaiDeveloperDocs]
url = "https://developers.openai.com/mcp"

Key tools: mcp__openaiDeveloperDocs__search_openai_docs, fetch_openai_doc, list_api_endpoints, get_openapi_spec.

Rules: Cite fetched docs. Never speculate on field names, defaults, or current model IDs — fetch first. Keep quotes under 125 chars.

Fallback when MCP is unavailable: https://developers.openai.com/api/docs/llms.txt (plain-text index of all API docs; each entry has a .md twin at /api/docs/<slug>.md).

Reference Documentation

Offline/quick-lookup snippets. Verify model names and API signatures against the MCP or docs when accuracy matters.

• agents.md - Agent creation, Azure/LiteLLM integration
• tools.md - Function tools, hosted tools, agents as tools
• structured-output.md - Pydantic output, AgentOutputSchema
• streaming.md - Streaming patterns, SSE with FastAPI
• handoffs.md - Agent delegation
• guardrails.md - Input/output validation
• sessions.md - Sessions, conversation history
• patterns.md - Multi-agent workflows, LLM as judge, tracing

Official Documentation

• Docs: https://openai.github.io/openai-agents-python/
• Examples: https://github.com/openai/openai-agents-python/tree/main/examples
• Major update: https://openai.com/index/the-next-evolution-of-the-agents-sdk/
• Docs MCP setup: https://developers.openai.com/learn/docs-mcp
• Docs index (llms.txt): https://developers.openai.com/api/docs/llms.txt
• Current model IDs: https://platform.openai.com/docs/models

Agents

Basic Agent Creation

from agents import Agent, Runner

agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant.",
    model="gpt-5.4",  # or "gpt-5.4-mini", "gpt-5.4-nano"
)

# Synchronous execution
result = Runner.run_sync(agent, "Tell me a joke")
print(result.final_output)

# Asynchronous execution
result = await Runner.run(agent, "Tell me a joke")

Azure OpenAI (LiteLLM)

import os
from typing import Union
from agents import Agent, ModelSettings
from agents.extensions.models.litellm_model import LitellmModel

LLM_PROVIDER = os.getenv("LLM_PROVIDER", "azure")
MODEL = os.getenv("MODEL", "gpt-5.4")

def get_model() -> Union[str, LitellmModel]:
    """Get model based on provider."""
    if LLM_PROVIDER == "azure":
        # azure/ prefix tells LiteLLM to use Azure endpoint
        return LitellmModel(model=f"azure/{MODEL}")
    # Direct OpenAI
    return MODEL

agent = Agent(
    name="Assistant",
    instructions="You are helpful.",
    model=get_model(),  # Works with both Azure and OpenAI
)

Dynamic System Prompt

from agents import Agent, Runner, RunContextWrapper

def dynamic_instructions(
    ctx: RunContextWrapper[dict], agent: Agent[dict]
) -> str:
    user_name = ctx.context.get("user_name", "User")
    return f"You are helping {user_name}. Be friendly and helpful."

agent = Agent(
    name="DynamicBot",
    instructions=dynamic_instructions,  # Function instead of string
    model="gpt-5.4",
)

result = await Runner.run(
    agent,
    "Hello!",
    context={"user_name": "Alice"},
)

Loading Prompts from Files

from pathlib import Path

PROMPTS_DIR = Path(__file__).parent / "prompts"

def load_prompt(filename: str) -> str:
    return (PROMPTS_DIR / filename).read_text(encoding="utf-8")

agent = Agent(
    name="Planner",
    instructions=load_prompt("planner.md"),
    model="gpt-5.4",
)

Agent Configuration Options

Guardrails

Input Guardrails

Validate and filter input before the agent processes it:

from agents import Agent, Runner, input_guardrail
from agents import GuardrailFunctionOutput, RunContextWrapper

@input_guardrail
async def check_appropriate(
    ctx: RunContextWrapper, agent: Agent, input: str
) -> GuardrailFunctionOutput:
    # Check input for inappropriate content
    is_inappropriate = "bad_word" in input.lower()
    return GuardrailFunctionOutput(
        tripwire_triggered=is_inappropriate,
        output_info="Inappropriate content detected" if is_inappropriate else None,
    )

@input_guardrail
async def check_length(
    ctx: RunContextWrapper, agent: Agent, input: str
) -> GuardrailFunctionOutput:
    if len(input) > 10000:
        return GuardrailFunctionOutput(
            tripwire_triggered=True,
            output_info="Input too long (max 10000 characters)",
        )
    return GuardrailFunctionOutput(tripwire_triggered=False)

agent = Agent(
    name="SafeAgent",
    instructions="Be helpful.",
    input_guardrails=[check_appropriate, check_length],
)

Output Guardrails

Validate agent output before returning:

from agents import Agent, output_guardrail
from agents import GuardrailFunctionOutput, RunContextWrapper

@output_guardrail
async def check_no_pii(
    ctx: RunContextWrapper, agent: Agent, output: str
) -> GuardrailFunctionOutput:
    # Check for potential PII in output
    import re

    # Simple email pattern check
    has_email = bool(re.search(r'\b[\w.-]+@[\w.-]+\.\w+\b', output))
    # Simple phone pattern check
    has_phone = bool(re.search(r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b', output))

    if has_email or has_phone:
        return GuardrailFunctionOutput(
            tripwire_triggered=True,
            output_info="Output contains potential PII",
        )
    return GuardrailFunctionOutput(tripwire_triggered=False)

agent = Agent(
    name="PIISafeAgent",
    instructions="Help users with their questions.",
    output_guardrails=[check_no_pii],
)

Guardrail with Context

from agents import Agent, input_guardrail
from agents import GuardrailFunctionOutput, RunContextWrapper

@input_guardrail
async def check_user_permissions(
    ctx: RunContextWrapper[dict], agent: Agent, input: str
) -> GuardrailFunctionOutput:
    user_role = ctx.context.get("user_role", "guest")

    # Check if user can access admin features
    if "admin" in input.lower() and user_role != "admin":
        return GuardrailFunctionOutput(
            tripwire_triggered=True,
            output_info="Admin access not permitted for your role",
        )
    return GuardrailFunctionOutput(tripwire_triggered=False)

agent = Agent(
    name="RoleBasedAgent",
    instructions="Help users based on their role.",
    input_guardrails=[check_user_permissions],
)

# User without admin access
result = await Runner.run(
    agent,
    "Show me admin settings",
    context={"user_role": "user"},
)
# -> Guardrail triggered

Tool Guardrails

Validate tool inputs before execution:

from agents import function_tool, tool_guardrail
from agents import ToolGuardrailFunctionOutput, RunContextWrapper
from typing import Annotated

@tool_guardrail
async def validate_file_path(
    ctx: RunContextWrapper, agent: Agent, tool_input: dict
) -> ToolGuardrailFunctionOutput:
    path = tool_input.get("file_path", "")

    # Block access to sensitive directories
    forbidden = ["/etc", "/root", "~/.ssh"]
    for forbidden_path in forbidden:
        if path.startswith(forbidden_path):
            return ToolGuardrailFunctionOutput(
                tripwire_triggered=True,
                output_info=f"Access to {forbidden_path} not allowed",
            )
    return ToolGuardrailFunctionOutput(tripwire_triggered=False)

@function_tool(guardrails=[validate_file_path])
def read_file(file_path: Annotated[str, "Path to file"]) -> str:
    """Read contents of a file."""
    with open(file_path) as f:
        return f.read()

Handling Guardrail Errors

from agents import Agent, Runner, InputGuardrailTripwireTriggered

agent = Agent(
    name="SafeBot",
    instructions="Be helpful.",
    input_guardrails=[check_appropriate],
)

try:
    result = await Runner.run(agent, "Some bad_word input")
except InputGuardrailTripwireTriggered as e:
    print(f"Input blocked: {e.guardrail_result.output_info}")

GuardrailFunctionOutput Fields

Handoffs

Basic Handoffs

Handoffs allow agents to delegate tasks to specialized agents:

from agents import Agent, handoff

billing_agent = Agent(
    name="BillingAgent",
    instructions="Handle billing questions. You can help with invoices, payments, and subscriptions.",
)

support_agent = Agent(
    name="SupportAgent",
    instructions="Handle general support. Handoff billing questions to the billing agent.",
    handoffs=[billing_agent],
)

# LLM automatically decides when to delegate to another agent
result = await Runner.run(support_agent, "I have a question about my invoice")

Multiple Handoffs

billing_agent = Agent(
    name="BillingAgent",
    instructions="Handle billing and payment questions.",
)

technical_agent = Agent(
    name="TechnicalAgent",
    instructions="Handle technical issues and troubleshooting.",
)

sales_agent = Agent(
    name="SalesAgent",
    instructions="Handle sales inquiries and pricing.",
)

triage_agent = Agent(
    name="TriageAgent",
    instructions="""You are a customer service triage agent.
    Route customers to the appropriate specialist:
    - Billing questions -> BillingAgent
    - Technical issues -> TechnicalAgent
    - Sales/pricing -> SalesAgent
    """,
    handoffs=[billing_agent, technical_agent, sales_agent],
)

result = await Runner.run(triage_agent, "My app keeps crashing")
# -> Delegates to TechnicalAgent

Handoff with Context

from agents import Agent, handoff, RunContextWrapper

def escalation_instructions(
    ctx: RunContextWrapper[dict], agent: Agent[dict]
) -> str:
    priority = ctx.context.get("priority", "normal")
    return f"""You are handling an escalated case.
    Priority level: {priority}
    Be thorough and professional."""

escalation_agent = Agent(
    name="EscalationAgent",
    instructions=escalation_instructions,
)

support_agent = Agent(
    name="SupportAgent",
    instructions="Handle support. Escalate complex issues.",
    handoffs=[escalation_agent],
)

result = await Runner.run(
    support_agent,
    "This is urgent, I need help immediately!",
    context={"priority": "high"},
)

Handoff vs Agents as Tools

Handoff Example

# Child takes over the conversation
support_agent = Agent(
    name="Support",
    handoffs=[billing_agent],  # Billing agent takes over
)

Agent as Tool Example

# Parent stays in control
orchestrator = Agent(
    name="Orchestrator",
    tools=[
        billing_agent.as_tool(
            tool_name="check_billing",
            tool_description="Get billing info",
        ),
    ],
)

Message Filtering

Control what messages are passed during handoff:

from agents import Agent, handoff
from agents import TResponseInputItem

def filter_messages(messages: list[TResponseInputItem]) -> list[TResponseInputItem]:
    # Only keep last 5 messages
    return messages[-5:]

specialist = Agent(
    name="Specialist",
    instructions="Handle specialized tasks.",
)

agent = Agent(
    name="Router",
    instructions="Route to specialist when needed.",
    handoffs=[
        handoff(
            agent=specialist,
            input_filter=filter_messages,
        ),
    ],
)

Patterns

Multi-Agent Workflow Pipeline

Example: 3-stage pipeline (ProductSelector -> SetOptimizer -> PlanGenerator)

from pathlib import Path
from pydantic import BaseModel, Field
from agents import Agent, AgentOutputSchema, ModelSettings, RunConfig, Runner
from openai.types.responses import ResponseTextDeltaEvent
from openai.types.shared.reasoning import Reasoning
from collections.abc import AsyncIterator

# --- Pydantic Output Schemas ---

class ProductLite(BaseModel):
    product_id: str
    name: str
    score: float = Field(ge=0, le=1)

class ProductsOutput(BaseModel):
    products: list[ProductLite]

class TravelSet(BaseModel):
    set_id: str  # "compact", "balanced", "extended"
    name: str
    product_ids: list[str]

class SetsOutput(BaseModel):
    sets: list[TravelSet]
    recommended_set_id: str

# --- Prompt Loading ---

PROMPTS_DIR = Path(__file__).parent / "prompts"

def load_prompt(name: str) -> str:
    return (PROMPTS_DIR / name).read_text(encoding="utf-8")

# --- Agents ---

# Step 1: Select products (structured output)
product_selector = Agent(
    name="ProductSelector",
    instructions=load_prompt("product_selector.md"),
    model=get_model(),
    model_settings=ModelSettings(max_tokens=64000),
    output_type=AgentOutputSchema(ProductsOutput, strict_json_schema=True),
)

# Step 2: Optimize sets (structured output)
set_optimizer = Agent(
    name="SetOptimizer",
    instructions=load_prompt("set_optimizer.md"),
    model=get_model(),
    model_settings=ModelSettings(
        max_tokens=16000,
        reasoning=Reasoning(effort="low"),
    ),
    output_type=AgentOutputSchema(SetsOutput, strict_json_schema=True),
)

# Step 3: Generate plan (streaming, no schema)
plan_generator = Agent(
    name="PlanGenerator",
    instructions=load_prompt("plan_generator.md"),
    model=get_model(),
    model_settings=ModelSettings(
        max_tokens=32000,
        reasoning=Reasoning(effort="low"),
    ),
    # No output_type = free text for streaming
)

# --- Runner Functions ---

async def select_products(user_prompt: str, context: str) -> list[ProductLite]:
    """Step 1: Select products."""
    result = await Runner.run(
        product_selector,
        input=f"User: {user_prompt}\n\nProducts:\n{context}",
        run_config=RunConfig(
            workflow_name="ProductSelector",
            trace_metadata={"step": "select"},
        ),
    )
    output: ProductsOutput = result.final_output
    return output.products

async def optimize_sets(products: list[dict]) -> tuple[list[TravelSet], str]:
    """Step 2: Create optimized sets."""
    result = await Runner.run(
        set_optimizer,
        input=f"Products:\n{products}",
        run_config=RunConfig(workflow_name="SetOptimizer"),
    )
    output: SetsOutput = result.final_output
    return output.sets, output.recommended_set_id

async def generate_plan_stream(products: list[dict]) -> AsyncIterator[str]:
    """Step 3: Generate plan with streaming."""
    result = Runner.run_streamed(
        plan_generator,
        input=f"Create travel plan for:\n{products}",
        run_config=RunConfig(workflow_name="PlanGenerator"),
    )
    async for event in result.stream_events():
        if event.type == "raw_response_event":
            if isinstance(event.data, ResponseTextDeltaEvent):
                yield event.data.delta

# --- Full Workflow ---

async def travel_workflow(user_prompt: str, products_context: str):
    # Step 1
    products = await select_products(user_prompt, products_context)
    print(f"Selected {len(products)} products")

    # Step 2
    sets, recommended = await optimize_sets([p.model_dump() for p in products])
    print(f"Created {len(sets)} sets, recommended: {recommended}")

    # Step 3 - stream
    async for chunk in generate_plan_stream([p.model_dump() for p in products]):
        print(chunk, end="", flush=True)

LLM as a Judge

Iterative improvement with evaluator agent:

from dataclasses import dataclass
from typing import Literal
from agents import Agent, Runner, TResponseInputItem, trace

@dataclass
class Evaluation:
    score: Literal["pass", "needs_improvement", "fail"]
    feedback: str

generator = Agent(
    name="Generator",
    instructions="Generate content based on feedback.",
)

evaluator = Agent(
    name="Evaluator",
    instructions="Evaluate and provide feedback.",
    output_type=Evaluation,
)

async def generate_with_feedback(prompt: str) -> str:
    inputs: list[TResponseInputItem] = [{"role": "user", "content": prompt}]

    with trace("LLM as a judge"):
        while True:
            gen_result = await Runner.run(generator, inputs)
            inputs = gen_result.to_input_list()

            eval_result = await Runner.run(evaluator, inputs)
            evaluation: Evaluation = eval_result.final_output

            if evaluation.score == "pass":
                return gen_result.final_output

            inputs.append({"role": "user", "content": f"Feedback: {evaluation.feedback}"})

Tracing

Group related agent runs together:

from agents import Agent, Runner, trace, RunConfig

async def workflow(user_input: str):
    with trace("MyWorkflow"):
        # All Runner.run() calls inside this block
        # appear in the same trace
        result1 = await Runner.run(agent1, user_input)
        result2 = await Runner.run(agent2, result1.to_input_list())

    return result2.final_output

# RunConfig for metadata
result = await Runner.run(
    agent,
    input=message,
    run_config=RunConfig(
        workflow_name="ProductSelector",
        trace_metadata={"agent": "selector", "locale": "fi"},
    ),
)

Parallelization

Run multiple agents concurrently:

import asyncio
from agents import Agent, Runner

agent1 = Agent(name="Researcher", instructions="Research topics.")
agent2 = Agent(name="Analyzer", instructions="Analyze data.")
agent3 = Agent(name="Writer", instructions="Write content.")

async def parallel_workflow(topic: str):
    # Run research and analysis in parallel
    research_task = Runner.run(agent1, f"Research: {topic}")
    analysis_task = Runner.run(agent2, f"Analyze: {topic}")

    research_result, analysis_result = await asyncio.gather(
        research_task, analysis_task
    )

    # Combine results for writer
    combined_input = f"""
    Research: {research_result.final_output}
    Analysis: {analysis_result.final_output}
    """

    writer_result = await Runner.run(agent3, combined_input)
    return writer_result.final_output

Routing

Route to specialized agents based on input:

from agents import Agent, Runner, function_tool
from typing import Literal

@function_tool
def classify_intent(query: str) -> Literal["billing", "technical", "sales"]:
    """Classify user intent."""
    # In real app, could use another LLM or classifier
    if "invoice" in query or "payment" in query:
        return "billing"
    elif "error" in query or "bug" in query:
        return "technical"
    return "sales"

router = Agent(
    name="Router",
    instructions="Classify user intent using the classify tool.",
    tools=[classify_intent],
)

agents = {
    "billing": Agent(name="Billing", instructions="Handle billing."),
    "technical": Agent(name="Technical", instructions="Handle tech support."),
    "sales": Agent(name="Sales", instructions="Handle sales."),
}

async def route_and_handle(query: str):
    # First, classify
    router_result = await Runner.run(router, query)
    intent = router_result.final_output  # "billing", "technical", or "sales"

    # Route to specialist
    specialist = agents[intent]
    result = await Runner.run(specialist, query)
    return result.final_output

Deterministic Workflows

Force specific tool execution order:

from agents import Agent, ModelSettings

# Phase 1: Must search
search_agent = Agent(
    name="Searcher",
    instructions="Search for information.",
    tools=[search_tool],
    model_settings=ModelSettings(tool_choice="required"),
)

# Phase 2: Must analyze
analyzer = Agent(
    name="Analyzer",
    instructions="Analyze the search results.",
    tools=[analyze_tool],
    model_settings=ModelSettings(tool_choice="required"),
)

# Phase 3: Free response
writer = Agent(
    name="Writer",
    instructions="Write based on analysis.",
    # No tool_choice = free text response
)

async def deterministic_workflow(query: str):
    # Guaranteed order: search -> analyze -> write
    search_result = await Runner.run(search_agent, query)
    analysis_result = await Runner.run(analyzer, search_result.to_input_list())
    final_result = await Runner.run(writer, analysis_result.to_input_list())
    return final_result.final_output

Sessions

Conversation History with to_input_list()

Manual conversation history management:

from agents import Agent, Runner, TResponseInputItem

agent = Agent(name="ChatBot", instructions="Be helpful.")

# First message
result = await Runner.run(agent, "Hello!")

# Continue conversation with history
inputs = result.to_input_list()
inputs.append({"role": "user", "content": "Tell me more"})

result = await Runner.run(agent, inputs)

SQLite Session

Automatic conversation history with SQLite:

from agents import Agent, Runner
from agents.extensions.sessions import SQLiteSession

agent = Agent(name="ChatBot", instructions="Remember our conversation.")

# Session stores and loads history automatically
session = SQLiteSession("conversation_123")

result1 = await Runner.run(agent, "My name is John", session=session)
result2 = await Runner.run(agent, "What's my name?", session=session)
# -> "Your name is John"

Advanced SQLite Session

from agents import Agent, Runner
from agents.extensions.sessions import SQLiteSession

# Custom database path
session = SQLiteSession(
    session_id="user_456_chat",
    db_path="./data/conversations.db",
)

agent = Agent(
    name="MemoryBot",
    instructions="Remember user preferences and history.",
)

# Multiple conversations with same agent
await Runner.run(agent, "I prefer dark mode", session=session)
await Runner.run(agent, "Set language to Finnish", session=session)

# Later session retrieval
session2 = SQLiteSession(session_id="user_456_chat", db_path="./data/conversations.db")
result = await Runner.run(agent, "What are my preferences?", session=session2)
# -> Remembers dark mode and Finnish language

Redis Session

For distributed systems:

from agents import Agent, Runner
from agents.extensions.sessions import RedisSession

session = RedisSession(
    session_id="user_789",
    redis_url="redis://localhost:6379",
    ttl=3600,  # 1 hour expiry
)

agent = Agent(name="ScalableBot", instructions="Be helpful.")

result = await Runner.run(agent, "Hello!", session=session)

OpenAI Session

Using OpenAI's built-in memory:

from agents import Agent, Runner
from agents.extensions.sessions import OpenAISession

session = OpenAISession(session_id="openai_session_123")

agent = Agent(
    name="OpenAIMemoryBot",
    instructions="Use your memory to help users.",
)

result = await Runner.run(agent, "Remember I like Python", session=session)

Compaction Session

Automatically summarize long conversations:

from agents import Agent, Runner
from agents.extensions.sessions import CompactionSession, SQLiteSession

base_session = SQLiteSession("long_conversation")

# Compacts history when it exceeds threshold
session = CompactionSession(
    base_session=base_session,
    max_messages=20,  # Compact after 20 messages
    summary_model="gpt-5.4-mini",  # Model for summarization
)

agent = Agent(name="LongChatBot", instructions="Have long conversations.")

# After many messages, older ones are summarized
for i in range(30):
    await Runner.run(agent, f"Message {i}", session=session)

Encrypted Session

For sensitive conversations:

from agents import Agent, Runner
from agents.extensions.sessions import EncryptedSession, SQLiteSession

base_session = SQLiteSession("sensitive_chat")

session = EncryptedSession(
    base_session=base_session,
    encryption_key="your-32-byte-encryption-key-here",
)

agent = Agent(name="SecureBot", instructions="Handle sensitive information.")

result = await Runner.run(agent, "My SSN is 123-45-6789", session=session)
# Data stored encrypted in SQLite

Session Comparison

Streaming

Basic Streaming

from openai.types.responses import ResponseTextDeltaEvent
from agents import Agent, Runner

agent = Agent(name="Writer", instructions="Write stories.")

result = Runner.run_streamed(agent, input="Write a short story")

async for event in result.stream_events():
    if event.type == "raw_response_event":
        if isinstance(event.data, ResponseTextDeltaEvent):
            print(event.data.delta, end="", flush=True)

Stream Items

from agents import Agent, Runner

agent = Agent(name="Assistant", instructions="Be helpful.")

result = Runner.run_streamed(agent, input="Tell me about Python")

async for item in result.stream_items():
    print(f"Item type: {item.type}")
    if hasattr(item, "text"):
        print(f"Text: {item.text}")

SSE Streaming with FastAPI

import json
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai.types.responses import ResponseTextDeltaEvent
from agents import Agent, Runner

app = FastAPI()

agent = Agent(name="Assistant", instructions="Be helpful.")

def sse(event: str, data: dict) -> str:
    return f"event: {event}\ndata: {json.dumps(data)}\n\n"

@app.post("/stream")
async def stream_response(prompt: str):
    async def generate():
        result = Runner.run_streamed(agent, input=prompt)

        async for event in result.stream_events():
            if event.type == "raw_response_event":
                if isinstance(event.data, ResponseTextDeltaEvent):
                    yield sse("delta", {"text": event.data.delta})

        yield sse("done", {})

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
    )

Streaming with Tool Calls

from agents import Agent, Runner, function_tool
from openai.types.responses import ResponseTextDeltaEvent, ResponseFunctionCallArgumentsDeltaEvent

@function_tool
def get_data(query: str) -> str:
    return f"Data for {query}"

agent = Agent(
    name="DataBot",
    instructions="Fetch data when asked.",
    tools=[get_data],
)

result = Runner.run_streamed(agent, input="Get data about sales")

async for event in result.stream_events():
    if event.type == "raw_response_event":
        if isinstance(event.data, ResponseTextDeltaEvent):
            print(f"Text: {event.data.delta}", end="")
        elif isinstance(event.data, ResponseFunctionCallArgumentsDeltaEvent):
            print(f"Tool args: {event.data.delta}", end="")

Streaming with Guardrails

from agents import Agent, Runner, input_guardrail
from agents import GuardrailFunctionOutput, RunContextWrapper

@input_guardrail
async def check_input(
    ctx: RunContextWrapper, agent: Agent, input: str
) -> GuardrailFunctionOutput:
    if "bad" in input.lower():
        return GuardrailFunctionOutput(
            tripwire_triggered=True,
            output_info="Inappropriate content",
        )
    return GuardrailFunctionOutput(tripwire_triggered=False)

agent = Agent(
    name="SafeBot",
    instructions="Be helpful.",
    input_guardrails=[check_input],
)

try:
    result = Runner.run_streamed(agent, input="Hello")
    async for event in result.stream_events():
        # Process events
        pass
except Exception as e:
    print(f"Guardrail triggered: {e}")

Collecting Full Response

result = Runner.run_streamed(agent, input="Tell me a story")

# Stream first
async for event in result.stream_events():
    if event.type == "raw_response_event":
        if isinstance(event.data, ResponseTextDeltaEvent):
            print(event.data.delta, end="")

# Then get full result
final_result = await result.final_result()
print(f"\n\nFull output: {final_result.final_output}")

Structured Output

AgentOutputSchema with Pydantic

from pydantic import BaseModel, Field
from agents import Agent, Runner, AgentOutputSchema, ModelSettings
from openai.types.shared.reasoning import Reasoning

# Pydantic model for response structure
class ProductRecommendationLite(BaseModel):
    product_id: str = Field(description="Unique product ID")
    name: str = Field(description="Product name")
    relevance_reason: str = Field(description="Why this product matches")
    match_score: float = Field(ge=0, le=1, description="Match score 0-1")

class ProductSelectionOutput(BaseModel):
    products: list[ProductRecommendationLite] = Field(description="Selected products")

# Agent with strict JSON schema output
agent = Agent(
    name="ProductSelector",
    instructions="Select the 10 best products matching user request...",
    model=get_model(),
    model_settings=ModelSettings(
        max_tokens=64000,
        # Reasoning effort: "none", "low", "medium", "high"
        reasoning=Reasoning(effort="low"),
    ),
    # strict_json_schema=True forces LLM to return valid JSON
    output_type=AgentOutputSchema(ProductSelectionOutput, strict_json_schema=True),
)

result = await Runner.run(agent, "Find products for family hiking trip")
output: ProductSelectionOutput = result.final_output

# Use the result
for product in output.products:
    print(f"{product.name}: {product.score} - {product.relevance_reason}")

Simple Output Type

from dataclasses import dataclass
from typing import Literal

@dataclass
class EvaluationFeedback:
    feedback: str
    score: Literal["pass", "needs_improvement", "fail"]

evaluator = Agent[None](
    name="Evaluator",
    instructions="Evaluate content and provide feedback.",
    output_type=EvaluationFeedback,
)

result = await Runner.run(evaluator, "Review this story outline...")
evaluation: EvaluationFeedback = result.final_output
print(f"Score: {evaluation.score}, Feedback: {evaluation.feedback}")

ModelSettings

from agents import Agent, ModelSettings
from openai.types.shared.reasoning import Reasoning

agent = Agent(
    name="Assistant",
    instructions="Be helpful.",
    model="gpt-5.4",
    model_settings=ModelSettings(
        max_tokens=32000,
        temperature=0.7,
        tool_choice="required",  # Force tool usage
        reasoning=Reasoning(effort="medium"),  # GPT-5 reasoning
    ),
)

ModelSettings Options

Non-Strict Output

For schemas that don't support strict mode:

from agents import Agent, AgentOutputSchema

class FlexibleOutput(BaseModel):
    data: dict  # dict type not supported in strict mode
    notes: str

agent = Agent(
    name="Flexible",
    instructions="Return flexible data.",
    output_type=AgentOutputSchema(FlexibleOutput, strict_json_schema=False),
)

Tools

Function Tools (@function_tool)

from typing import Annotated
from agents import Agent, Runner, function_tool

@function_tool
def get_weather(city: Annotated[str, "City name"]) -> str:
    """Get weather for a city."""
    return f"Weather in {city}: Sunny, 20C"

@function_tool
async def search_database(query: Annotated[str, "Search query"]) -> list[dict]:
    """Search products in database."""
    # Async function - can await database calls
    return [{"id": "1", "name": "Hiking boots"}]

agent = Agent(
    name="Assistant",
    instructions="Help users find information.",
    tools=[get_weather, search_database],
)

Tool with Multiple Parameters

@function_tool
def book_flight(
    origin: Annotated[str, "Departure city"],
    destination: Annotated[str, "Arrival city"],
    date: Annotated[str, "Travel date (YYYY-MM-DD)"],
    passengers: Annotated[int, "Number of passengers"] = 1,
) -> dict:
    """Book a flight between two cities."""
    return {
        "confirmation": "ABC123",
        "route": f"{origin} -> {destination}",
        "date": date,
        "passengers": passengers,
    }

Hosted Tools (Built-in)

from agents import Agent, WebSearchTool, CodeInterpreterTool

agent = Agent(
    name="Researcher",
    instructions="Search the web and analyze data.",
    tools=[
        WebSearchTool(user_location={"type": "approximate", "city": "Helsinki"}),
        CodeInterpreterTool(),
    ],
)

Agents as Tools

Use other agents as tools for orchestration:

from agents import Agent, Runner

translator_es = Agent(
    name="SpanishTranslator",
    instructions="Translate to Spanish.",
)

translator_fr = Agent(
    name="FrenchTranslator",
    instructions="Translate to French.",
)

orchestrator = Agent(
    name="Orchestrator",
    instructions="Use translation tools as needed.",
    tools=[
        translator_es.as_tool(
            tool_name="translate_spanish",
            tool_description="Translate text to Spanish",
        ),
        translator_fr.as_tool(
            tool_name="translate_french",
            tool_description="Translate text to French",
        ),
    ],
)

result = await Runner.run(orchestrator, "Translate 'hello' to Spanish and French")

Tool Guardrails

from agents import Agent, function_tool, tool_guardrail
from agents import ToolGuardrailFunctionOutput, RunContextWrapper

@tool_guardrail
async def validate_query(
    ctx: RunContextWrapper, agent: Agent, tool_input: dict
) -> ToolGuardrailFunctionOutput:
    query = tool_input.get("query", "")
    if len(query) < 3:
        return ToolGuardrailFunctionOutput(
            tripwire_triggered=True,
            output_info="Query too short",
        )
    return ToolGuardrailFunctionOutput(tripwire_triggered=False)

@function_tool(guardrails=[validate_query])
def search(query: Annotated[str, "Search query"]) -> list[str]:
    """Search for items."""
    return ["result1", "result2"]

Forcing Tool Use

from agents import Agent, ModelSettings

agent = Agent(
    name="ToolUser",
    instructions="Always use tools to answer.",
    tools=[get_weather, search_database],
    model_settings=ModelSettings(
        tool_choice="required",  # Force tool usage
    ),
)