Openai Agents

OpenAI Agents SDK Patterns

The OpenAI Agents SDK (openai-agents) orchestrates LLM agents with tools, handoffs, and tracing.

When to Activate

• Defining agents with system prompts, tools, and handoffs
• Writing @function_tool decorators and tool schemas
• Running agents with Runner.run() or streaming with Runner.run_streamed()
• Implementing multi-agent handoffs (triage → specialist)
• Debugging tool call errors, context leaks, or infinite loops
• Integrating with Agentex ADK via adk.providers.openai
• Adding tracing spans for observability

Core Concepts

Agent
├── name, instructions (system prompt)
├── tools     — functions the agent can call
├── handoffs  — other agents it can delegate to
├── model     — LLM to use (default: gpt-4o)
└── output_type — structured Pydantic output (optional)

Runner
├── .run()          — async, returns final output
├── .run_streamed() — async generator, streams events
└── .run_sync()     — sync wrapper (testing/scripts)

Minimal Agent

from agents import Agent, Runner, function_tool

@function_tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    return f"It's sunny and 72°F in {city}."

agent = Agent(
    name="Weather Agent",
    instructions="You help users check weather. Always use the get_weather tool.",
    tools=[get_weather],
    model="gpt-4o-mini",
)

# Run
result = await Runner.run(agent, "What's the weather in Tokyo?")
print(result.final_output)

Defining Tools

from agents import function_tool
from pydantic import BaseModel

# Simple tool — docstring becomes the tool description
@function_tool
def search_web(query: str) -> str:
    """Search the web for current information. Returns the top results."""
    return web_search_api(query)

# Tool with multiple typed params
# NOTE: `eval()` on a tool argument is RCE — the model can pass `__import__(...)`.
# Use a constrained evaluator like `simpleeval` (or `ast.literal_eval` for literals).
from simpleeval import simple_eval

@function_tool
def calculate(expression: str, precision: int = 2) -> str:
    """Evaluate a mathematical expression and return the result."""
    result = simple_eval(expression)
    return str(round(result, precision))

# Tool returning structured data
class SearchResult(BaseModel):
    title: str
    url: str
    snippet: str

@function_tool
def search_docs(query: str, limit: int = 5) -> list[SearchResult]:
    """Search the documentation. Returns matching articles."""
    return [SearchResult(...) for r in docs_search(query, limit)]

# Async tool
@function_tool
async def fetch_user(user_id: str) -> dict:
    """Fetch user profile from the database."""
    user = await db.get_user(user_id)
    return user.model_dump()

Tool naming: the function name becomes the tool name. Keep names short and action-oriented (search_web, not search_the_web_for_information).

Context (passing data to tools without LLM)

from agents import Agent, Runner, RunContextWrapper, function_tool
from dataclasses import dataclass

@dataclass
class AppContext:
    user_id: str
    db_session: AsyncSession

# Tools receive context as first param (not exposed to LLM)
@function_tool
async def get_my_orders(ctx: RunContextWrapper[AppContext]) -> list[dict]:
    """Get the current user's orders."""
    orders = await OrderCRUD(ctx.context.db_session).list_for_user(ctx.context.user_id)
    return [o.model_dump() for o in orders]

agent = Agent[AppContext](
    name="Order Agent",
    instructions="Help users check their orders.",
    tools=[get_my_orders],
)

context = AppContext(user_id="u-123", db_session=session)
result = await Runner.run(agent, "Show my recent orders", context=context)

Structured Output

from pydantic import BaseModel
from agents import Agent, Runner

class EmailDraft(BaseModel):
    subject: str
    body: str
    tone: Literal["formal", "casual", "urgent"]

agent = Agent(
    name="Email Writer",
    instructions="Draft professional emails based on user requests.",
    output_type=EmailDraft,   # forces structured JSON response
)

result = await Runner.run(agent, "Write a follow-up email for a job interview")
email: EmailDraft = result.final_output   # typed, validated by Pydantic
print(email.subject)

Handoffs (Multi-Agent)

Handoffs let one agent delegate to another specialized agent. The triage agent decides which specialist handles the task.

from agents import Agent, handoff, Runner

coding_agent = Agent(
    name="Coding Assistant",
    instructions="You solve programming problems. Write clean, working code.",
    tools=[search_docs, run_code],
)

writing_agent = Agent(
    name="Writing Assistant",
    instructions="You help with writing, editing, and proofreading.",
)

triage_agent = Agent(
    name="Triage",
    instructions="""Route the user to the right specialist:
    - For code/programming questions → coding_assistant
    - For writing/editing requests → writing_assistant
    - Handle simple questions yourself.""",
    handoffs=[coding_agent, writing_agent],
)

result = await Runner.run(triage_agent, "Fix this Python bug: ...")
# triage_agent may hand off to coding_agent, which runs to completion
print(result.final_output)

Customizing handoff behavior:

from agents import handoff

def on_handoff_to_billing(ctx: RunContextWrapper[AppContext]):
    # Called when handoff happens — log, update state, etc.
    logger.info(f"Handing off to billing for user {ctx.context.user_id}")

billing_agent = Agent(name="Billing", instructions="...")

triage_agent = Agent(
    handoffs=[
        handoff(billing_agent,
    ]
)

Streaming

from agents import Runner
from agents.stream_events import RunItemStreamEvent, AgentUpdatedStreamEvent

async def stream_agent(agent, prompt: str):
    stream = Runner.run_streamed(agent, prompt)
    async for event in stream.stream_events():
        if isinstance(event, RunItemStreamEvent):
            # Each completed item (tool call, tool result, message)
            item = event.item
            if hasattr(item, "raw_item"):
                raw = item.raw_item
                if raw.get("type") == "response.output_text.delta":
                    print(raw["delta"], end="", flush=True)

    return await stream.get_final_output()

# FastAPI SSE endpoint
@router.get("/stream")
async def stream_response(prompt: str):
    async def generate():
        stream = Runner.run_streamed(agent, prompt)
        async for event in stream.stream_events():
            if isinstance(event, RunItemStreamEvent):
                item = event.item
                if hasattr(item, "raw_item"):
                    delta = item.raw_item.get("delta", "")
                    if delta:
                        yield f"data: {delta}\n\n"
    return StreamingResponse(generate(), media_type="text/event-stream")

Guardrails

Guardrails validate input/output before the agent processes or responds.

from agents import Agent, input_guardrail, output_guardrail, GuardrailFunctionOutput

@input_guardrail
async def no_pii(ctx, agent, input) -> GuardrailFunctionOutput:
    text = input if isinstance(input, str) else str(input)
    if contains_pii(text):
        return GuardrailFunctionOutput(
            output_info="PII detected",
            tripwire_triggered=True,    # stops the agent
        )
    return GuardrailFunctionOutput(output_info="clean", tripwire_triggered=False)

@output_guardrail
async def no_harmful_content(ctx, agent, output) -> GuardrailFunctionOutput:
    if is_harmful(str(output)):
        return GuardrailFunctionOutput(
            output_info="harmful content",
            tripwire_triggered=True,
        )
    return GuardrailFunctionOutput(output_info="safe", tripwire_triggered=False)

agent = Agent(
    name="Safe Agent",
    instructions="...",
    input_guardrails=[no_pii],
    output_guardrails=[no_harmful_content],
)

Tracing

from agents import Agent, Runner
from agents.tracing import trace, custom_span

# Runner automatically creates a root trace
result = await Runner.run(agent, "Hello", run_config=RunConfig(
    trace_id="my-trace-123",        # link to your own tracing system
    trace_metadata={"user_id": "u-123"},
))

# Add custom spans inside tools
@function_tool
async def complex_search(query: str) -> str:
    """Search across multiple sources."""
    with custom_span("db_search"):
        db_results = await db.search(query)
    with custom_span("web_search"):
        web_results = await web.search(query)
    return combine(db_results, web_results)

Agentex ADK Integration

In Agentex Temporal agents, use adk.providers.openai instead of calling Runner directly — it handles message streaming to the UI automatically.

from agentex.lib import adk
from agents import Agent, function_tool

@function_tool
async def search_web(query: str) -> str:
    """Search the web for information."""
    return await web_search(query)

agent = Agent(
    name="Research Agent",
    instructions="Research topics thoroughly using web search.",
    tools=[search_web],
    model="gpt-4o",
)

# In a Temporal activity:
async def run_research_agent(params: AgentParams) -> str:
    result = await adk.providers.openai.run_agent_streamed_auto_send(
        agent=agent,
        task_id=params.task_id,
        input=params.user_message,
        # context=AppContext(...)  if using typed context
    )
    return result.final_output

# run_agent_streamed_auto_send:
# - streams each token to the Agentex UI via adk.messages
# - wraps Runner.run_streamed internally
# - handles tracing integration

RunConfig Options

from agents import RunConfig

result = await Runner.run(
    agent,
    "Hello",
    run_config=RunConfig(
        model="gpt-4o",                    # override agent's model
        model_settings=ModelSettings(
            temperature=0.2,
            max_tokens=2000,
        ),
        max_turns=10,                       # prevent infinite agent loops
        trace_id="req-abc-123",
        workflow_name="my-workflow",        # appears in traces
    ),
)

Common Errors

Red Flags

• No max_turns set — without a turn limit an agent that calls a tool whose result triggers another tool call can loop indefinitely; always pass RunConfig(max_turns=N) to cap runaway execution
• Passing app state (DB session, user ID) through the LLM — including session objects or sensitive IDs in the prompt or tool return values exposes them to the model and wastes tokens; use typed context (RunContextWrapper) so tools receive state without the LLM ever seeing it
• Vague tool docstrings — the docstring is the only description the LLM sees; "Does stuff with the database" gives the model no signal on when to call it; write one sentence that says exactly what the tool returns and when to use it
• Vague handoff instructions in the triage agent — "Route to the right agent" with no criteria leads to random or wrong handoffs; list the exact conditions for each handoff in the triage agent's instructions
• Using Runner.run() directly inside an Agentex activity — Runner.run() doesn't stream tokens to the Agentex UI; use adk.providers.openai.run_agent_streamed_auto_send() which wraps Runner.run_streamed() and handles token delivery automatically
• No input/output guardrails on user-facing agents — agents that handle user-supplied text without guardrails can be prompted to leak context, call wrong tools, or produce harmful output; add @input_guardrail and @output_guardrail for sensitive deployments
• Tool names that are long or vague — the model uses the tool name as a primary signal; search_the_web_for_current_information is worse than search_web; keep tool names short, lowercase, and verb-noun

Checklist

• Tool docstrings are clear — they become the LLM-facing description
• Context used for app state (DB session, user ID) — not passed through LLM
• output_type set for structured outputs — avoids parsing LLM text
• max_turns set to prevent runaway agent loops
• Handoff instructions are specific — vague routing leads to wrong agent
• run_agent_streamed_auto_send used in Agentex activities (not Runner directly)
• Guardrails added for user-facing agents handling sensitive data
• Tool names are short verb-noun: get_user, search_docs, send_email