# 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?
| Problem | MCP Solution |
|---|---|
| Every framework has own tool format | Single standard |
| Hard to share tools across frameworks | Portable tool definitions |
| Vendor lock-in | Framework-agnostic |
| No discovery mechanism | Built-in capability discovery |
| No standard for resources/prompts | Unified interface |

### 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)
```json
{
  "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)
```json
{
  "uri": "file:///data/reports/quarterly.pdf",
  "name": "Q4 Report",
  "description": "Quarterly financial report",
  "mimeType": "application/pdf"
}
```

### 3. Prompts (Templates)
```json
{
  "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

| Method | Direction | Purpose |
|---|---|---|
| `initialize` | Client → Server | Handshake, capabilities |
| `tools/list` | Client → Server | Discover tools |
| `tools/call` | Client → Server | Invoke tool |
| `resources/list` | Client → Server | Discover resources |
| `resources/read` | Client → Server | Read resource |
| `resources/templates/list` | Client → Server | Resource templates |
| `prompts/list` | Client → Server | Discover prompts |
| `prompts/get` | Client → Server | Get prompt template |
| `ping` | Both | Heartbeat |
| `notifications/progress` | Server → Client | Progress updates |
| `notifications/resources/updated` | Server → Client | Resource changed |

### Initialize Handshake
```json
// 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
```json
// 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
```python
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
```python
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
```python
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
```python
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
```python
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
```python
# 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
```python
# 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
```python
# 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
```python
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
```python
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
```python
from agents.mcp import MCPClient

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

agent = Agent(tools=tools)
```

---

## Tool Composition

### Proxy/Router
```python
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
```python
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
```python
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
```python
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

- [[ToolUse/Function-Calling.md]] — Function calling basics
- [[ToolUse/Code-Execution.md]] — Code execution via MCP
- [[ToolUse/Browser-Automation.md]] — Browser automation via MCP
- [[ToolUse/API-Integration.md]] — API integration via MCP
- [[ToolUse/Custom-Tools.md]] — Building MCP servers
- [[ToolUse/Tool-Discovery.md]] — Dynamic discovery
- [[Frameworks/OpenAI-Agents-SDK.md]] — MCP in OpenAI SDK
- [[Frameworks/LangGraph.md]] — MCP in LangGraph
- [[Frameworks/PydanticAI.md]] — MCP in PydanticAI
- [[Frameworks/LangGraph.md]] — MCP in LangGraph
- [[Frameworks/AgentKit.md]] — MCP in AgentKit
- [[Production/Security.md]] — MCP security
- [[Production/Observability.md]] — 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.).*