# Understanding and Building MCP Protocol Servers for AI Agents: A Complete Guide

> Learn to build MCP protocol servers for AI agents using FastMCP. Discover how to enable AI agents to invoke remote tools as services with this comprehensive guide.

- Repository: [Bojie Li/ai-agent-book](https://github.com/bojieli/ai-agent-book)
- Tags: how-to-guide
- Published: 2026-08-18

---

**The Model Context Protocol (MCP) is a lightweight JSON-over-HTTP framework that enables AI agents to discover and invoke remote tools as services, with servers implemented using the `FastMCP` class and tools defined as subclasses of `MCPToolBase`.**

MCP protocol servers bridge the gap between large language models and external capabilities. The `ai-agent-book` repository by bojieli provides a production-ready implementation of this pattern, showing how to wrap APIs like Wikipedia, PubChem, and YouTube into standardized services that any agent can consume. This guide examines the architecture, key source files, and practical implementation patterns from the codebase.

---

## What Is the MCP Protocol?

The **Model Context Protocol (MCP)** defines a contract between AI agents and tool services. Unlike ad-hoc API integrations, MCP enforces consistent request/response schemas that let agents dynamically discover and call capabilities without hardcoded logic.

The protocol operates on three principles:

- **Standardized schemas** — Every tool declares its input arguments and output response using regex-compatible raw string patterns
- **Environment-driven configuration** — Server URLs, authentication tokens, and enabled services load from environment variables
- **Decoupled deployment** — MCP servers run as independent HTTP services that multiple agents can share

---

## MCP Architecture in ai-agent-book

The repository organizes MCP functionality under `chapter9/gaia-experience/AWorld/examples/gaia/mcp_collections/`. The architecture follows a clear separation between server infrastructure, base abstractions, and concrete tool implementations.

### Core Components

| Component | Responsibility | Source Location |
|-----------|--------------|---------------|
| **FastMCP server** | HTTP request parsing, routing, and response serialization | [[`mcp_collections/base.py`](https://github.com/bojieli/ai-agent-book/blob/main/mcp_collections/base.py)](https://github.com/bojieli/ai-agent-book/blob/main/chapter9/gaia-experience/AWorld/examples/gaia/mcp_collections/base.py) |
| **MCPToolBase** | Abstract class defining argument/response schemas and execution interface | [[`mcp_collections/base.py`](https://github.com/bojieli/ai-agent-book/blob/main/mcp_collections/base.py)](https://github.com/bojieli/ai-agent-book/blob/main/chapter9/gaia-experience/AWorld/examples/gaia/mcp_collections/base.py) |
| **Tool implementations** | Domain-specific logic for external APIs (Wikipedia, PubChem, etc.) | [`mcp_collections/tools/*.py`](https://github.com/bojieli/ai-agent-book/blob/main/chapter9/gaia-experience/AWorld/examples/gaia/mcp_collections/tools/) |
| **Server bootstrap** | Registration and startup sequence | [[`gaia-experience/run_with_experience.py`](https://github.com/bojieli/ai-agent-book/blob/main/gaia-experience/run_with_experience.py)](https://github.com/bojieli/ai-agent-book/blob/main/chapter9/gaia-experience/gaia-experience/run_with_experience.py) |
| **Agent client** | HTTP client wrapper for remote MCP calls | [[`gaia_agent_runner.py`](https://github.com/bojieli/ai-agent-book/blob/main/gaia_agent_runner.py)](https://github.com/bojieli/ai-agent-book/blob/main/chapter9/gaia-experience/AWorld/examples/gaia/gaia_agent_runner.py) |

### Request/Response Contract

Every MCP tool adheres to two raw-string schema definitions as implemented in [[`base.py`](https://github.com/bojieli/ai-agent-book/blob/main/base.py)](https://github.com/bojieli/ai-agent-book/blob/main/chapter9/gaia-experience/AWorld/examples/gaia/mcp_collections/base.py):

```text
r"""Protocol: MCP Action Arguments"""

# Format: {"action": "tool_name", "arg1": value, "arg2": value, ...}

r"""Protocol: MCP Action Response"""

# Format: {"status": "ok|error", "output": "result string", "metadata": {...}}

```

This contract allows agents to construct valid requests programmatically and parse responses without tool-specific parsing logic.

---

## Building an MCP Server: Step-by-Step

### Step 1: Define a Tool Class

Subclass `MCPToolBase` and implement the `run()` coroutine. The [[`wiki.py`](https://github.com/bojieli/ai-agent-book/blob/main/wiki.py)](https://github.com/bojieli/ai-agent-book/blob/main/chapter9/gaia-experience/AWorld/examples/gaia/mcp_collections/tools/wiki.py) implementation demonstrates this pattern:

```python

# mcp_collections/tools/wiki.py

from .base import MCPToolBase, FastMCP
import httpx


class WikiTool(MCPToolBase):
    """MCP service for Wikipedia searches."""
    name = "wiki_search"
    
    # Schema definition for agent discovery

    args_schema = r'{"action": "wiki_search", "query": "string", "limit": "int"}'
    response_schema = r'{"status": "ok|error", "output": "string", "results": [{"title": "string", "url": "string"}]}'
    
    async def run(self, query: str, limit: int = 5) -> dict:
        """Execute Wikipedia search and return formatted results."""
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                "https://en.wikipedia.org/w/api.php",
                params={
                    "action": "query",
                    "list": "search",
                    "srsearch": query,
                    "srlimit": limit,
                    "format": "json"
                }
            )
            data = resp.json()
            
            results = [
                {"title": item["title"], "url": f"https://en.wikipedia.org/wiki/{item['title'].replace(' ', '_'}"}
                for item in data["query"]["search"]
            ]
            
            return {
                "status": "ok",
                "output": f"Found {len(results)} results for '{query}'",
                "results": results
            }

```

Key implementation requirements from the [[`base.py`](https://github.com/bojieli/ai-agent-book/blob/main/base.py)](https://github.com/bojieli/ai-agent-book/blob/main/chapter9/gaia-experience/AWorld/examples/gaia/mcp_collections/base.py) contract:

- Declare `name` as a class attribute — this becomes the action identifier in requests
- Implement `run()` as an async method accepting typed arguments
- Return a dictionary matching the response schema with `status` and `output` fields

### Step 2: Register and Launch the Server

The bootstrap sequence in [[`run_with_experience.py`](https://github.com/bojieli/ai-agent-book/blob/main/run_with_experience.py)](https://github.com/bojieli/ai-agent-book/blob/main/chapter9/gaia-experience/gaia-experience/run_with_experience.py) shows proper server initialization:

```python

# launch_server.py

import os
import asyncio
from mcp_collections.base import FastMCP
from mcp_collections.tools.wiki import WikiTool
from mcp_collections.tools.pubchem import PubChemTool
from mcp_collections.tools.youtube import YouTubeTool


async def main():
    # Initialize server with service name

    server = FastMCP("gaia-mcp-server")
    
    # Register all available tools

    server.register(WikiTool)
    server.register(PubChemTool)
    server.register(YouTubeTool)
    
    # Start HTTP listener with configuration from environment

    host = os.getenv("MCP_HOST", "0.0.0.0")
    port = int(os.getenv("MCP_PORT", 8000))
    
    await server.start(host=host, port=port)


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

```

**Environment variables controlling server behavior:**

| Variable | Purpose | Default |
|----------|---------|---------|
| `MCP_HOST` | Bind address | `0.0.0.0` |
| `MCP_PORT` | Listen port | `8000` |
| `MCP_SERVER_TOKEN` | Authentication token (required) | None |
| `MCP_LOG_LEVEL` | Server logging verbosity | `INFO` |

### Step 3: Consume from an Agent

The agent-side integration in [[`gaia_agent_runner.py`](https://github.com/bojieli/ai-agent-book/blob/main/gaia_agent_runner.py)](https://github.com/bojieli/ai-agent-book/blob/main/chapter9/gaia-experience/AWorld/examples/gaia/gaia_agent_runner.py) demonstrates client construction:

```python

# agent_client.py

import os
import json
import httpx
from typing import Dict, Any, List


class MCPClient:
    """Client for invoking remote MCP protocol servers."""
    
    def __init__(self, server_url: str, token: str):
        self.server_url = server_url.rstrip("/")
        self.headers = {"Authorization": f"Bearer {token}"}
    
    async def call(self, action: str, **kwargs) -> Dict[str, Any]:
        """Execute an MCP action on the remote server."""
        payload = {"action": action, **kwargs}
        
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.server_url}/mcp/execute",
                json=payload,
                headers=self.headers,
                timeout=60.0
            )
            response.raise_for_status()
            return response.json()
    
    async def list_tools(self) -> List[str]:
        """Discover available tools from server metadata endpoint."""
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.server_url}/mcp/tools",
                headers=self.headers
            )
            response.raise_for_status()
            return response.json()["tools"]


# Load configuration from environment per best practice

async def create_mcp_client() -> MCPClient:
    server_url = os.environ["MCP_SERVER_URL"]
    token = os.environ["MCP_SERVER_TOKEN"]
    return MCPClient(server_url, token)


# Example: Agent reasoning loop invoking MCP tool

async def research_topic(query: str) -> str:
    client = await create_mcp_client()
    
    # Discover available tools

    tools = await client.list_tools()
    print(f"Available MCP tools: {tools}")
    
    # Invoke Wikipedia search

    result = await client.call(
        action="wiki_search",
        query=query,
        limit=3
    )
    
    if result["status"] == "ok":
        return f"Research complete: {result['output']}"
    else:
        return f"Error: {result.get('error', 'Unknown failure')}"

```

---

## Production Tool Implementations

The repository includes several reference implementations demonstrating different integration patterns:

### Wikipedia Search ([[`wiki.py`](https://github.com/bojieli/ai-agent-book/blob/main/wiki.py)](https://github.com/bojieli/ai-agent-book/blob/main/chapter9/gaia-experience/AWorld/examples/gaia/mcp_collections/tools/wiki.py))

- **Pattern**: REST API wrapper with result transformation
- **Key feature**: Converts Wikipedia's nested API response into flat, agent-friendly metadata

### PubChem Chemical Database ([[`pubchem.py`](https://github.com/bojieli/ai-agent-book/blob/main/pubchem.py)](https://github.com/bojieli/ai-agent-book/blob/main/chapter9/gaia-experience/AWorld/examples/gaia/mcp_collections/tools/pubchem.py))

- **Pattern**: Scientific data with binary handling
- **Key feature**: Supports chemical structure downloads (SDF, PNG) with base64 encoding for JSON transport

### YouTube Metadata ([[`youtube.py`](https://github.com/bojieli/ai-agent-book/blob/main/youtube.py)](https://github.com/bojieli/ai-agent-book/blob/main/chapter9/gaia-experience/AWorld/examples/gaia/mcp_collections/tools/youtube.py))

- **Pattern**: Media streaming with pagination
- **Key feature**: Extracts transcript data and thumbnail URLs without full video download

---

## Testing MCP Implementations

The repository provides test infrastructure for validating MCP servers:

| Test File | Purpose |
|-----------|---------|
| [[`tests/mcp/streamable_server.py`](https://github.com/bojieli/ai-agent-book/blob/main/tests/mcp/streamable_server.py)](https://github.com/bojieli/ai-agent-book/blob/main/chapter9/gaia-experience/AWorld/tests/mcp/streamable_server.py) | Mock server for integration testing with configurable responses |
| [[`tests/local/mcp_test.py`](https://github.com/bojieli/ai-agent-book/blob/main/tests/local/mcp_test.py)](https://github.com/bojieli/ai-agent-book/blob/main/chapter9/gaia-experience/AWorld/tests/local/mcp_test.py) | Unit tests for client/server contract validation |

Example test pattern:

```python

# tests/local/mcp_test.py

import pytest
from mcp_collections.base import FastMCP
from mcp_collections.tools.wiki import WikiTool


@pytest.mark.asyncio
async def test_wiki_tool_contract():
    """Verify WikiTool satisfies MCP response schema."""
    tool = WikiTool()
    result = await tool.run(query="Python programming", limit=2)
    
    # Schema validation

    assert "status" in result
    assert "output" in result
    assert result["status"] in ("ok", "error")
    
    if result["status"] == "ok":
        assert "results" in result
        assert len(result["results"]) <= 2

```

---

## Security and Deployment Configuration

The [[`run_with_experience.py`](https://github.com/bojieli/ai-agent-book/blob/main/run_with_experience.py)](https://github.com/bojieli/ai-agent-book/blob/main/chapter9/gaia-experience/gaia-experience/run_with_experience.py) implementation enforces security through environment-driven configuration:

```bash

# Required environment

export MCP_SERVER_URL="https://mcp.example.com"
export MCP_SERVER_TOKEN="sk-xxxxxxxxxxxxxxxx"
export MCP_SERVERS="wiki-server,pubchem-server,youtube-server"

# Optional tuning

export MCP_REQUEST_TIMEOUT=30
export MCP_MAX_CONCURRENT=10
export MCP_ENABLE_METRICS=true

```

**Security best practices from the codebase:**

- **Token validation** — Every request requires a valid `Authorization: Bearer` header
- **No secrets in code** — URLs and tokens load exclusively from environment
- **Input sanitization** — The `FastMCP` base class validates action names against registered tools before execution
- **Request timeouts** — Default 60-second ceiling prevents runaway tool executions

---

## Summary

- **MCP protocol servers** standardize how AI agents invoke external tools through a JSON-over-HTTP contract defined in [[`mcp_collections/base.py`](https://github.com/bojieli/ai-agent-book/blob/main/mcp_collections/base.py)](https://github.com/bojieli/ai-agent-book/blob/main/chapter9/gaia-experience/AWorld/examples/gaia/mcp_collections/base.py)
- **Tool implementation** requires subclassing `MCPToolBase`, defining `name`, `args_schema`, `response_schema`, and implementing the `run()` coroutine
- **Server bootstrap** uses `FastMCP` registration pattern shown in [[`run_with_experience.py`](https://github.com/bojieli/ai-agent-book/blob/main/run_with_experience.py)](https://github.com/bojieli/ai-agent-book/blob/main/chapter9/gaia-experience/gaia-experience/run_with_experience.py) with environment-variable configuration
- **Agent consumption** follows the client pattern from [[`gaia_agent_runner.py`](https://github.com/bojieli/ai-agent-book/blob/main/gaia_agent_runner.py)](https://github.com/bojieli/ai-agent-book/blob/main/chapter9/gaia-experience/AWorld/examples/gaia/gaia_agent_runner.py) using `Authorization` headers and structured JSON payloads
- **Production tools** in the repository demonstrate REST wrapping, binary data handling, and streaming media extraction

---

## Frequently Asked Questions

### What is the difference between MCP and traditional REST API integration?

MCP enforces a standardized request/response contract with discoverable schemas, while traditional integrations require custom code per API. As implemented in [`ai-agent-book`](https://github.com/bojieli/ai-agent-book), an MCP server lets agents dynamically invoke tools by name without hardcoding endpoints or parsing logic. The `MCPToolBase` abstraction guarantees consistent error handling and response structure across all services.

### How do I add a new tool to an existing MCP server?

Create a Python file in [`mcp_collections/tools/`](https://github.com/bojieli/ai-agent-book/blob/main/chapter9/gaia-experience/AWorld/examples/gaia/mcp_collections/tools/) that subclasses `MCPToolBase`, implement the `run()` method with your business logic, and add `server.register(YourToolClass)` to the bootstrap script. No changes to agent code are required if the agent uses `list_tools()` for discovery.

### Can MCP servers handle authentication beyond bearer tokens?

The base implementation in [[`base.py`](https://github.com/bojieli/ai-agent-book/blob/main/base.py)](https://github.com/bojieli/ai-agent-book/blob/main/chapter9/gaia-experience/AWorld/examples/gaia/mcp_collections/base.py) supports custom middleware injection through `server.add_middleware()`. You can implement OAuth2, mTLS, or API key validation by subclassing `FastMCP` and overriding the `authenticate()` method before calling `super().start()`.

### How does the agent discover which MCP servers are available?

Per [[`gaia_agent_runner.py`](https://github.com/bojieli/ai-agent-book/blob/main/gaia_agent_runner.py)](https://github.com/bojieli/ai-agent-book/blob/main/chapter9/gaia-experience/AWorld/examples/gaia/gaia_agent_runner.py), the agent reads the `MCP_SERVERS` environment variable as a comma-separated list of server identifiers. For each identifier, it constructs a client using `MCP_{IDENTIFIER}_URL` and `MCP_{IDENTIFIER}_TOKEN` variables. The client then calls the `/mcp/tools` endpoint to enumerate available actions.