Model Context Protocol (MCP)

Standardized protocol for connecting AI models to tools, data sources, and capabilities.

What is MCP?

MCP = Model Context Protocol

A standardized way for AI models to discover, invoke, and interact with external tools, resources, and prompts.

Why MCP?

Architecture

┌─────────────┐     MCP Protocol      ┌─────────────┐
│   Client    │ ◄────────────────────► │   Server    │
│  (Model)    │   JSON-RPC 2.0        │  (Tools)    │
└─────────────┘                        └─────────────┘
        │                                      │
        ▼                                      ▼
   Function calls                        Tool execution
   Resource access                       Data sources
   Prompt templates                      Prompt templates

Core Concepts

1. Tools (Actions)

{
  "name": "search_web",
  "description": "Search the web for current information",
  "inputSchema": {
    "type": "object",
    "properties": {
      "query": {"type": "string"},
      "max_results": {"type": "integer", "default": 10}
    },
    "required": ["query"]
  }
}

2. Resources (Data)

{
  "uri": "file:///data/reports/quarterly.pdf",
  "name": "Q4 Report",
  "description": "Quarterly financial report",
  "mimeType": "application/pdf"
}

3. Prompts (Templates)

{
  "name": "code_review",
  "description": "Code review prompt template",
  "arguments": [
    {"name": "code", "description": "Code to review", "required": true},
    {"name": "language", "description": "Programming language", "required": false}
  ],
  "template": "Review this {{language}} code:\n{{code}}"
}

Protocol Specification

Transport

• stdio (local processes)
• HTTP/SSE (remote servers)
• WebSocket (bidirectional)

JSON-RPC 2.0 Methods

Initialize Handshake

// Client → Server
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {
    "protocolVersion": "2024-11-05",
    "capabilities": {
      "tools": {},
      "resources": {},
      "prompts": {}
    },
    "clientInfo": {"name": "my-agent", "version": "1.0.0"}
  }
}

// Server → Client
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "protocolVersion": "2024-11-05",
    "capabilities": {
      "tools": {"listChanged": true},
      "resources": {"subscribe": true, "listChanged": true},
      "prompts": {"listChanged": true}
    },
    "serverInfo": {"name": "my-tools", "version": "1.0.0"}
  }
}

Tool Call

// Client → Server
{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "tools/call",
  "params": {
    "name": "search_web",
    "arguments": {"query": "MCP protocol", "max_results": 5}
  }
}

// Server → Client
{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "content": [
      {"type": "text", "text": "Search results..."}
    ],
    "isError": false
  }
}

Server Implementation

Python SDK

from mcp import Server, Tool, Resource, Prompt
from mcp.types import TextContent

server = Server("my-tools")

@server.tool()
async def search_web(query: str, max_results: int = 10) -> list[TextContent]:
    """Search the web for current information."""
    results = await web_search(query, max_results)
    return [TextContent(type="text", text=str(r)) for r in results]

@server.resource("file:///data/{path}")
async def read_file(path: str) -> TextContent:
    """Read a file from the data directory."""
    content = await read_file_safe(f"/data/{path}")
    return TextContent(type="text", text=content)

@server.prompt()
async def code_review(code: str, language: str = "python") -> str:
    return f"Review this {language} code:\n{code}"

# Run
if __name__ == "__main__":
    server.run_stdio()

HTTP/SSE Server

from mcp import Server
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Route, Mount

server = Server("my-tools")
# ... define tools ...

sse = SseServerTransport("/messages")

async def handle_sse(request):
    async with sse.connect_sse(request.scope, request.receive, request.send) as (read, write):
        await server.run(read, write)

app = Starlette(routes=[
    Route("/sse", handle_sse),
    Mount("/messages", app=sse.handle_post_message)
])

Client Implementation

Python Client

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    # Connect to stdio server
    server_params = StdioServerParameters(
        command="python",
        args=["-m", "my_mcp_server"]
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize
            await session.initialize()

            # List tools
            tools = await session.list_tools()
            print(f"Available tools: {[t.name for t in tools.tools]}")

            # Call tool
            result = await session.call_tool("search_web", {"query": "MCP protocol"})
            print(result.content)

            # List resources
            resources = await session.list_resources()

            # Read resource
            resource = await session.read_resource("file:///data/config.json")

            # Get prompt
            prompt = await session.get_prompt("code_review", {"code": "def foo(): pass"})

if __name__ == "__main__":
    asyncio.run(main())

HTTP/SSE Client

from mcp.client.sse import sse_client

async with sse_client("http://localhost:8000/sse") as (read, write):
    async with ClientSession(read, write) as session:
        await session.initialize()
        # ... same as above

Capability Discovery

Dynamic Tool Loading

class MCPToolManager:
    def __init__(self):
        self.servers: dict[str, ClientSession] = {}
        self.tools_cache: dict[str, Tool] = {}

    async def connect_server(self, name: str, transport: Transport):
        read, write = await transport.connect()
        session = ClientSession(read, write)
        await session.initialize()
        self.servers[name] = session
        await self._refresh_tools()

    async def _refresh_tools(self):
        self.tools_cache.clear()
        for name, session in self.servers.items():
            tools = await session.list_tools()
            for tool in tools.tools:
                self.tools_cache[f"{name}.{tool.name}"] = tool

    def get_tool(self, qualified_name: str) -> Tool | None:
        return self.tools_cache.get(qualified_name)

    async def call_tool(self, qualified_name: str, args: dict) -> Any:
        server_name, tool_name = qualified_name.split(".", 1)
        session = self.servers[server_name]
        return await session.call_tool(tool_name, args)

Authentication & Security

Transport Security

# TLS for HTTP/SSE
ssl_context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
ssl_context.load_verify_locations("ca.pem")

async with sse_client(url, ssl=ssl_context) as (read, write):
    ...

Authentication

# Bearer token in headers
headers = {"Authorization": "Bearer <token>"}
async with sse_client(url, headers=headers) as (read, write):
    ...

# mTLS
ssl_context = ssl.create_default_context()
ssl_context.load_cert_chain("client.pem", "client.key")
ssl_context.load_verify_locations("ca.pem")

Tool-Level Authorization

# Server-side
@server.tool()
async def sensitive_tool(user: User, action: str) -> TextContent:
    if not user.has_permission("admin"):
        raise PermissionError("Admin required")
    ...

Integration with Frameworks

LangGraph

from langgraph.prebuilt import create_react_agent
from langchain_mcp_adapters import MCPToolkit

async def create_agent():
    toolkit = MCPToolkit.from_server("stdio", command="python", args=["-m", "my_mcp"])
    tools = await toolkit.get_tools()

    agent = create_react_agent(model, tools)
    return agent

PydanticAI

from pydantic_ai import Agent
from pydantic_ai.mcp import MCPClient

client = MCPClient("http://localhost:8000/mcp")
tools = await client.list_tools()

agent = Agent("openai:gpt-4o", tools=tools)

OpenAI Agents SDK

from agents.mcp import MCPClient

client = MCPClient("http://localhost:8000/mcp")
tools = await client.list_tools()

agent = Agent(tools=tools)

Tool Composition

Proxy/Router

class MCPRouter:
    def __init__(self):
        self.toolkit = MCPToolkit()

    async def add_server(self, name: str, transport: Transport):
        await self.toolkit.add_server(name, transport)

    def get_tools(self) -> list[Tool]:
        return self.toolkit.get_all_tools()

    # Namespace tools: "github.create_issue", "slack.send_message"
    def get_namespaced_tools(self) -> dict[str, Tool]:
        return {f"{ns}.{t.name}": t for ns, tools in self.toolkit.tools_by_namespace.items() for t in tools}

Tool Filtering

def filter_tools(tools: list[Tool], 
                 allowed_namespaces: list[str] = None,
                 allowed_names: list[str] = None,
                 blocked_names: list[str] = None) -> list[Tool]:
    filtered = tools
    if allowed_namespaces:
        filtered = [t for t in filtered if t.namespace in allowed_namespaces]
    if allowed_names:
        filtered = [t for t in filtered if t.name in allowed_names]
    if blocked_names:
        filtered = [t for t in filtered if t.name not in blocked_names]
    return filtered

Testing

Mock Server

from mcp.testing import MockServer

@pytest.fixture
async def mcp_server():
    server = MockServer()
    server.add_tool("search_web", lambda q: [{"title": "Test", "url": "http://test.com"}])
    server.add_resource("file:///config.json", '{"key": "value"}')
    await server.start()
    yield server
    await server.stop()

Integration Testing

async def test_mcp_integration(mcp_server):
    async with stdio_client(mcp_server.params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            tools = await session.list_tools()
            assert "search_web" in [t.name for t in tools.tools]

            result = await session.call_tool("search_web", {"query": "test"})
            assert len(result.content) > 0

Related Notes

• Function Calling — Function calling basics
• Code Execution — Code execution via MCP
• Browser Automation — Browser automation via MCP
• API Integration — API integration via MCP
• Custom Tools — Building MCP servers
• Tool Discovery — Dynamic discovery
• OpenAI Agents SDK — MCP in OpenAI SDK
• LangGraph — MCP in LangGraph
• PydanticAI — MCP in PydanticAI
• LangGraph — MCP in LangGraph
• AgentKit — MCP in AgentKit
• Security — MCP security
• Observability — MCP observability

Sources: MCP Spec (modelcontextprotocol.io), MCP Python SDK, MCP TypeScript SDK, Anthropic MCP announcement, LangChain MCP adapters, PydanticAI MCP, OpenAI Agents SDK MCP, community MCP servers (GitHub, Slack, Notion, PostgreSQL, etc.).