# How to Implement MCP Protocol for AI Agent Tool Integration: A Complete Guide > Learn to implement the MCP protocol for AI agent tool integration. This guide covers using the mcp SDK Server class and registration decorators for seamless external tool calls and structured results. - Repository: [Bojie Li/ai-agent-book](https://github.com/bojieli/ai-agent-book) - Tags: how-to-guide - Published: 2026-08-01 --- **The Model‑Context‑Protocol (MCP) is a lightweight JSON‑RPC style protocol that enables AI models to call external tools and receive structured results using the `mcp` SDK's `Server` class with four registration decorators.** Implementing the MCP protocol lets your AI agents securely invoke browser automation, APIs, and custom services through a standardized interface. This guide walks through the implementation patterns found in `bojieli/ai-agent-book`, which uses MCP to expose browser-use capabilities to LLM agents. ## Core MCP Architecture Components ### The Server Foundation Every MCP implementation starts with the **Server** class from `mcp.server`. In [`browser_use/mcp/server.py`](https://github.com/bojieli/ai-agent-book/blob/main/browser_use/mcp/server.py), the `BrowserUseServer` initializes this core component: ```python class BrowserUseServer: def __init__(self): self.server = Server('browser-use') # Server identifier ``` This server handles JSON‑RPC request routing and manages typed responses across client connections. ### Four Required Handler Decorators MCP servers must register four handlers using decorators: | Decorator | Purpose | Return Type | |-----------|---------|-------------| | `@server.list_tools()` | Declare available tools | `list[types.Tool]` | | `@server.list_resources()` | Expose file/data resources | `list[types.Resource]` | | `@server.list_prompts()` | Provide prompt templates | `list[types.Prompt]` | | `@server.call_tool()` | Execute tool logic | `list[types.TextContent]` | In [`browser_use/mcp/server.py`](https://github.com/bojieli/ai-agent-book/blob/main/browser_use/mcp/server.py) lines 113‑179, the `list_tools` handler defines browser automation capabilities: ```python @self.server.list_tools() async def list_tools() -> list[types.Tool]: return [ types.Tool( name='navigate', description='Navigate to a URL', inputSchema={ 'type': 'object', 'properties': { 'url': {'type': 'string', 'description': 'Target URL'}, 'session_id': {'type': 'string'} }, 'required': ['url', 'session_id'] } ), # Additional tools: click, type, scroll, etc. ] ``` ### Typed Models for Schema Validation The **mcp SDK** provides `types` modules that enforce structured communication. Every tool declaration requires: - **name**: Unique identifier for the LLM to reference - **description**: Natural language explanation for model reasoning - **inputSchema**: JSON‑Schema defining argument types and required fields Returned content uses `types.TextContent` or `types.ImageContent` for multimedia responses. ## Implementing Tool Execution Logic ### The call_tool Handler Tool execution routes through the `@server.call_tool()` decorator. The `_execute_tool` method in [`browser_use/mcp/server.py`](https://github.com/bojieli/ai-agent-book/blob/main/browser_use/mcp/server.py) (lines 178‑225) demonstrates dispatch pattern: ```python @self.server.call_tool() async def call_tool(name: str, arguments: dict | None) -> list[types.TextContent]: return await self._execute_tool(name, arguments or {}) async def _execute_tool(self, name: str, arguments: dict) -> list[types.TextContent]: tool_mapping = { 'navigate': self._navigate, 'click': self._click, 'type': self._type_text, # ... additional tool mappings } handler = tool_mapping.get(name) if not handler: return [types.TextContent(type='text', text=f'Unknown tool: {name}')] result = await handler(**arguments) return [types.TextContent(type='text', text=result)] ``` ### Stateful Session Management Browser automation requires persistent state. The `BrowserUseServer` tracks active sessions in `self.active_sessions` and runs `_cleanup_expired_sessions` as a background task (lines 254‑274): ```python async def _cleanup_expired_sessions(self): """Remove idle sessions to prevent resource exhaustion.""" while True: await asyncio.sleep(SESSION_CLEANUP_INTERVAL) expired = [ sid for sid, session in self.active_sessions.items() if session.is_expired(SESSION_TIMEOUT) ] for sid in expired: await self.active_sessions[sid].close() del self.active_sessions[sid] ``` ## Step‑by‑Step MCP Server Implementation ### 1. Install Dependencies ```bash pip install mcp ``` ### 2. Create Server Class Structure ```python # my_mcp_server.py import asyncio from mcp.server import Server, NotificationOptions, InitializationOptions from mcp.server import stdio import mcp.types as types class MyToolServer: def __init__(self): self.server = Server('my-tools') self._setup_handlers() def _setup_handlers(self): """Register all four MCP handlers.""" self._register_list_tools() self._register_call_tool() # Optional: list_resources, list_prompts def _register_list_tools(self): @self.server.list_tools() async def list_tools() -> list[types.Tool]: return [ types.Tool( name='echo', description='Return the input text unchanged', inputSchema={ 'type': 'object', 'properties': { 'message': {'type': 'string'} }, 'required': ['message'] } ), types.Tool( name='calculate', description='Perform basic arithmetic', inputSchema={ 'type': 'object', 'properties': { 'operation': {'enum': ['add', 'subtract', 'multiply', 'divide']}, 'a': {'type': 'number'}, 'b': {'type': 'number'} }, 'required': ['operation', 'a', 'b'] } ) ] def _register_call_tool(self): @self.server.call_tool() async def call_tool(name: str, arguments: dict | None) -> list[types.TextContent]: args = arguments or {} if name == 'echo': return [types.TextContent(type='text', text=args.get('message', ''))] elif name == 'calculate': op, a, b = args['operation'], args['a'], args['b'] operations = { 'add': a + b, 'subtract': a - b, 'multiply': a * b, 'divide': a / b if b != 0 else 'Error: division by zero' } result = operations.get(op, 'Unknown operation') return [types.TextContent(type='text', text=str(result))] return [types.TextContent(type='text', text=f'Unknown tool: {name}')] ``` ### 3. Implement Async Entry Point ```python async def main(): server = MyToolServer() async with stdio.stdio_server() as (read_stream, write_stream): await server.server.run( read_stream, write_stream, InitializationOptions( server_name='my-tools', server_version='1.0.0', capabilities=server.server.get_capabilities( notification_options=NotificationOptions(), experimental_capabilities={} ) ) ) if __name__ == '__main__': asyncio.run(main()) ``` ### 4. Configure Client Connection Add your server to an MCP‑compatible client (Claude Desktop, OpenRouter, or custom agent): ```json { "mcpServers": { "my-tools": { "command": "python", "args": ["/path/to/my_mcp_server.py"], "env": {} } } } ``` ## Reference Implementation: browser-use MCP Server The production‑grade implementation in [`browser_use/mcp/server.py`](https://github.com/bojieli/ai-agent-book/blob/main/browser_use/mcp/server.py) demonstrates advanced patterns: - **Telemetry integration**: `self._telemetry.capture()` reports tool call timing and session events (lines 419‑426, 460‑470) - **Resource cleanup**: Background tasks prevent memory leaks from abandoned browser sessions - **Error handling**: Structured exceptions convert to `types.TextContent` with diagnostic information - **Schema richness**: Complex `inputSchema` definitions guide LLM reasoning about parameter relationships Example tool definitions span navigation, element interaction, screenshot capture, and session lifecycle management. ## Minimal Skeleton for Quick Start For rapid prototyping, refer to [`browser-use/examples/mcp/simple_server.py`](https://github.com/bojieli/ai-agent-book/blob/main/browser-use/examples/mcp/simple_server.py): ```python import asyncio from mcp.server import Server import mcp.types as types from mcp.server import stdio def create_server(): server = Server("simple-demo") @server.list_tools() async def list_tools() -> list[types.Tool]: return [ types.Tool( name="hello", description="Say hello to someone", inputSchema={ "type": "object", "properties": {"name": {"type": "string"}}, "required": ["name"] } ) ] @server.call_tool() async def call_tool(name: str, arguments: dict) -> list[types.TextContent]: if name == "hello": return [types.TextContent(type="text", text=f"Hello, {arguments['name']}!")] raise ValueError(f"Unknown tool: {name}") return server async def main(): server = create_server() async with stdio.stdio_server() as (r, w): await server.run(r, w) if __name__ == "__main__": asyncio.run(main()) ``` ## Telemetry and Observability Production MCP servers should emit structured events. The `BrowserUseServer` captures: - Server start/stop lifecycle events - Tool call latency and success rates - Session creation and expiration counts Implement similar patterns using your observability stack: ```python async def _execute_tool(self, name: str, arguments: dict) -> list[types.TextContent]: start = asyncio.get_event_loop().time() try: result = await self._dispatch_tool(name, arguments) self._telemetry.capture('tool.success', { 'tool': name, 'duration_ms': (asyncio.get_event_loop().time() - start) * 1000 }) return result except Exception as e: self._telemetry.capture('tool.error', {'tool': name, 'error': str(e)}) raise ``` ## Summary - **MCP protocol implementation** requires the `mcp` SDK's `Server` class and four decorator‑based handlers: `list_tools`, `list_resources`, `list_prompts`, and `call_tool` - **Type safety** comes from `mcp.types` models—`Tool`, `TextContent`, `Resource`, and `Prompt` enforce schema contracts - **Tool execution** maps string names to async methods via a dispatch dictionary in `call_tool` - **Stateful operations** need session tracking and background cleanup tasks to prevent resource leaks - **Production deployment** adds telemetry, structured error handling, and client configuration for stdio‑based communication - **Reference code** in [`browser_use/mcp/server.py`](https://github.com/bojieli/ai-agent-book/blob/main/browser_use/mcp/server.py) and [`examples/mcp/simple_server.py`](https://github.com/bojieli/ai-agent-book/blob/main/examples/mcp/simple_server.py) provide working patterns from the `bojieli/ai-agent-book` repository ## Frequently Asked Questions ### What transport does MCP use for communication? MCP standardizes on **stdio transport** for local tool servers. The `mcp.server.stdio.stdio_server()` context manager provides `read` and `write` streams that map to stdin/stdout, allowing any language runtime to host an MCP server without network configuration. Remote implementations can extend this with HTTP or WebSocket transports. ### How does an LLM discover available MCP tools? The LLM client calls the **`list_tools`** handler via JSON‑RPC and receives a structured manifest. Each `types.Tool` includes a `name` for invocation, `description` for the model's reasoning context, and `inputSchema` for argument validation. The model uses this metadata to construct valid tool calls without hardcoded integration. ### Can MCP servers maintain state between tool calls? **Yes**, but you must implement state management explicitly. The `BrowserUseServer` demonstrates this with `self.active_sessions`—a dictionary mapping session IDs to `BrowserSession` objects. The server validates session IDs on each call and runs `_cleanup_expired_sessions` to prevent memory exhaustion. Stateless tools can skip this pattern. ### What's the difference between tools, resources, and prompts in MCP? **Tools** execute actions and return results—like browser navigation or API calls. **Resources** expose readable data (files, database records) with URI-based addressing. **Prompts** provide reusable template strings for LLM context construction. Most AI agent integrations focus on tools, while resources and prompts support advanced retrieval and prompt engineering workflows.