how-to-guide

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

August 18, 2026 bojieli/ai-agent-book ↗

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/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/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`
Server bootstrap	Registration and startup sequence	[`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/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/chapter9/gaia-experience/AWorld/examples/gaia/mcp_collections/base.py):

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/chapter9/gaia-experience/AWorld/examples/gaia/mcp_collections/tools/wiki.py) implementation demonstrates this pattern:


# 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/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/chapter9/gaia-experience/gaia-experience/run_with_experience.py) shows proper server initialization:


# 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/chapter9/gaia-experience/AWorld/examples/gaia/gaia_agent_runner.py) demonstrates client construction:


# 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/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/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/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/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/chapter9/gaia-experience/AWorld/tests/local/mcp_test.py)	Unit tests for client/server contract validation

Example test pattern:


# 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/chapter9/gaia-experience/gaia-experience/run_with_experience.py) implementation enforces security through environment-driven configuration:


# 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/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/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/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, 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/ 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/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/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.

Have a question about this repo?

These articles cover the highlights, but your codebase questions are specific. Give your agent direct access to the source. Share this with your agent to get started:

Share the following with your agent to get started:

curl -s "https://instagit.com/install.md"

Add to your MCP client configuration:

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

Ask your agent:

"Use Instagit MCP to understand how bojieli/ai-agent-book works."

Works with

Claude Codex Cursor VS Code OpenClaw Any MCP Client

Maintain an open-source project? Get it listed too →