18_MCP_INTEGRATION

Shared from "Study" 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

Standardization - Consistent tool interface

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],
)

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],
)

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],
)

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
)

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,
)

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,
)

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,
)

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."""

Python

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

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(),
)

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,
)

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)

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

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

Python

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

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

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],
)

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

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"]

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

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}"

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
    )
)

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)],
)

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

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)

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