InkdownInkdown
Start writing

OpenAI Agents Python

21 files·0 subfolders

Shared Workspace

OpenAI Agents Python
00_OVERVIEW.md

18_MCP_INTEGRATION

Shared from "OpenAI Agents Python" on Inkdown

MCP Integration - Comprehensive Deep Dive

Overview

MCP (Model Context Protocol) Integration in the OpenAI Agents SDK enables agents to use tools provided by MCP servers. Think of MCP as a "universal tool protocol" that allows different systems to expose tools to AI models in a standardized way. This is essential for integrating with external tool providers, sharing tools across applications, and building a tool ecosystem.

Core Concepts

What is MCP?

MCP (Model Context Protocol) is:

  • A protocol for exposing tools to AI models
  • Standardized - consistent interface across providers
  • Language-agnostic - works with any programming language
  • Extensible - supports custom tool types
  • Community-driven - open ecosystem of tool providers
Why MCP Matters
  1. - Consistent tool interface
01_AGENT_SYSTEM.md
02_RUNNER_SYSTEM.md
03_TOOL_SYSTEM.md
04_ITEMS_SYSTEM.md
05_GUARDRAILS.md
06_HANDOFFS.md
07_MEMORY_SESSIONS.md
08_MODEL_PROVIDERS.md
09_SANDBOX_SYSTEM.md
10_TRACING.md
11_RUN_STATE.md
12_CONTEXT.md
13_LIFECYCLE_HOOKS.md
14_CONFIGURATION.md
15_ERROR_HANDLING.md
16_STREAMING.md
17_EXTENSIONS.md
18_MCP_INTEGRATION.md
19_BEST_PRACTICES.md
20_ARCHITECTURE_PATTERNS.md
Standardization
  • Integration - Easy integration with external systems
  • Ecosystem - Access to community tools
  • Flexibility - Mix and match tool providers
  • Future-proof - Protocol evolves with community input
  • Interoperability - Works across different AI platforms

    • MCP Architecture

    MCP Components
    Plain text
    Agent
  ↓
MCP Client (SDK)
  ↓
MCP Server (External)
  ↓
Tools

    Flow:

    1. Agent requests tools
    2. MCP client connects to MCP server
    3. MCP server lists available tools
    4. Tools are exposed to agent
    5. Agent calls tools via MCP client
    6. MCP server executes tools
    7. Results returned to agent

    MCP Server Types

    MCPServerStdio

    Standard I/O based MCP server:

    Python
    from agents.mcp import MCPServerStdio, MCPServerStdioParams

server = MCPServerStdio(
    params=MCPServerStdioParams(
        command="python",
        args=["-m", "my_mcp_server"],
    )
)

agent = Agent(
    name="mcp_agent",
    instructions="Use MCP tools",
    mcp_servers=[server],
)

    When to use:

    • Local MCP servers
    • Command-line based servers
    • Simple deployments
    MCPServerSse

    Server-Sent Events based MCP server:

    Python
    from agents.mcp import MCPServerSse, MCPServerSseParams

server = MCPServerSse(
    params=MCPServerSseParams(
        url="https://example.com/mcp/sse",
    )
)

agent = Agent(
    name="mcp_agent",
    instructions="Use MCP tools",
    mcp_servers=[server],
)

    When to use:

    • Remote MCP servers
    • Real-time updates
    • Web-based deployments
    MCPServerStreamableHttp

    Streamable HTTP based MCP server:

    Python
    from agents.mcp import MCPServerStreamableHttp, MCPServerStreamableHttpParams

server = MCPServerStreamableHttp(
    params=MCPServerStreamableHttpParams(
        url="https://example.com/mcp",
    )
)

agent = Agent(
    name="mcp_agent",
    instructions="Use MCP tools",
    mcp_servers=[server],
)

    When to use:

    • HTTP-based MCP servers
    • Streaming tool execution
    • Complex server deployments

    MCP Configuration

    MCPConfig

    Configure MCP behavior:

    Python
    from agents.mcp import MCPConfig

config = MCPConfig(
    convert_schemas_to_strict=True,  # Convert to strict JSON schema
    failure_error_function=default_tool_error_function,  # Error handling
    tool_filter=tool_filter,  # Filter tools
)
    Schema Conversion

    Convert tool schemas to strict mode:

    Python
    config = MCPConfig(
    convert_schemas_to_strict=True,
)

    Benefits:

    • Better model understanding
    • Reduced errors
    • Improved tool calling
    Error Handling

    Custom error handling for MCP tools:

    Python
    def mcp_error_function(
    context: RunContextWrapper,
    tool_name: str,
    error: Exception,
) -> str:
    """Custom MCP error handler."""
    return f"MCP tool {tool_name} error: {str(error)}"

config = MCPConfig(
    failure_error_function=mcp_error_function,
)

    MCP Tool Filtering

    Static Tool Filter

    Filter tools statically:

    Python
    from agents.mcp import ToolFilterStatic

filter_config = ToolFilterStatic(
    include=["tool1", "tool2"],  # Only include these
    exclude=["tool3"],  # Exclude these
)

config = MCPConfig(
    tool_filter=filter_config,
)
    Dynamic Tool Filter

    Filter tools dynamically:

    Python
    from agents.mcp import ToolFilterCallable, ToolFilterContext

def dynamic_filter(context: ToolFilterContext) -> ToolFilter:
    """Dynamic tool filtering."""
    # Filter based on context
    if context.context.user_tier == "free":
        return ToolFilterStatic(include=["basic_tool"])
    else:
        return ToolFilterStatic(include=["basic_tool", "premium_tool"])

config = MCPConfig(
    tool_filter=dynamic_filter,
)
    Tool Filter Context

    Context available for filtering:

    Python
    @dataclass
class ToolFilterContext:
    agent: Agent
    """The agent requesting tools."""
    
    context: Any
    """User-provided context."""
    
    tool_name: str | None
    """Specific tool name if filtering one tool."""

    MCP Tool Metadata

    MCPToolMetaContext

    Context for tool metadata resolution:

    Python
    @dataclass
class MCPToolMetaContext:
    agent: Agent
    """The agent."""
    
    tool_name: str
    """The tool name."""
    
    original_schema: dict
    """The original schema from MCP."""
    MCPToolMetaResolver

    Resolve tool metadata:

    Python
    from agents.mcp import MCPToolMetaResolver

class CustomMetaResolver(MCPToolMetaResolver):
    def resolve(
        self,
        context: MCPToolMetaContext,
    ) -> ToolMetadata:
        """Resolve tool metadata."""
        # Customize metadata
        return ToolMetadata(
            name=context.tool_name,
            description="Custom description",
            schema=context.original_schema,
        )

config = MCPConfig(
    tool_meta_resolver=CustomMetaResolver(),
)

    MCP Approval

    Local Approval

    Require approval for MCP tools:

    Python
    from agents.mcp import LocalMCPApprovalCallable

async def approve_mcp_tool(
    context: RunContextWrapper,
    tool_name: str,
    arguments: str,
) -> bool:
    """Approve or reject MCP tool."""
    # Check if tool should be approved
    if tool_name in ["safe_tool1", "safe_tool2"]:
        return True
    return False

server = MCPServerStdio(
    params=...,
    approval=approve_mcp_tool,
)
    Approval Items

    MCP approval creates items:

    Python
    # MCPApprovalRequestItem - Request approval
# MCPApprovalResponseItem - Response to approval

    MCP Manager

    MCPServerManager

    Manage multiple MCP servers:

    Python
    from agents.mcp import MCPServerManager

manager = MCPServerManager()

# Add servers
manager.add_server("server1", server1)
manager.add_server("server2", server2)

# Get all tools
all_tools = await manager.get_all_tools()

# Call a tool
result = await manager.call_tool("server1", "tool_name", arguments)
    Server Lifecycle

    Manage server lifecycle:

    Python
    manager = MCPServerManager()

# Start servers
await manager.start()

# Use servers
result = await Runner.run(agent, input, mcp_manager=manager)

# Stop servers
await manager.stop()

    MCP and Agents

    Adding MCP to Agents

    Add MCP servers to agents:

    Python
    server = MCPServerStdio(params=...)

agent = Agent(
    name="mcp_agent",
    instructions="Use MCP tools when needed",
    mcp_servers=[server],
)
    MCP Tool Resolution

    Tools are resolved at runtime:

    Python
    # Tools are fetched when agent runs
result = await Runner.run(agent, input)

# The agent can use MCP tools
    MCP and Handoffs

    MCP tools work with handoffs:

    Python
    agent1 = Agent(
    name="agent1",
    instructions="Use MCP tools",
    mcp_servers=[server1],
)

agent2 = Agent(
    name="agent2",
    instructions="Use MCP tools",
    mcp_servers=[server2],
)

# Handoff preserves MCP server context

    MCP and Tracing

    MCP Tracing

    MCP operations are traced:

    Python
    result = await Runner.run(agent, input)

# Trace includes MCP tool calls
for span in result.trace.spans:
    if span.name.startswith("mcp_"):
        print(f"MCP operation: {span.name}")
    MCP Span Types

    MCP creates specific spans:

    • mcp_list_tools - Listing available tools
    • mcp_call_tool - Calling a tool
    • mcp_approve - Approval process

    MCP Best Practices

    1. Use Appropriate Server Type

    Choose the right server type:

    Python
    # Good - local server
server = MCPServerStdio(params=...)

# Good - remote server
server = MCPServerSse(params=...)

# Avoid - wrong type for use case
server = MCPServerStdio(params=...)  # For remote server
    2. Filter Tools

    Filter tools to reduce complexity:

    Python
    # Good - filter tools
config = MCPConfig(
    tool_filter=ToolFilterStatic(include=["tool1", "tool2"]),
)

# Avoid - all tools
config = MCPConfig()  # Might be too many tools
    3. Handle Errors

    Handle MCP errors gracefully:

    Python
    # Good - custom error handler
config = MCPConfig(
    failure_error_function=custom_error_handler,
)

# Avoid - default errors
config = MCPConfig()  # Generic error messages
    4. Use Strict Schemas

    Use strict schemas for better results:

    Python
    # Good - strict schemas
config = MCPConfig(
    convert_schemas_to_strict=True,
)

# Avoid - loose schemas
config = MCPConfig(
    convert_schemas_to_strict=False,
)
    5. Approve Sensitive Tools

    Require approval for sensitive tools:

    Python
    # Good - approval for sensitive tools
server = MCPServerStdio(
    params=...,
    approval=approve_sensitive,
)

# Avoid - no approval
server = MCPServerStdio(params=...)  # All tools auto-approved

    Common MCP Patterns

    1. Multiple MCP Servers

    Use multiple MCP servers:

    Python
    server1 = MCPServerStdio(params=MCPServerStdioParams(...))
server2 = MCPServerSse(params=MCPServerSseParams(...))

agent = Agent(
    name="multi_mcp",
    instructions="Use tools from multiple MCP servers",
    mcp_servers=[server1, server2],
)
    2. Conditional MCP

    Use MCP conditionally:

    Python
    def should_use_mcp(context: RunContextWrapper) -> bool:
    """Check if MCP should be used."""
    return context.context.user_tier == "premium"

if should_use_mcp(context):
    agent = Agent(mcp_servers=[server])
else:
    agent = Agent()  # No MCP
    3. MCP Fallback

    Fallback if MCP unavailable:

    Python
    try:
    server = MCPServerStdio(params=...)
    agent = Agent(mcp_servers=[server])
except Exception:
    # Fallback to local tools
    agent = Agent(tools=[local_tool])
    4. MCP Tool Namespacing

    Namespace MCP tools:

    Python
    config = MCPConfig(
    tool_meta_resolver=NamespacingMetaResolver(prefix="mcp_"),
)
    5. MCP Caching

    Cache MCP tool listings:

    Python
    class CachedMCPServerManager(MCPServerManager):
    def __init__(self):
        super().__init__()
        self._tool_cache = {}
    
    async def get_all_tools(self):
        if "tools" not in self._tool_cache:
            self._tool_cache["tools"] = await super().get_all_tools()
        return self._tool_cache["tools"]

    MCP and Testing

    Testing MCP Integration

    Test MCP tool calls:

    Python
    @pytest.mark.asyncio
async def test_mcp_tool():
    """Test MCP tool call."""
    server = MockMCPServer()
    agent = Agent(mcp_servers=[server])
    
    result = await Runner.run(agent, "Call the MCP tool")
    
    assert "mcp_result" in result.final_output
    Mock MCP Server

    Mock MCP server for testing:

    Python
    class MockMCPServer:
    """Mock MCP server for testing."""
    
    async def list_tools(self):
        return [MockTool("test_tool")]
    
    async def call_tool(self, name, arguments):
        return f"Mock result for {name}"

    MCP and Security

    Secure Connections

    Use secure connections for MCP:

    Python
    # Good - secure connection
server = MCPServerSse(
    params=MCPServerSseParams(
        url="https://example.com/mcp",
        headers={"Authorization": "Bearer token"},
    )
)

# Avoid - insecure connection
server = MCPServerSse(
    params=MCPServerSseParams(
        url="http://example.com/mcp",  # Not secure
    )
)
    Tool Validation

    Validate MCP tools:

    Python
    def validate_mcp_tool(tool: MCPServerTool) -> bool:
    """Validate MCP tool."""
    # Check tool schema
    if not tool.input_schema:
        return False
    
    # Check tool name
    if not re.match(r'^[a-z_]+$', tool.name):
        return False
    
    return True

config = MCPConfig(
    tool_filter=lambda ctx: [t for t in tools if validate_mcp_tool(t)],
)

    MCP and Performance

    Connection Pooling

    Pool MCP connections:

    Python
    class PooledMCPServerManager(MCPServerManager):
    def __init__(self, pool_size=10):
        super().__init__()
        self.pool = asyncio.Queue(maxsize=pool_size)
    Tool Caching

    Cache MCP tool results:

    Python
    class CachedMCPCalls:
    def __init__(self):
        self.cache = {}
    
    async def call_tool(self, name, arguments):
        key = f"{name}:{arguments}"
        if key in self.cache:
            return self.cache[key]
        
        result = await actual_call(name, arguments)
        self.cache[key] = result
        return result

    MCP and Monitoring

    Track MCP Usage

    Track MCP tool usage:

    Python
    from prometheus_client import Counter

mcp_tool_calls = Counter(
    "mcp_tool_calls_total",
    "MCP tool calls",
    ["tool_name", "server"]
)

async def tracked_call_tool(name, arguments):
    """Track MCP tool calls."""
    mcp_tool_calls.labels(tool_name=name, server="server1").inc()
    return await actual_call(name, arguments)
    Monitor MCP Performance

    Monitor MCP performance:

    Python
    from prometheus_client import Histogram

mcp_duration = Histogram(
    "mcp_tool_duration_seconds",
    "MCP tool duration",
    ["tool_name"]
)

async def monitored_call_tool(name, arguments):
    """Monitor MCP tool performance."""
    start = time.time()
    result = await actual_call(name, arguments)
    duration = time.time() - start
    mcp_duration.labels(tool_name=name).observe(duration)
    return result

    Summary

    MCP Integration enables standardized tool access. Key takeaways:

    1. MCP is a protocol for exposing tools to AI models
    2. MCPServerStdio - Standard I/O based server
    3. MCPServerSse - Server-Sent Events based server
    4. MCPServerStreamableHttp - Streamable HTTP server
    5. MCPConfig - Configure MCP behavior
    6. Schema conversion - Convert to strict JSON schema
    7. Error handling - Custom error handlers
    8. Tool filtering - Static and dynamic filtering
    9. Tool metadata - Resolve tool metadata
    10. Approval - Require approval for tools
    11. MCPServerManager - Manage multiple servers
    12. Agent integration - Add MCP to agents
    13. Tracing - MCP operations are traced
    14. Appropriate server type - choose right type
    15. Filter tools - reduce complexity
    16. Handle errors - graceful error handling
    17. Strict schemas - better model understanding
    18. Approval - for sensitive tools
    19. Multiple servers - use multiple MCP servers
    20. Security - use secure connections

    MCP Integration is essential for accessing a standardized ecosystem of tools.