MCP servers with OpenAI Agents SDK

OpenAI Agents SDK MCP architecture

The SDK treats MCP servers as a source of tools at the agent level, separate from native Python tools. When a connection opens, the SDK calls tools/list and converts each MCP tool's inputSchema into an OpenAI FunctionDefinition. These are merged with any native Python tools and sent to the model in a single tools array.

By default, the SDK opens an MCP connection at the start of each Runner.run() call and closes it when the run completes. For a service handling many requests, this means a full MCP handshake per request — use persistent connections to eliminate that overhead.

Basic setup with MCPServerHTTP

import asyncio
from openai_agents import Agent, Runner
from openai_agents.mcp import MCPServerHTTP

# Attach an HTTP/SSE MCP server to an agent
research_agent = Agent(
    name="ResearchAgent",
    model="gpt-4o",
    instructions="You are a research assistant. Use the search and fetch tools to answer questions thoroughly.",
    mcp_servers=[
        MCPServerHTTP(
            url="https://search.internal/mcp",
            headers={"Authorization": "Bearer sk-..."},
            timeout=30,
        ),
    ],
)

async def main():
    result = await Runner.run(
        research_agent,
        "What are the most common MCP server failure modes in production?"
    )
    print(result.final_output)

asyncio.run(main())

The MCPServerHTTP class supports both HTTP streaming (SSE transport) and streamable HTTP. The SDK negotiates the transport version during initialize. For MCP servers using stdio transport (local processes), use MCPServerStdio instead.

Stdio MCP servers for local tools

from openai_agents.mcp import MCPServerStdio

code_agent = Agent(
    name="CodeAgent",
    model="gpt-4o",
    instructions="You are a code execution agent. Write and run Python and JavaScript code to solve problems.",
    mcp_servers=[
        MCPServerStdio(
            command="npx",
            args=["-y", "@acme/code-runner-mcp"],
            env={"NODE_ENV": "production"},
        ),
    ],
)

# The SDK spawns the subprocess at Runner.run() and terminates it on completion

Each MCPServerStdio instance spawns a subprocess for the duration of the agent run. If you run multiple agents with the same stdio MCP server config, each agent run creates its own subprocess instance — there's no subprocess sharing. Use HTTP/SSE MCP servers in production to avoid subprocess management overhead and to share a single server instance across multiple agent runs.

Persistent MCP connections for services

In a FastAPI or ASGI service, re-opening the MCP connection on every request adds 50–300 ms of handshake latency. Use agent.run_mcp_servers() to open the connection once at startup and reuse it across all requests:

from contextlib import asynccontextmanager
from fastapi import FastAPI
from openai_agents import Agent, Runner
from openai_agents.mcp import MCPServerHTTP

search_agent = Agent(
    name="SearchAgent",
    model="gpt-4o",
    instructions="Answer questions using the search tools.",
    mcp_servers=[
        MCPServerHTTP(url="https://search.internal/mcp"),
    ],
)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Open MCP connection once at service startup
    async with search_agent.run_mcp_servers():
        yield
    # Connection closes at shutdown

app = FastAPI(lifespan=lifespan)

@app.post("/ask")
async def ask(question: str):
    result = await Runner.run(search_agent, question)
    return {"answer": result.final_output}

The tool list is fetched once when run_mcp_servers() opens the connection. If the MCP server adds or removes tools while the connection is live, the agent won't see those changes until the connection is closed and reopened. Build MCP servers so that tool additions are backward-compatible (add tools, don't remove them) to avoid requiring service restarts.

Multi-agent handoffs with MCP tools

The OpenAI Agents SDK's Handoffs feature routes the conversation to a specialist agent based on the current context. MCP tools from the delegating agent are not automatically passed to the target agent — each agent carries its own MCP server list:

from openai_agents import Agent, Runner, handoff
from openai_agents.mcp import MCPServerHTTP

# Specialist agent with its own MCP tools
data_agent = Agent(
    name="DataAgent",
    model="gpt-4o",
    instructions="Query databases and return structured data. Summarize findings clearly.",
    mcp_servers=[
        MCPServerHTTP(url="https://analytics.internal/mcp"),
    ],
)

# Triage agent — no MCP tools, but can hand off to data_agent
triage_agent = Agent(
    name="TriageAgent",
    model="gpt-4o-mini",
    instructions="Understand the user's question. If it requires data analysis, hand off to the DataAgent.",
    handoffs=[handoff(data_agent)],
)

async def handle_request(question: str) -> str:
    result = await Runner.run(triage_agent, question)
    return result.final_output

When triage_agent hands off to data_agent, the SDK opens a new MCP connection for data_agent's MCPServerHTTP instance. If you're using persistent connections via run_mcp_servers(), open connections for all agents in your handoff graph at startup — not just the entry-point agent.

Guardrails and MCP tool results

OpenAI Agents SDK Guardrails run as tripwires on agent input and output. They execute in parallel with the agent run and can trigger an early termination if a safety condition is met. MCP tool results flow through the agent's message history and are therefore visible to output Guardrails:

from openai_agents import Agent, Runner, GuardrailFunctionOutput, output_guardrail
from openai_agents.mcp import MCPServerHTTP
from pydantic import BaseModel

class SafetyCheck(BaseModel):
    is_safe: bool
    reason: str

# Guardrail that checks agent output (including tool-informed responses) for safety
@output_guardrail
async def content_safety_guardrail(ctx, agent, output) -> GuardrailFunctionOutput:
    # Run a fast safety check on the final output
    check = await fast_safety_model.run(output.final_output)
    return GuardrailFunctionOutput(
        output_info=check.data,
        tripwire_triggered=not check.data.is_safe,
    )

monitored_agent = Agent(
    name="MonitoredAgent",
    model="gpt-4o",
    instructions="Answer questions using available tools.",
    mcp_servers=[MCPServerHTTP(url="https://tools.internal/mcp")],
    output_guardrails=[content_safety_guardrail],
)

MCP tool call arguments (what the LLM sends to the tool) are visible to tracing but not to Guardrails by default. If your MCP tools can receive sensitive instructions through LLM-generated arguments, add input Guardrails that inspect the agent's input before tool calls execute. The SDK does not currently provide a per-tool-call hook for Guardrails.

Tracing MCP tool calls

The SDK's built-in tracing captures the full agent run timeline, including which MCP tools were called and with what arguments. Traces are sent to the OpenAI dashboard by default, or can be routed to external systems via a custom trace exporter:

from openai_agents import Agent, Runner, add_trace_processor
from openai_agents.tracing import TracingProcessor

class MCPAuditProcessor(TracingProcessor):
    """Log every MCP tool call for audit and debugging."""
    def on_span_end(self, span):
        if span.span_data.type == "function":
            # span.span_data captures: tool name, input, output, duration
            print(f"MCP tool: {span.span_data.name}")
            print(f"  Input:  {span.span_data.input}")
            print(f"  Output: {span.span_data.output[:200]}")
            print(f"  Time:   {span.ended_at - span.started_at:.3f}s")

add_trace_processor(MCPAuditProcessor())

# All subsequent Runner.run() calls will emit MCP tool call spans

MCP tool call duration in traces tells you the response-time profile of each server. If a particular MCP tool consistently takes 2+ seconds when others complete in 200 ms, that server is a latency outlier — worth investigating with distributed tracing inside the MCP server itself. AliveMCP's per-server response-time history complements SDK-level traces by providing an external measurement of server health over time.

Monitoring MCP servers in OpenAI agent pipelines

The SDK's error behavior when an MCP server is unavailable depends on when the failure occurs. If the server is down when run_mcp_servers() opens the connection, the SDK raises a connection error immediately — visible and catchable. If the server goes down while the persistent connection is live (between agent runs), the SDK's next tools/call attempt fails with a transport error mid-run — the agent sees no tool results and may hallucinate answers or loop.

Neither failure mode is detectable from inside the SDK without external observability. AliveMCP probes the MCP server's initialize endpoint every 60 seconds and alerts on the first failure. In a persistent-connection setup, this means you know the server is down within a minute — before most agent runs would detect the failure themselves. Set AliveMCP alerts to route to the same on-call channel as your OpenAI usage alerts: when both fire together, it's the MCP server; when only the OpenAI alert fires, it's the model or the application.

Further reading

• MCP servers with AWS Bedrock — action groups and Converse API integration
• MCP servers with Google Gemini — function calling and ADK integration
• MCP servers with Groq — ultra-fast inference and parallel tool calls
• MCP server LangChain integration — MultiServerMCPClient and ReAct agents
• MCP server tool design — flat schemas and LLM-friendly descriptions
• MCP server tracing — distributed observability for tool call pipelines
• AliveMCP — continuous protocol monitoring for MCP servers