OpenAI Agents SDK Integration with MCP

This skill provides comprehensive guidance for building AI agents using the OpenAI Agents SDK with MCP (Model Context Protocol) server integration.

Before Implementation

Source	Gather
Codebase	Existing agent patterns, MCP server implementations, tool definitions
Conversation	User's specific requirements, transport type (HTTP/stdio), tool needs
Skill References	Domain patterns from references/ (SDK patterns, MCP primitives, examples)
User Guidelines	Project-specific conventions, authentication requirements

Core Concepts

Three Primitives of OpenAI Agents SDK

Primitive	Purpose	Control Model
Agents	LLMs with instructions and tools	Framework-controlled
Handoffs	Delegation between agents	Agent-controlled
Guardrails	Input/output validation	Developer-controlled

Three Primitives of MCP

Primitive	Purpose	Control Model
Tools	Functions AI can invoke	Model-controlled
Resources	Read-only data access	Application-controlled
Prompts	Pre-crafted instructions	User-controlled

Quick Start

1. Create MCP Server

from mcp.server.fastmcp import FastMCP

mcp = FastMCP(
    name="my-mcp-server",
    stateless_http=True,
    json_response=True
)

@mcp.tool(name="my_tool", description="Tool description")
def my_tool(param: str) -> str:
    return f"Result: {param}"

# Create HTTP app
streamable_http_app = mcp.streamable_http_app()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("server:streamable_http_app", host="0.0.0.0", port=8001)

2. Create Agent with MCP Connection

import asyncio
from agents import Agent, Runner, OpenAIChatCompletionsModel
from agents.mcp import MCPServerStreamableHttp, MCPServerStreamableHttpParams
from openai import AsyncOpenAI

MCP_SERVER_URL = "http://localhost:8001/mcp/"

async def main():
    client = AsyncOpenAI(api_key="your-api-key")
    mcp_params = MCPServerStreamableHttpParams(url=MCP_SERVER_URL)

    async with MCPServerStreamableHttp(
        params=mcp_params,
        name="MyMCPClient"
    ) as mcp_server:
        agent = Agent(
            name="MyAgent",
            instructions="You are a helpful assistant.",
            mcp_servers=[mcp_server],
            model=OpenAIChatCompletionsModel(model="gpt-4o", openai_client=client)
        )

        result = await Runner.run(agent, "Your query here")
        print(result.final_output)

asyncio.run(main())

MCP Transport Types

Transport	Class	Use Case
HTTP Streamable	MCPServerStreamableHttp	Production, remote servers
SSE	MCPServerSse	Legacy systems (deprecated)
Stdio	MCPServerStdio	Local development, subprocesses

Key Patterns

Tool Caching

MCPServerStreamableHttp(
    params=mcp_params,
    cache_tools_list=True  # Enable caching
)

Static Tool Filtering

from agents.mcp import create_static_tool_filter

tool_filter = create_static_tool_filter(
    allowed_tool_names=["allowed_tool"],
    blocked_tool_names=["blocked_tool"]
)

MCPServerStreamableHttp(params=mcp_params, tool_filter=tool_filter)

Dynamic Tool Filtering

from agents.mcp import ToolFilterContext

def context_aware_filter(context: ToolFilterContext, tool) -> bool:
    return context.agent.name == "MyAgent" and tool.name == "my_tool"

MCPServerStreamableHttp(params=mcp_params, tool_filter=context_aware_filter)

Multiple MCP Servers

from contextlib import AsyncExitStack

async def connect_multiple_servers(urls: list):
    mcp_servers = []
    async with AsyncExitStack() as stack:
        for url in urls:
            params = MCPServerStreamableHttpParams(url=url)
            client = await stack.enter_async_context(
                MCPServerStreamableHttp(params=params, name=f"Client_{url}")
            )
            mcp_servers.append(client)

        agent = Agent(
            name="MultiServerAgent",
            mcp_servers=mcp_servers,
            model=model
        )
        return agent

Agent Configuration

Agent Parameters

Parameter	Type	Description
name	str	Agent identifier
instructions	str/callable	System prompt or dynamic function
model	Model	LLM configuration
tools	list	Local function tools
mcp_servers	list	MCP server connections
handoffs	list	Delegation targets
output_type	Pydantic	Structured output schema

Runner Methods

Method	Type	Description
Runner.run()	async	Returns RunResult
Runner.run_sync()	sync	Wrapper for .run()
Runner.run_streamed()	async	Streaming with RunResultStreaming

MCP Server Patterns

Defining Tools

@mcp.tool(name="tool_name", description="Tool description")
def my_tool(param: str, optional: str = "default") -> str:
    """Docstring becomes parameter description."""
    return f"Result: {param}"

Defining Resources

@mcp.resource("docs://{doc_id}", mime_type="text/plain")
def get_document(doc_id: str) -> str:
    return documents[doc_id]

Defining Prompts

from pydantic import Field

@mcp.prompt(name="my_prompt", description="Prompt description")
def generate_instructions(focus: str = Field(description="Focus area")) -> str:
    return f"Instructions focused on {focus}..."

Guardrails

Input Guardrail

from agents import input_guardrail, GuardrailFunctionOutput

@input_guardrail
async def validate_input(ctx, agent, input):
    if "forbidden" in input:
        return GuardrailFunctionOutput(tripwire_triggered=True)
    return GuardrailFunctionOutput(tripwire_triggered=False)

Output Guardrail

from agents import output_guardrail

@output_guardrail
async def validate_output(ctx, agent, output):
    # Validate final output
    return GuardrailFunctionOutput(tripwire_triggered=False)

Handoffs

Basic Handoff

from agents import handoff

specialist = Agent(name="Specialist", instructions="...")

main_agent = Agent(
    name="Main",
    handoffs=[specialist]  # Direct handoff
)

Custom Handoff

from agents import handoff

custom_handoff = handoff(
    agent=specialist,
    tool_name_override="escalate_to_specialist",
    tool_description_override="Transfer to specialist for complex issues"
)

Context Management

from dataclasses import dataclass
from agents import RunContextWrapper

@dataclass
class MyContext:
    user_id: str
    session_id: str

async def main():
    ctx = MyContext(user_id="123", session_id="abc")
    result = await Runner.run(agent, "query", context=ctx)

Tracing

from agents import trace

with trace("My Workflow"):
    result1 = await Runner.run(agent, "First query")
    result2 = await Runner.run(agent, "Second query")

Best Practices

1. Use stateless HTTP transport for production MCP servers
2. Enable tool caching when tool definitions are stable
3. Implement tool filtering for access control
4. Use AsyncExitStack for multiple server connections
5. Add guardrails for input/output validation
6. Use context management for dependency injection
7. Enable tracing for observability

Error Handling

try:
    result = await Runner.run(agent, query)
except MaxTurnsExceeded:
    # Handle max turns exceeded
    pass
except InputGuardrailTripwireTriggered:
    # Handle guardrail triggered
    pass

Reference Files

File	Content
references/agent-patterns.md	Agent creation and configuration patterns
references/mcp-server-patterns.md	MCP server implementation patterns
references/tool-filtering.md	Static and dynamic tool filtering
references/multi-agent-orchestration.md	Multi-agent and handoff patterns
references/transport-options.md	HTTP, SSE, and stdio transports
references/guardrails-tracing.md	Guardrails and observability