Openai Agent Sdk Integration

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

Core Concepts

Three Primitives of OpenAI Agents SDK

Three Primitives of MCP

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

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

Runner Methods

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