OpenAI Agents SDK: Building Production Multi-Agent Systems

The Responses API gives you a single responses.create() call with tool execution baked in. But once your application needs multiple specialized agents, persistent sessions, human-in-the-loop approvals, guardrails, or structured output schemas, you need an orchestration layer. That layer is the OpenAI Agents SDK — the production-ready successor to Swarm, now on PyPI as openai-agents [1].

This guide covers the SDK’s primitives end-to-end: defining agents, registering tools, chaining handoffs, enforcing guardrails, enabling tracing, and choosing between orchestration patterns.

Why the Agents SDK Exists

The SDK sits one layer above the Responses API. The Responses API handles a single turn: input → tool calls → output. The SDK adds:

• Agent loop — Automatically re-invokes the model after tool calls until a final output is produced or a max-turn limit is reached [2]
• Handoffs — Agents delegate tasks to other agents as a native tool type [3]
• Guardrails — Input and output checks that run at every turn, not just at the start [4]
• Tracing — Built-in event logging for debugging and monitoring production agents [5]
• Sessions — Persistent conversation state across multiple Runner.run() calls [6]
• Structured output — Enforce agent output schemas via Pydantic models [7]

If one model call plus tools and application-owned logic is enough, use the Responses API directly. If you need multi-agent orchestration, runtime checks, or production observability, use the SDK.

Step 1: Installing and Configuring the SDK

pip install openai-agents

For Redis-backed session persistence:

pip install 'openai-agents[redis]'

The SDK discovers your OpenAI API key from the OPENAI_API_KEY environment variable, same as the openai Python package. If you use a different backend, set the model provider via set_default_openai_key() or configure openai_api_key per agent.

Your first agent

from agents import Agent, Runner

agent = Agent(
    name="Research Assistant",
    instructions="You are a helpful research assistant. Answer questions accurately and cite sources when possible.",
)

result = Runner.run_sync(agent, "What are the latest advances in multi-agent systems?")
print(result.final_output)

Runner.run_sync() is the synchronous entry point. Under the hood it calls Runner.run() (async). The SDK handles the full agent loop: invoke the model, process tool calls (if any), feed results back, and repeat until the agent produces a final output or reaches the max turn limit [2].

Step 2: Defining Agents with Tools

Tools are Python functions decorated with @function_tool that the agent can invoke. The SDK infers the JSON schema from type hints automatically.

from agents import Agent, Runner, function_tool
import httpx
import json

@function_tool
def search_web(query: str) -> str:
    """Search the web for recent information."""
    # In production, integrate with your search API
    # For example: Tavily, SerpAPI, or a custom RAG pipeline
    return json.dumps([
        {"title": f"Result about {query}", "url": "https://example.com", "snippet": f"Summary of {query}..."}
    ])

@function_tool
def fetch_page(url: str) -> str:
    """Fetch the content of a web page."""
    response = httpx.get(url, timeout=30)
    return response.text[:5000]  # Truncate to avoid token overflow

researcher = Agent(
    name="Researcher",
    instructions="You are a web researcher. Use search_web to find information and get_page to read full articles.",
    tools=[search_web, fetch_page],
)

result = Runner.run_sync(researcher, "Research the latest developments in AI agent frameworks.")
print(result.final_output)

Tool features

• Type hints are schema — No need to write JSON Schema manually. str, int, float, bool, list, dict, and Optional types all map to the correct JSON Schema types [8]
• Docstrings become descriptions — The function docstring is used as the tool description the model sees. Write clear, concise descriptions
• Async tools — Define async def functions and the runner handles them in the event loop
• Custom tool names — Override via @function_tool(name_override="my_custom_name")

Structured output

For agents that must return data in a predictable format, define an output schema:

from pydantic import BaseModel
from agents import Agent, Runner

class ResearchReport(BaseModel):
    title: str
    summary: str
    key_findings: list[str]
    sources: list[str]
    confidence_score: float  # 0.0 to 1.0

agent = Agent(
    name="Research Agent",
    instructions="Research the given topic and produce a structured report.",
    output_type=ResearchReport,
)

result = Runner.run_sync(agent, "Research the state of small language models in 2026.")
report = result.final_output  # typed as ResearchReport

print(f"Title: {report.title}")
print(f"Findings: {len(report.key_findings)} key points")
print(f"Confidence: {report.confidence_score}")

When output_type is set, the agent loops internally until it produces output matching the schema. The SDK uses OpenAI’s structured output mode internally, so JSON mode, function calling, and the schema are all handled without additional configuration [7].

Step 3: Handoffs — Multi-Agent Orchestration

Handoffs are the SDK’s mechanism for agent-to-agent delegation. One agent recognizes that a task is better handled by a specialist and transfers control. Handoffs are surfaced to the model as tools — the agent “calls” a handoff tool, and the runner switches to the target agent.

from agents import Agent, Runner, function_tool

@function_tool
def get_order_status(order_id: str) -> str:
    """Look up the status of a customer order."""
    # In production, query your order management system
    return f"Order {order_id} is out for delivery, expected today."

@function_tool
def initiate_refund(order_id: str, reason: str) -> str:
    """Process a refund for a completed order."""
    return f"Refund initiated for {order_id}. 5-7 business days."

# Specialist agents
order_agent = Agent(
    name="Order Agent",
    instructions="You handle order status and shipping inquiries.",
    tools=[get_order_status],
)

refund_agent = Agent(
    name="Refund Agent",
    instructions="You handle refund and return requests. Only process refunds for completed orders.",
    tools=[get_order_status, initiate_refund],
)

# Triage agent that routes to specialists
triage_agent = Agent(
    name="Triage Agent",
    instructions="""You are a customer support triage agent.
    Route queries to the right specialist based on the customer's needs.
    - Order status / shipping → Order Agent
    - Refunds / returns → Refund Agent
    - Complex questions → escalate to an agent named Support Manager
    """,
    handoffs=[order_agent, refund_agent],  # handoff destinations
)

result = Runner.run_sync(
    triage_agent,
    "I want to check the status of my order #ORD-78923",
)

Handoff mechanics

• Handoffs are tools from the LLM’s perspective — the model decides when to call them
• When a handoff happens, the runner switches to the target agent and continues the loop with that agent’s instructions and tools [3]
• Handoffs are bi-directional — the SDK tracks the agent stack, so a specialist can hand back to the triage agent
• Each handoff can carry a custom handoff_message that the receiving agent sees: handoffs=[order_agent, refund_agent, Agent(name="Support Manager", ...)]

Agent orchestration pattern: handoffs vs. tools

The SDK supports two orchestration patterns. Handoffs are not always the right choice:

Pattern	How it works	Best for
Handoffs	Agents delegate to specialists; runner switches context	When specialists need their own instructions, tools, and state
Agent-as-tool	One orchestrator agent calls specialists as function tools	When the orchestrator must stay in control, track progress, or apply shared rules

Choose handoffs when each specialist needs a different system prompt and tool set, so the model can focus on one domain. Choose agent-as-tool when you need a central orchestrator that decides what each sub-agent does and aggregates results.

# Agent-as-tool pattern: specialist is registered as a function tool
@function_tool
def run_research_topic(topic: str) -> str:
    """Use the research specialist to deeply investigate a topic."""
    specialist = Agent(
        name="Research Specialist",
        instructions="Research the given topic thoroughly. Return findings as a structured report.",
    )
    result = Runner.run_sync(specialist, topic)
    return result.final_output

@function_tool
def write_content_section(outline: str) -> str:
    """Use the writing specialist to draft a content section."""
    writer = Agent(
        name="Writing Specialist",
        instructions="Write a clear, engaging content section based on the outline provided.",
    )
    result = Runner.run_sync(writer, outline)
    return result.final_output

editor_agent = Agent(
    name="Editor-in-Chief",
    instructions="""You manage a content pipeline.
    Step 1: Research the topic using run_research_topic.
    Step 2: Write each section using write_content_section.
    Step 3: Combine into a final article."""
    tools=[run_research_topic, write_content_section],
)

Step 4: Guardrails

Guardrails are functions that run on every turn to block bad inputs or outputs. There are two types [4]:

• Input guardrails — Check the user’s input before it reaches the agent. Run on the first turn and on subsequent turns
• Output guardrails — Check the agent’s output before it’s returned to the caller

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

@input_guardrail
async def block_pii_triage(ctx, agent, input_data):
    """Block inputs containing personally identifiable information."""
    import re
    text = input_data if isinstance(input_data, str) else str(input_data)
    # Check for email addresses
    if re.search(r'[\w\.-]+@[\w\.-]+\.\w+', text):
        return GuardrailFunctionOutput(
            tripwire_triggered=True,
            output_info="Input blocked: contains PII (email address). Remove personal information and try again.",
        )
    # Check for SSN patterns
    if re.search(r'\d{3}-\d{2}-\d{4}', text):
        return GuardrailFunctionOutput(
            tripwire_triggered=True,
            output_info="Input blocked: appears to contain a social security number.",
        )
    return GuardrailFunctionOutput(
        tripwire_triggered=False,
        output_info="Input passed PII check.",
    )

agent = Agent(
    name="Safe Support Agent",
    instructions="Help customers with their inquiries. Never ask for or store PII.",
    input_guardrails=[block_pii_triage],
)

# This input will be blocked
result = Runner.run_sync(agent, "My email is [email protected] and I need help with my order.")
if result.blocked:
    print(f"Guardrail: {result.guardrail_message}")

Guardrails run every turn, not just on the first turn. This means a malicious payload injected mid-conversation gets caught. Guardrails can inspect the full conversation history if needed, making them useful for detecting gradual prompt injection attempts.

Output guardrails

from agents import GuardrailFunctionOutput, output_guardrail

@output_guardrail
async def check_hallucination(ctx, agent, output):
    """Flag outputs that make unsupported factual claims without citations."""
    text = output if isinstance(output, str) else str(output)
    factual_triggers = ["in 2026", "according to research", "studies show", "the current"]
    has_claim = any(trigger in text.lower() for trigger in factual_triggers)
    has_source = "[" in text or "source" in text.lower() or "citation" in text.lower()

    if has_claim and not has_source and len(text) > 100:
        return GuardrailFunctionOutput(
            tripwire_triggered=False,  # warn but don't block
            output_info="WARNING: Output contains factual claims without citations. Review before sending.",
        )
    return GuardrailFunctionOutput(tripwire_triggered=False, output_info="Passed.")

agent = Agent(
    name="Editorial Agent",
    instructions="...",
    output_guardrails=[check_hallucination],
)

Output guardrails default to non-blocking (advisory) — they flag issues but don’t stop the response. Set block_on=True in a @output_guardrail function to make it blocking.

Step 5: Tracing and Monitoring

The SDK includes built-in tracing that captures every event in an agent run: LLM generations, tool calls, handoffs, guardrail checks, and custom application events [5].

from agents import Agent, Runner, trace

# Tracing is automatic — every Runner.run() call creates a trace
result = Runner.run_sync(agent, "Analyze the quarterly report.")

# For multi-step workflows, wrap in a trace context
with trace("quarterly-analysis-workflow") as workflow_trace:
    step1 = Runner.run_sync(research_agent, "Gather Q2 data")
    step2 = Runner.run_sync(analysis_agent, f"Analyze: {step1.final_output}")
    step3 = Runner.run_sync(report_agent, f"Write report: {step2.final_output}")

# Add custom events for application-level tracking
workflow_trace.add_event("user_feedback", {"rating": 5, "comment": "Great analysis"})

By default, traces are sent to OpenAI’s dashboard for debugging. In production you can export them to your own observability stack:

from agents import set_trace_processors
from agents.tracing import BatchTraceProcessor, FileTraceExporter

# Write traces to JSON files for log aggregation
set_trace_processors([
    BatchTraceProcessor(
        exporter=FileTraceExporter("logs/agent-traces/"),
        batch_size=50,
    ),
])

Trace data includes:

Event	What’s recorded
agent_run	LLM response, tokens used, model name, duration
function_tool_call	Tool name, arguments, result, duration
handoff	From agent, to agent, reason
guardrail	Guardrail type, passed/blocked, message
custom	Any application-level event you add

This is the difference between wondering why an agent did something and knowing exactly which turn, tool, and guardrail was involved.

Step 6: Sessions — Persistent Conversations

By default, each Runner.run() call is stateless — no conversation history is preserved. For chat-style applications, use sessions [6]:

from agents import Agent, Runner, Session
from agents.memory import MemoryStore

store = MemoryStore(filepath="./session-store.json")

# Create or load a session
session = Session(
    id="user-abc-123",
    store=store,
)

# Run with session — conversation history persists
result1 = Runner.run_sync(
    agent,
    "What was the last topic we discussed?",
    session=session,
)

result2 = Runner.run_sync(
    agent,
    "Continue the research from where we left off.",
    session=session,  # same session, same history
)

# Session persists across application restarts
print(f"Turn count: {session.turn_count}")

For Redis-backed persistence in distributed deployments:

from agents.memory import RedisMemoryStore

store = RedisMemoryStore(
    host="redis.example.com",
    port=6379,
    ttl_seconds=86400,  # sessions expire after 24h
)

Step 7: Multi-Agent Workflow Example

Here’s a complete production pipeline that ties together all the concepts:

from agents import Agent, Runner, function_tool, input_guardrail, GuardrailFunctionOutput
from pydantic import BaseModel

# --- Step 1: Define tools ---

@function_tool
def query_database(sql: str) -> str:
    """Execute a SQL query against the analytics database."""
    # In production: connect to your data warehouse
    return f"Query executed: {sql[:50]}... returning 42 rows"

@function_tool
def send_slack_notification(channel: str, message: str) -> str:
    """Send a message to a Slack channel."""
    # In production: use Slack SDK
    return f"Message sent to #{channel}"

# --- Step 2: Define output schema ---

class AnalysisReport(BaseModel):
    period: str
    metric: str
    value: float
    comparison_to_previous: str
    actionable_insight: str
    confidence: float

# --- Step 3: Define agents ---

data_agent = Agent(
    name="Data Analyst",
    instructions="You query the database and return raw results. Only query for the specific metrics requested.",
    tools=[query_database],
)

insight_agent = Agent(
    name="Insight Generator",
    instructions="You take raw data and generate actionable business insights.",
    output_type=AnalysisReport,
)

notifier_agent = Agent(
    name="Notifier",
    instructions="You send notifications about completed analyses.",
    tools=[send_slack_notification],
)

# --- Step 4: Orchestrate ---

@input_guardrail
async def no_sql_injection(ctx, agent, input_data):
    text = input_data if isinstance(input_data, str) else str(input_data)
    if "DROP TABLE" in text.upper() or "DELETE FROM" in text.upper():
        return GuardrailFunctionOutput(
            tripwire_triggered=True,
            output_info="Destructive SQL operations are not allowed.",
        )
    return GuardrailFunctionOutput(tripwire_triggered=False)

orchestrator = Agent(
    name="Orchestrator",
    instructions="""You run the weekly analytics pipeline.
    1. Ask the Data Analyst for metrics
    2. Give raw data to the Insight Generator
    3. Have the Notifier send the summary
    """,
    input_guardrails=[no_sql_injection],
    handoffs=[data_agent, insight_agent, notifier],
)

# --- Step 5: Run (with tracing) ---
from agents import trace

with trace("weekly-metrics-run"):
    result = Runner.run_sync(
        orchestrator,
        "Run the weekly analysis for Q2 2026 revenue metrics.",
    )

Best Practices

1. Keep agent instructions focused — Each agent should have a single responsibility. If an agent’s instructions exceed 500 words, it likely needs to be split into multiple agents [1].

2. Use handoffs for domain shifts, agent-as-tool for control — Handoffs are cleaner when agents operate in entirely different domains (support triage → refund specialist). Agent-as-tool is better when you need an orchestrator that controls the sequence and aggregates results.

3. Set max_turns on production agents — Default is 10. Set it lower (3-5) for tightly scoped agents to prevent loops and control costs. Set higher for exploratory agents.

agent = Agent(
    name="Scoped Agent",
    instructions="You handle one specific task per invocation.",
    max_turns=3,  # prevent runaway loops
)

4. Structure guardrails carefully — Non-blocking guardrails that return advisory messages are useful for monitoring. Blocking guardrails should only be used for policy violations (PII, toxic content, destructive operations) that must never reach the user.

5. Use trace contexts for multi-step workflows — Wrapping a sequence of Runner.run() calls in a single with trace("workflow-name"): block links all events under one workflow ID for debugging [5].

6. Isolate agent tools by domain — Don’t share a tool across agents that operate in different domains. A search_web tool on a PII-handling agent creates a data exfiltration risk. Give each agent its own tool set scoped to its responsibility.

7. Test with max_turns=1 first — Set max_turns=1 during development to force single-turn responses and verify your agent can handle the simplest case before adding tool call complexity.

Key Takeaways

• The OpenAI Agents SDK (openai-agents) provides Agent, Runner, Tools, Handoffs, Guardrails, Tracing, and Sessions as first-class primitives on top of the Responses API.
• Handoffs delegate control to specialist agents; agent-as-tool keeps the orchestrator in control — choose the pattern that matches your architecture.
• Guardrails run on every turn, catching prompt injection and PII leaks that single-turn checks miss.
• Built-in tracing captures LLM generations, tool calls, handoffs, and guardrail events for debugging and production monitoring.
• Sessions persist conversation state across runs, with optional Redis backend for distributed deployments.
• Use the Responses API directly when one model call with tools is enough; use the SDK when you need orchestration, state, and guardrails.

References

• OpenAI Agents SDK Documentation — Official documentation and quickstart [1]
• Running Agents — Runner class, agent loop lifecycle, max turns [2]
• Handoffs in the Agents SDK — Agent routing, handoff mechanics, escalation [3]
• Guardrails Reference — Input and output guardrail API [4]
• Tracing in the Agents SDK — Built-in tracing, trace processors, file export [5]
• Sessions and Memory — Persistent conversation state, MemoryStore, Redis [6]
• Agent Output Type — Structured output with Pydantic schemas [7]
• Function Tools in the Agents SDK — Tool registration, type inference, async tools [8]
• OpenAI: A Practical Guide to Building Agents — Official best practices for agent design [9]
• PyPI: openai-agents — Package index, release history, optional redis extras [10]

📖 Related Reads

• NiteAgent: OpenAI Responses API Agent Guide 2026 — The underlying API the Agents SDK orchestrates
• NiteAgent: PydanticAI Structured Agent Workflows — Type-safe agent development patterns
• NiteAgent: Production Tool-Calling Architecture — Patterns for reliable tool dispatch in production

Cross-links automatically generated from NiteAgent.

← Back to all posts