Openai Agents Sdk

name: openai-agents-sdk description: OpenAI Agents SDK patterns for multi-agent systems with handoffs, guardrails, tracing, and MCP support. Use when building production agent applications using OpenAI models with tool use, agent delegation, or MCP server integration. tags: [openai, agents, multi-agent, mcp]

OpenAI Agents SDK

Build production-grade multi-agent AI applications using the OpenAI Agents SDK with handoffs, guardrails, tracing, and MCP server integration.

When to Use

• Building multi-agent systems with specialized agents that hand off to each other
• Integrating MCP servers as tool providers for OpenAI agents
• Adding guardrails for input/output validation and safety
• Implementing tracing and observability for agent workflows
• Building agents that use web search, file search, or code interpreter

Built-in Tools

Patterns

1. Basic Agent with Function Tools

from agents import Agent, Runner, function_tool

@function_tool
def get_customer(customer_id: str) -> dict:
    """Look up customer details by ID."""
    # Call your enterprise API
    return {"id": customer_id, "name": "Acme Corp", "plan": "enterprise"}

@function_tool
def create_ticket(subject: str, description: str, priority: str = "medium") -> dict:
    """Create a support ticket."""
    return {"ticket_id": "TK-1234", "status": "open"}

agent = Agent(
    name="Customer Service",
    instructions="Help customers by looking up their info and creating tickets when needed.",
    tools=[get_customer, create_ticket],
)

async def main():
    result = await Runner.run(agent, "I need help with my account, customer ID is C-456")
    print(result.final_output)

2. Multi-Agent Handoffs

from agents import Agent, Runner

billing_agent = Agent(
    name="Billing Specialist",
    instructions="Handle billing inquiries. Look up invoices and payment status.",
    tools=[get_invoice, process_refund],
)

shipping_agent = Agent(
    name="Shipping Specialist",
    instructions="Handle shipping inquiries. Track orders and manage returns.",
    tools=[track_order, initiate_return],
)

triage_agent = Agent(
    name="Triage Agent",
    instructions="""Route customer inquiries to the right specialist:
    - Billing questions -> transfer to Billing Specialist
    - Shipping/delivery questions -> transfer to Shipping Specialist""",
    handoffs=[billing_agent, shipping_agent],
)

async def main():
    result = await Runner.run(triage_agent, "Where is my order #12345?")
    # Triage agent hands off to shipping_agent automatically
    print(result.final_output)

3. MCP Server Integration

from agents import Agent, Runner
from agents.mcp import MCPServerStdio

# Connect to enterprise MCP servers
postgres_server = MCPServerStdio(
    command="npx",
    args=["-y", "@modelcontextprotocol/server-postgres", "postgresql://user:pass@host/db"],
)

servicenow_server = MCPServerStdio(
    command="python",
    args=["-m", "mcp_servicenow"],
    env={"SN_INSTANCE": "mycompany", "SN_USER": "admin", "SN_PASS": "secret"},
)

agent = Agent(
    name="IT Operations Agent",
    instructions="Query the database for system metrics and create ServiceNow incidents for issues.",
    mcp_servers=[postgres_server, servicenow_server],
)

async def main():
    async with postgres_server, servicenow_server:
        result = await Runner.run(agent, "Check if any servers have CPU > 90% and create incidents")
        print(result.final_output)

4. Guardrails

from agents import Agent, Runner, InputGuardrail, GuardrailFunctionOutput

@InputGuardrail
async def check_for_pii(ctx, agent, input_text: str) -> GuardrailFunctionOutput:
    """Block requests containing PII."""
    import re
    ssn_pattern = r'\b\d{3}-\d{2}-\d{4}\b'
    if re.search(ssn_pattern, input_text):
        return GuardrailFunctionOutput(
            output_info={"reason": "SSN detected in input"},
            tripwire_triggered=True,
        )
    return GuardrailFunctionOutput(output_info={"reason": "clean"}, tripwire_triggered=False)

agent = Agent(
    name="Secure Agent",
    instructions="Help with customer inquiries.",
    input_guardrails=[check_for_pii],
)

5. Streaming with Tracing

from agents import Agent, Runner

agent = Agent(name="Assistant", instructions="Help the user.")

async def main():
    result = Runner.run_streamed(agent, "Explain quantum computing")
    async for event in result.stream_events():
        if event.type == "raw_response_event":
            # Stream tokens as they arrive
            if hasattr(event.data, "delta"):
                print(event.data.delta, end="", flush=True)

    # Access trace for observability
    print(f"\nTrace ID: {result.trace_id}")

Anti-Patterns

• Creating deeply nested handoff chains -- keep delegation to 2-3 levels max
• Running MCP servers without async with context manager -- leaks resources
• Skipping guardrails for user-facing agents -- always validate input/output
• Using synchronous function tools for I/O-bound operations -- use async tools

References

• OpenAI Agents SDK GitHub
• OpenAI Agents Documentation
• OpenAI Function Calling