How to Implement MCP Protocol for AI Agent Tool Integration: A Complete Guide

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, the BrowserUseServer initializes this core component:

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 lines 113‑179, the list_tools handler defines browser automation capabilities:

@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 (lines 178‑225) demonstrates dispatch pattern:

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

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

pip install mcp

2. Create Server Class Structure

# 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

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

{
  "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 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:

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:

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

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}