OpenAI Agents SDK

The mental model

An agent is a model plus instructions plus tools. The runner loops it.

The whole SDK rests on three nouns and one verb. An Agent is an LLM configured with a name, instructions, and a set of tools. A tool is a Python function the model can ask to run. The Runner takes your input and runs the agent loop: call the model, and if the model asks for a tool, run it and feed the result back; if the model hands off to another agent, switch to it; when the model returns a plain answer, that is the final output and the loop stops.

That is it. There is no graph to wire, no state schema to declare. Compared to LangGraph (an explicit state graph you own) this SDK hides the loop and hands you the pieces you actually assemble around it: handoffs, guardrails, sessions, streaming, and tracing. It reached production maturity through 2026, so it is a credible default rather than an experiment.

flowchart LR
  IN(["input"]) --> M["call model"]
  M -->|"tool call"| T["run tool, append result"]
  T --> M
  M -->|"handoff"| H["switch agent"]
  H --> M
  M -->|"final output"| OUT(["result.final_output"])
  classDef model fill:#EEE6FF,stroke:#6D28D9,color:#09244B;
  classDef code fill:#D6F5E3,stroke:#1B7F46,color:#09244B;
  class M,H model;
  class T,IN,OUT code;
    

Install & setup

One package, one API key.

The SDK is Python-first. A separate, actively maintained TypeScript/JavaScript port ships as @openai/agents with the same concepts (Agent, tools, run, handoffs, guardrails). We use Python throughout this course.

# the SDK. the import name is `agents`, the package is `openai-agents`
pip install openai-agents

# the SDK reads this automatically. no client object to construct.
export OPENAI_API_KEY="sk-..."

There is no client to instantiate and no key to pass around: the SDK picks up OPENAI_API_KEY from the environment. To point at a different model provider, you set a client explicitly, but the default path is zero-config.

Hello world

An agent and a run, in five lines.

The smallest useful program: define an Agent with a name and instructions, then hand it to Runner.run_sync with your input. The answer is on result.final_output.

from agents import Agent, Runner

agent = Agent(
    name="Assistant",
    instructions="You are a concise, helpful assistant.",
)

result = Runner.run_sync(agent, "Write a haiku about recursion in programming.")
print(result.final_output)

run_sync is the blocking convenience wrapper; run is the async one you will use in real services

The core · the Agent

The Agent is a config object. Every knob lives here.

An Agent holds everything about one participant in the loop. The parameters you will reach for most:

Structured output with output_type

Set output_type to a Pydantic model and the SDK asks the model for structured output, then parses it. result.final_output is now a typed instance instead of a string. This is the same discipline you learn with Pydantic AI, available inline here.

from agents import Agent, Runner, ModelSettings
from pydantic import BaseModel

class CalendarEvent(BaseModel):
    name: str
    date: str
    participants: list[str]

extractor = Agent(
    name="Event extractor",
    instructions="Extract the event details from the user text.",
    model="gpt-4o",
    model_settings=ModelSettings(temperature=0),
    output_type=CalendarEvent,   # final_output is now a CalendarEvent, not a str
)

result = Runner.run_sync(extractor, "Standup with Sam and Priya on 2026-07-20")
event = result.final_output          # typed: event.name, event.date, event.participants
print(event.participants)

Instructions can also be a callable, which is how you inject per-request context (a user name, a plan tier) without rebuilding the agent: instructions=lambda ctx, agent: f"Help {ctx.context.user_name}."

Function tools

Decorate a Python function; it becomes a tool.

The @function_tool decorator turns any typed function into a tool. The SDK builds the JSON schema from the type hints and reads the tool description straight from the docstring (parsed with griffe, so Google, Sphinx, and NumPy docstring styles all work). Argument descriptions come from the Args: section.

from agents import Agent, Runner, function_tool

@function_tool
def get_weather(city: str) -> str:
    """Get the current weather for a city.

    Args:
        city: The city to look up, for example London.
    """
    return f"It is 22C and sunny in {city}."

agent = Agent(
    name="Weather bot",
    instructions="Answer weather questions using the tool.",
    tools=[get_weather],
)

print(Runner.run_sync(agent, "What is the weather in Oslo?").final_output)

Reading run context inside a tool

Add a first parameter typed RunContextWrapper and the SDK injects the run context (any object you pass to Runner.run(..., context=...)). This is how a tool sees the current user, a DB handle, or an auth token, none of which the model should ever supply. The model never sees this argument; it is stripped from the schema.

from dataclasses import dataclass
from agents import function_tool, RunContextWrapper

@dataclass
class AppContext:
    user_id: str

@function_tool
def list_orders(ctx: RunContextWrapper[AppContext]) -> list[dict]:
    """List the current user's recent orders."""
    return db_orders_for(ctx.context.user_id)   # user_id from code, never from the model

OpenAI also ships hosted tools that run on their infrastructure, not yours, and only with OpenAI Responses models: WebSearchTool(), FileSearchTool(vector_store_ids=[...]), and CodeInterpreterTool(). Drop them straight into the tools list alongside your own functions.

The Runner & the agent loop

Three ways to run, one loop underneath.

Every run goes through the Runner. The three entry points differ only in how you await them and how you read output:

Inside any of them the runner executes the agent loop: call the model, run any tool calls the model requested and append the results, follow a handoff if the model asked for one, and stop when the model returns a final output. One call to run is one application turn, however many model calls and tool calls that took.

import asyncio
from agents import Runner

async def main():
    result = await Runner.run(agent, "What is the weather in Oslo?", max_turns=8)
    print(result.final_output)
    # result also carries: .new_items (tool calls, messages), .last_agent,
    # and .to_input_list() to feed this run's history into the next one

asyncio.run(main())

The max_turns argument caps how many times the loop may go around before the SDK raises MaxTurnsExceeded. It is your circuit breaker against a model that keeps calling tools forever. To carry a conversation forward manually, pass result.to_input_list() + [new_message] as the next input; or let a session do it for you (next section but one).

Handoffs

Delegation is just a tool the model can call.

A handoff lets one agent pass the conversation to another. Under the hood each handoff becomes a tool named transfer_to_<agent_name>; when the model calls it, the runner switches the active agent and keeps looping. This is the SDK's answer to multi-agent systems: a triage agent that routes to specialists, each with its own instructions and tools.

flowchart TB
  U(["user message"]) --> TRI{"triage agent"}
  TRI -->|"transfer_to_billing"| B["billing agent"]
  TRI -->|"transfer_to_refunds"| R["refunds agent"]
  TRI -->|"answers directly"| A(["reply"])
  B --> A
  R --> A
  classDef model fill:#EEE6FF,stroke:#6D28D9,color:#09244B;
  classDef code fill:#D6F5E3,stroke:#1B7F46,color:#09244B;
  class TRI,B,R model;
    

Pass agents directly in the handoffs list, or wrap one in handoff(...) to customize it. The handoff_description on the target agent tells the triage model when to route there.

from agents import Agent, handoff

billing = Agent(
    name="Billing",
    handoff_description="Handles invoices, charges, and payment questions.",
    instructions="You resolve billing issues.",
)
refunds = Agent(
    name="Refunds",
    handoff_description="Handles refund requests and status.",
    instructions="You process refunds.",
)

triage = Agent(
    name="Triage",
    instructions="Route the user to the right specialist. Do not answer billing yourself.",
    handoffs=[billing, handoff(refunds)],
)

Running code on handoff

Wrap the target in handoff(...) to attach an on_handoff callback and an input_type, so the model must supply structured data as it delegates (a reason, a priority). Use this to log the escalation, warm a cache, or record why the route happened.

from pydantic import BaseModel
from agents import handoff, RunContextWrapper

class EscalationData(BaseModel):
    reason: str

async def on_escalate(ctx: RunContextWrapper[None], data: EscalationData):
    log(f"Escalated: {data.reason}")     # fires the moment the handoff is invoked

escalation_agent = Agent(name="Escalation", instructions="Handle escalations.")
to_escalation = handoff(escalation_agent, on_handoff=on_escalate, input_type=EscalationData)

Guardrails

Cheap checks that can trip a wire and stop the run.

Guardrails run alongside your agent to validate what goes in and what comes out. An input guardrail runs before the agent on the user's message; an output guardrail runs after, on the final output. Each returns a GuardrailFunctionOutput; if tripwire_triggered is true, the SDK raises immediately (InputGuardrailTripwireTriggered / OutputGuardrailTripwireTriggered) and the run stops. The idiom is to use a small, fast model to judge, so the expensive agent never runs on bad input.

from pydantic import BaseModel
from agents import (
    Agent, Runner, GuardrailFunctionOutput, RunContextWrapper,
    TResponseInputItem, input_guardrail, InputGuardrailTripwireTriggered,
)

class Check(BaseModel):
    is_allowed: bool
    reasoning: str

# a small, cheap agent that only judges the input
checker = Agent(
    name="Policy check",
    instructions="Return is_allowed=false if the user asks for medical or legal advice.",
    model="gpt-4o-mini",
    output_type=Check,
)

@input_guardrail
async def policy_guardrail(
    ctx: RunContextWrapper[None],
    agent: Agent,
    user_input: str | list[TResponseInputItem],
) -> GuardrailFunctionOutput:
    result = await Runner.run(checker, user_input, context=ctx.context)
    return GuardrailFunctionOutput(
        output_info=result.final_output,
        tripwire_triggered=not result.final_output.is_allowed,
    )

agent = Agent(
    name="Assistant",
    instructions="Help with general questions.",
    input_guardrails=[policy_guardrail],
)

try:
    await Runner.run(agent, "Diagnose my symptoms.")
except InputGuardrailTripwireTriggered:
    print("Blocked before the main model ever ran.")

Output guardrails work the same way with the @output_guardrail decorator; their function receives the agent's output instead of the user input. Reach for them to catch leaked PII, off-brand tone, or a structured answer that fails a business rule before it reaches the user.

Sessions & memory

Pass a session and history is handled for you.

By default each Runner.run is stateless: the agent forgets the previous turn. A Session fixes that. Create a SQLiteSession with a conversation id, pass it on every run, and the SDK loads the prior history before the run and saves the new items after, so follow-up questions just work. Give it a file path and the history survives restarts; omit the path for an in-memory session.

from agents import Agent, Runner, SQLiteSession

agent = Agent(name="Assistant", instructions="Reply concisely.")

# id "user-42", persisted to conversations.db (omit the path for in-memory)
session = SQLiteSession("user-42", "conversations.db")

await Runner.run(agent, "What city is the Golden Gate Bridge in?", session=session)
# -> San Francisco

result = await Runner.run(agent, "And what state is that in?", session=session)
print(result.final_output)   # -> California. The bridge context was reloaded from the session.

A session also exposes direct operations for correcting history: await session.get_items(), add_items(...), pop_item() (handy for undo before a retry), and clear_session(). SQLiteSession is the built-in default; the SDK also ships other session backends, and you can implement the session protocol against your own store.

Streaming

Subscribe to events as the run unfolds.

Call Runner.run_streamed(...), which returns immediately with a RunResultStreaming, then iterate result.stream_events(). You get three kinds of event: raw_response_event for token-level deltas (the typing effect), run_item_stream_event when a whole item completes (a tool call, a message), and agent_updated_stream_event when a handoff changes the active agent.

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

async def main():
    agent = Agent(name="Joker", instructions="Tell short jokes.")
    result = Runner.run_streamed(agent, input="Tell me two jokes.")

    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)   # token by token
        elif event.type == "agent_updated_stream_event":
            print(f"\n[now: {event.new_agent.name}]")         # a handoff happened
        elif event.type == "run_item_stream_event":
            if event.item.type == "tool_call_item":
                print("\n[calling a tool...]")

asyncio.run(main())

the run is not finished until stream_events stops yielding, keep consuming to the end

Tracing

Every run is traced, out of the box.

Tracing is on by default. The SDK records each run as a trace of spans covering model generations, tool calls, handoffs, and guardrails, and ships them to the OpenAI Traces dashboard at platform.openai.com/traces. You get a step-by-step timeline for free, which is the single biggest reason to reach for a vendor SDK over a hand-rolled loop.

To group several runs into one logical workflow (a multi-turn conversation, a batch job), wrap them in a trace(...) context manager so they share one trace instead of appearing as separate ones.

from agents import Agent, Runner, trace

async def main():
    agent = Agent(name="Assistant", instructions="Be helpful.")

    # both runs land under a single "Support session" trace
    with trace("Support session"):
        r1 = await Runner.run(agent, "What is my order status?")
        r2 = await Runner.run(agent, "Can you cancel it?")

Disable it for a run with RunConfig(tracing_disabled=True), globally with set_tracing_disabled(True), or via the OPENAI_AGENTS_DISABLE_TRACING=1 environment variable. You can also register custom trace processors to fan traces out to a third-party observability backend.

Putting it together

A support system: guardrail, triage, two specialists.

This is the shape you will actually ship. An input guardrail blocks off-topic requests cheaply. A triage agent routes to a billing specialist or an orders specialist. The orders specialist has a real tool that reads run context for the user id. Everything is grouped under one trace and remembered across turns with a session. One fuzzy step, code on both sides of it.

import asyncio
from dataclasses import dataclass
from pydantic import BaseModel
from agents import (
    Agent, Runner, SQLiteSession, function_tool, trace,
    RunContextWrapper, TResponseInputItem,
    GuardrailFunctionOutput, input_guardrail, InputGuardrailTripwireTriggered,
)

@dataclass
class Ctx:
    user_id: str

# --- a real tool, reads user_id from code not from the model ---
@function_tool
def lookup_order(ctx: RunContextWrapper[Ctx], order_id: str) -> dict:
    """Look up an order's status by its ID for the current user."""
    return {"id": order_id, "user": ctx.context.user_id, "status": "delayed"}

# --- input guardrail: a cheap model gates the expensive one ---
class Topic(BaseModel):
    on_topic: bool

topic_check = Agent(
    name="Topic check",
    instructions="Set on_topic=false unless the message is about billing or orders.",
    model="gpt-4o-mini",
    output_type=Topic,
)

@input_guardrail
async def on_topic_only(
    ctx: RunContextWrapper[Ctx], agent: Agent,
    user_input: str | list[TResponseInputItem],
) -> GuardrailFunctionOutput:
    r = await Runner.run(topic_check, user_input, context=ctx.context)
    return GuardrailFunctionOutput(
        output_info=r.final_output,
        tripwire_triggered=not r.final_output.on_topic,
    )

# --- specialists ---
billing = Agent(
    name="Billing",
    handoff_description="Invoices, charges, payment questions.",
    instructions="Resolve billing questions clearly and concisely.",
)
orders = Agent(
    name="Orders",
    handoff_description="Order status and shipping questions.",
    instructions="Use lookup_order to answer order questions.",
    tools=[lookup_order],
)

# --- triage routes to a specialist, guardrail runs first ---
triage = Agent(
    name="Triage",
    instructions="Route the user to Billing or Orders. Do not answer yourself.",
    handoffs=[billing, orders],
    input_guardrails=[on_topic_only],
)

async def main():
    ctx = Ctx(user_id="u_42")
    session = SQLiteSession("u_42", "support.db")
    with trace("Support session u_42"):
        try:
            r = await Runner.run(
                triage, "Where is order A-1001?",
                context=ctx, session=session, max_turns=8,
            )
            print(r.final_output)
            print("handled by:", r.last_agent.name)   # -> Orders
        except InputGuardrailTripwireTriggered:
            print("Off topic, refused before the triage model ran.")

asyncio.run(main())

Production

From script to service.

• Use await Runner.run(...) inside your async web handlers; keep run_sync for scripts and tests only.
• Set max_turns on every run so a looping agent raises MaxTurnsExceeded instead of billing you forever.
• Persist conversations with a session keyed by a stable id (user or thread), and trim old items so context stays bounded.
• Put an input guardrail on user-facing agents and an output guardrail on anything that can leak PII or go off-brand.
• Keep tracing on; add a custom trace processor to forward spans to your own observability stack, and wrap multi-run flows in trace(...).
• Do authorization and argument validation inside tool functions, reading identity from RunContextWrapper, never from model-supplied arguments.
• Pin the version. The SDK is pre-1.0 (currently 0.18.x) and still moving; read the changelog before you bump it.

Gotchas

What trips people up.

Reach for it when

• Your stack is OpenAI, as it is throughout this course.
• You are building a single agent, or a few that hand off to each other, not a sprawling stateful graph.
• You want tracing and guardrails for free without adopting a heavy orchestration framework.
• You value a small mental model: agent, tools, runner, and the loop is hidden.

Skip it when

• You need complex stateful branching graphs with durable checkpoints and resume: LangGraph owns that control flow.
• You need provider portability across model vendors: put a gateway in front, or use a provider-neutral framework.
• You need human-in-the-loop pauses that survive a crash for days: that is checkpointed-graph territory, not this SDK's loop.

The principle it teaches

Where it fits your labs

A Week 1 tool that returns in Week 4. The agent patterns you learn early (a tool-using loop, a router, an orchestrator handing to workers) are exactly what this SDK expresses with Agent plus handoffs. Lab 4 has you wrap one of these agents in input and output guardrails for production, which is the guardrails section above, verbatim. Contrast it with LangGraph (heavier, graph-based, durable, for branching flows you must own) and Pydantic AI (when strict typed output is the whole point). This SDK is the lightweight default; you graduate to the others only when a lab outgrows the loop.