Building MCP Servers

Understanding MCP theory is step one. Building your own MCP Server is where you unlock the real power — giving any AI agent access to your APIs, databases, business logic, and internal tools through a standardized interface.

In this chapter you'll build three complete MCP Servers, progressing from a basic local tool server to a production-ready remote HTTP server.

---

1. Setup & Project Structure

Install the Python MCP SDK

pip install mcp anthropic
# For HTTP transport (remote servers)
pip install "mcp[cli]" uvicorn

Recommended Project Layout

my-mcp-server/
├── server.py          # Main MCP server entry point
├── tools/
│   ├── __init__.py
│   ├── search.py      # Web search tools
│   └── database.py    # Database tools
├── resources/
│   ├── __init__.py
│   └── files.py       # File resource handlers
├── requirements.txt
└── README.md

---

2. Your First MCP Server — stdio (Local)

Let's build a fully functional weather + calculator MCP server that Claude Desktop can use.

# server.py
from mcp.server.models import InitializationOptions
from mcp.server import NotificationOptions, Server
from mcp.server.stdio import stdio_server
import mcp.types as types
import asyncio
import json

# Create the MCP server instance
server = Server("my-first-mcp-server")

# ─────────────────────────────────────────────
# TOOLS — Actions the AI model can call
# ─────────────────────────────────────────────

@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    """Advertise all available tools to the client."""
    return [
        types.Tool(
            name="get_weather",
            description="Get current weather for a city. Returns temperature, conditions, and humidity.",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "City name, e.g. 'London' or 'New York'"
                    },
                    "units": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "Temperature unit",
                        "default": "celsius"
                    }
                },
                "required": ["city"]
            }
        ),
        types.Tool(
            name="calculate",
            description="Evaluate a mathematical expression. Supports +, -, *, /, **, sqrt, etc.",
            inputSchema={
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "Math expression to evaluate, e.g. '2 ** 10' or 'sqrt(144)'"
                    }
                },
                "required": ["expression"]
            }
        ),
        types.Tool(
            name="convert_units",
            description="Convert between measurement units (length, weight, temperature).",
            inputSchema={
                "type": "object",
                "properties": {
                    "value": {"type": "number", "description": "Numeric value to convert"},
                    "from_unit": {"type": "string", "description": "Source unit (e.g. 'km', 'kg', 'celsius')"},
                    "to_unit": {"type": "string", "description": "Target unit (e.g. 'miles', 'lbs', 'fahrenheit')"}
                },
                "required": ["value", "from_unit", "to_unit"]
            }
        )
    ]


@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    """Execute a tool call and return results."""
    
    if name == "get_weather":
        city = arguments["city"]
        units = arguments.get("units", "celsius")
        
        # In production: call a real weather API (OpenWeatherMap, etc.)
        # Simulated response for demonstration
        weather_data = {
            "city": city,
            "temperature": 18 if units == "celsius" else 64,
            "units": units,
            "conditions": "Partly cloudy",
            "humidity": "65%",
            "wind": "12 km/h NW",
            "feels_like": 16 if units == "celsius" else 61
        }
        
        return [types.TextContent(
            type="text",
            text=json.dumps(weather_data, indent=2)
        )]
    
    elif name == "calculate":
        expression = arguments["expression"]
        try:
            # Safe evaluation — whitelist approach
            import math
            allowed_names = {k: v for k, v in math.__dict__.items() if not k.startswith("_")}
            result = eval(expression, {"__builtins__": {}}, allowed_names)
            return [types.TextContent(
                type="text",
                text=f"Result: {result}\nExpression: {expression}"
            )]
        except Exception as e:
            return [types.TextContent(type="text", text=f"Error: {str(e)}")]
    
    elif name == "convert_units":
        value = arguments["value"]
        from_unit = arguments["from_unit"].lower()
        to_unit = arguments["to_unit"].lower()
        
        # Conversion table
        conversions = {
            ("km", "miles"): lambda x: x * 0.621371,
            ("miles", "km"): lambda x: x * 1.60934,
            ("kg", "lbs"): lambda x: x * 2.20462,
            ("lbs", "kg"): lambda x: x / 2.20462,
            ("celsius", "fahrenheit"): lambda x: x * 9/5 + 32,
            ("fahrenheit", "celsius"): lambda x: (x - 32) * 5/9,
            ("meters", "feet"): lambda x: x * 3.28084,
            ("feet", "meters"): lambda x: x / 3.28084,
        }
        
        key = (from_unit, to_unit)
        if key in conversions:
            result = conversions[key](value)
            return [types.TextContent(
                type="text",
                text=f"{value} {from_unit} = {result:.4f} {to_unit}"
            )]
        else:
            return [types.TextContent(
                type="text",
                text=f"Conversion from '{from_unit}' to '{to_unit}' not supported."
            )]
    
    else:
        raise ValueError(f"Unknown tool: {name}")


# ─────────────────────────────────────────────
# Run the server
# ─────────────────────────────────────────────

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="my-first-mcp-server",
                server_version="1.0.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={}
                )
            )
        )

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

Register With Claude Desktop

{
  "mcpServers": {
    "my-tools": {
      "command": "python",
      "args": ["/absolute/path/to/server.py"]
    }
  }
}

---

3. Adding Resources — Expose Data to the AI

Resources let you expose read-only data — files, database records, API responses — that the AI can read as context.

# Add this to your server.py

@server.list_resources()
async def handle_list_resources() -> list[types.Resource]:
    """List all resources the server exposes."""
    return [
        types.Resource(
            uri="config://app/settings",
            name="Application Settings",
            description="Current application configuration",
            mimeType="application/json"
        ),
        types.Resource(
            uri="logs://app/recent",
            name="Recent Application Logs",
            description="Last 100 lines of application logs",
            mimeType="text/plain"
        ),
        types.Resource(
            uri="db://users/summary",
            name="User Statistics",
            description="Aggregated user statistics from the database",
            mimeType="application/json"
        )
    ]


@server.read_resource()
async def handle_read_resource(uri: str) -> str:
    """Return the content of a requested resource."""
    import json
    
    if uri == "config://app/settings":
        config = {
            "app_name": "MyApp",
            "version": "2.1.0",
            "environment": "production",
            "features": {
                "auth": True,
                "payments": True,
                "analytics": True
            },
            "rate_limits": {
                "requests_per_minute": 60,
                "max_file_size_mb": 10
            }
        }
        return json.dumps(config, indent=2)
    
    elif uri == "logs://app/recent":
        # In production: read from your actual log files
        return """2025-01-15 10:23:41 INFO  Server started on port 8080
2025-01-15 10:24:01 INFO  User 'alice@example.com' logged in
2025-01-15 10:24:15 WARN  Rate limit approaching for IP 192.168.1.100
2025-01-15 10:25:03 INFO  Payment processed: order #8821 ($49.99)
2025-01-15 10:25:41 ERROR Failed to send email to 'bob@example.com': SMTP timeout
2025-01-15 10:26:00 INFO  Background job 'daily_report' completed in 3.2s"""
    
    elif uri == "db://users/summary":
        stats = {
            "total_users": 15420,
            "active_today": 1832,
            "new_signups_today": 47,
            "premium_users": 3210,
            "churn_rate_30d": "2.3%"
        }
        return json.dumps(stats, indent=2)
    
    else:
        raise ValueError(f"Resource not found: {uri}")

---

4. Adding Prompt Templates

Prompts are reusable prompt templates users can invoke from the host application.

@server.list_prompts()
async def handle_list_prompts() -> list[types.Prompt]:
    """List all available prompt templates."""
    return [
        types.Prompt(
            name="debug-error",
            description="Analyze an error message and suggest fixes",
            arguments=[
                types.PromptArgument(
                    name="error_message",
                    description="The full error message or stack trace",
                    required=True
                ),
                types.PromptArgument(
                    name="language",
                    description="Programming language (python, javascript, etc.)",
                    required=False
                )
            ]
        ),
        types.Prompt(
            name="summarize-logs",
            description="Summarize application logs and highlight issues",
            arguments=[
                types.PromptArgument(
                    name="time_period",
                    description="Time period to analyze (e.g. 'last hour', 'today')",
                    required=False
                )
            ]
        )
    ]


@server.get_prompt()
async def handle_get_prompt(name: str, arguments: dict | None) -> types.GetPromptResult:
    """Return a prompt template filled with the provided arguments."""
    
    if name == "debug-error":
        error = arguments.get("error_message", "") if arguments else ""
        lang = arguments.get("language", "unknown") if arguments else "unknown"
        
        return types.GetPromptResult(
            description="Debug error analysis prompt",
            messages=[
                types.PromptMessage(
                    role="user",
                    content=types.TextContent(
                        type="text",
                        text=f"""Analyze this {lang} error and provide:
1. **Root cause** — What is causing this error?
2. **Fix** — Exact code change to resolve it
3. **Prevention** — How to avoid this in the future

Error Details:
{error}"""
                    )
                )
            ]
        )
    
    elif name == "summarize-logs":
        time_period = arguments.get("time_period", "recent") if arguments else "recent"
        
        return types.GetPromptResult(
            description="Log summarization prompt",
            messages=[
                types.PromptMessage(
                    role="user",
                    content=types.TextContent(
                        type="text",
                        text=f"""Analyze the {time_period} application logs and provide:
1. **Health Summary** — Overall system health (🟢 Good / 🟡 Warning / 🔴 Critical)
2. **Key Events** — Most important events in chronological order
3. **Errors & Warnings** — List all errors/warnings with suggested actions
4. **Performance Notes** — Any slow operations or resource issues
5. **Recommendations** — Top 3 things to fix or investigate"""
                    )
                )
            ]
        )
    
    else:
        raise ValueError(f"Prompt not found: {name}")

---

5. Production Server — HTTP + SSE Transport (Remote)

For remote deployment (cloud, shared access, multi-user), use HTTP + SSE transport:

# remote_server.py
from mcp.server.fastmcp import FastMCP
import httpx
import os

# FastMCP is a higher-level wrapper that makes HTTP servers easy
mcp = FastMCP("production-api-server")

# ─────────────────────────────────────────────
# Production-grade tool with error handling
# ─────────────────────────────────────────────

@mcp.tool()
async def fetch_github_repo(owner: str, repo: str) -> dict:
    """Fetch metadata about a GitHub repository.
    
    Args:
        owner: Repository owner (username or org)
        repo: Repository name
    
    Returns:
        Repository metadata including stars, forks, description
    """
    token = os.environ.get("GITHUB_TOKEN", "")
    headers = {"Authorization": f"Bearer {token}"} if token else {}
    
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://api.github.com/repos/{owner}/{repo}",
            headers=headers,
            timeout=10.0
        )
        response.raise_for_status()
        data = response.json()
        
        return {
            "name": data["name"],
            "description": data.get("description", ""),
            "stars": data["stargazers_count"],
            "forks": data["forks_count"],
            "language": data.get("language", "Unknown"),
            "open_issues": data["open_issues_count"],
            "license": data.get("license", {}).get("name", "None"),
            "url": data["html_url"],
            "created_at": data["created_at"],
            "updated_at": data["updated_at"]
        }


@mcp.tool()
async def send_notification(
    channel: str,
    message: str,
    priority: str = "normal"
) -> dict:
    """Send a notification to a Slack channel or email.
    
    Args:
        channel: Target channel (#general) or email address
        message: Notification message text
        priority: Message priority — 'low', 'normal', or 'urgent'
    """
    # In production: call actual Slack API or email service
    # Simulated for this example
    webhook_url = os.environ.get("SLACK_WEBHOOK_URL")
    
    emoji = {"low": "ℹ️", "normal": "📢", "urgent": "🚨"}.get(priority, "📢")
    
    if webhook_url:
        async with httpx.AsyncClient() as client:
            await client.post(webhook_url, json={
                "text": f"{emoji} *[{priority.upper()}]* {message}",
                "channel": channel
            })
    
    return {
        "status": "sent",
        "channel": channel,
        "priority": priority,
        "message_preview": message[:100] + ("..." if len(message) > 100 else "")
    }


@mcp.tool()
async def query_database(sql: str, read_only: bool = True) -> list[dict]:
    """Execute a SQL query against the production database.
    
    Args:
        sql: SQL query to execute (SELECT only if read_only=True)
        read_only: If True, only SELECT statements are allowed
    
    Returns:
        List of result rows as dictionaries
    """
    import asyncpg
    
    # Safety check for read-only mode
    if read_only and not sql.strip().upper().startswith("SELECT"):
        raise ValueError("Only SELECT queries allowed in read_only mode")
    
    conn = await asyncpg.connect(os.environ["DATABASE_URL"])
    try:
        rows = await conn.fetch(sql)
        return [dict(row) for row in rows]
    finally:
        await conn.close()


# ─────────────────────────────────────────────
# Resources with dynamic content
# ─────────────────────────────────────────────

@mcp.resource("metrics://current")
async def get_current_metrics() -> str:
    """Real-time application metrics."""
    import json, time
    metrics = {
        "timestamp": time.time(),
        "requests_per_second": 142,
        "p95_latency_ms": 87,
        "error_rate": "0.3%",
        "active_connections": 1204,
        "cache_hit_rate": "94.2%",
        "memory_usage_mb": 512,
        "cpu_percent": 23
    }
    return json.dumps(metrics, indent=2)


# ─────────────────────────────────────────────
# Run as HTTP server
# ─────────────────────────────────────────────

if __name__ == "__main__":
    # Starts an HTTP server with SSE transport
    # Default: http://localhost:8000/sse
    mcp.run(transport="sse")

Run the remote server:

python remote_server.py
# Server running at http://localhost:8000/sse

Connect from a client:

from mcp.client.sse import sse_client
from mcp import ClientSession

async def connect_to_remote():
    async with sse_client("http://localhost:8000/sse") as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            tools = await session.list_tools()
            print([t.name for t in tools.tools])

---

6. MCP Server with FastMCP Decorators (Recommended Pattern)

FastMCP provides clean Python-native decorators — the recommended approach for new servers:

from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel
from typing import Optional
import json

mcp = FastMCP("task-manager-server")

# In-memory task store (use a real DB in production)
tasks: dict[str, dict] = {}
task_counter = 0

class Task(BaseModel):
    title: str
    description: Optional[str] = None
    priority: str = "medium"  # low, medium, high
    status: str = "todo"      # todo, in-progress, done

@mcp.tool()
def create_task(title: str, description: str = "", priority: str = "medium") -> dict:
    """Create a new task.
    
    Args:
        title: Task title
        description: Detailed description  
        priority: Task priority (low/medium/high)
    """
    global task_counter
    task_counter += 1
    task_id = f"task-{task_counter}"
    
    tasks[task_id] = {
        "id": task_id,
        "title": title,
        "description": description,
        "priority": priority,
        "status": "todo"
    }
    return {"created": task_id, "task": tasks[task_id]}

@mcp.tool()
def list_tasks(status: str = "all", priority: str = "all") -> list[dict]:
    """List tasks, optionally filtered by status or priority.
    
    Args:
        status: Filter by status: 'todo', 'in-progress', 'done', or 'all'
        priority: Filter by priority: 'low', 'medium', 'high', or 'all'
    """
    result = list(tasks.values())
    if status != "all":
        result = [t for t in result if t["status"] == status]
    if priority != "all":
        result = [t for t in result if t["priority"] == priority]
    return result

@mcp.tool()
def update_task_status(task_id: str, status: str) -> dict:
    """Update a task's status.
    
    Args:
        task_id: The task ID (e.g. 'task-1')
        status: New status: 'todo', 'in-progress', or 'done'
    """
    if task_id not in tasks:
        raise ValueError(f"Task {task_id} not found")
    if status not in ("todo", "in-progress", "done"):
        raise ValueError(f"Invalid status: {status}")
    
    tasks[task_id]["status"] = status
    return {"updated": task_id, "new_status": status}

@mcp.tool()
def delete_task(task_id: str) -> dict:
    """Delete a task permanently.
    
    Args:
        task_id: The task ID to delete
    """
    if task_id not in tasks:
        raise ValueError(f"Task {task_id} not found")
    deleted = tasks.pop(task_id)
    return {"deleted": task_id, "title": deleted["title"]}

@mcp.resource("tasks://all")
def get_all_tasks_resource() -> str:
    """All tasks as a JSON resource."""
    return json.dumps(list(tasks.values()), indent=2)

@mcp.resource("tasks://stats")
def get_task_stats() -> str:
    """Task statistics summary."""
    all_tasks = list(tasks.values())
    stats = {
        "total": len(all_tasks),
        "by_status": {
            "todo": sum(1 for t in all_tasks if t["status"] == "todo"),
            "in_progress": sum(1 for t in all_tasks if t["status"] == "in-progress"),
            "done": sum(1 for t in all_tasks if t["status"] == "done"),
        },
        "by_priority": {
            "high": sum(1 for t in all_tasks if t["priority"] == "high"),
            "medium": sum(1 for t in all_tasks if t["priority"] == "medium"),
            "low": sum(1 for t in all_tasks if t["priority"] == "low"),
        }
    }
    return json.dumps(stats, indent=2)

if __name__ == "__main__":
    mcp.run()  # stdio by default

Claude interaction example:

User: Create 3 tasks: buy groceries (high priority), finish report (high), call dentist (low)

Claude: [calls create_task × 3]
✅ Created 3 tasks:
- task-1: "buy groceries" (HIGH)
- task-2: "finish report" (HIGH)  
- task-3: "call dentist" (LOW)

User: Mark "buy groceries" as done

Claude: [calls update_task_status]
✅ task-1 status updated to: done

User: What's still pending with high priority?

Claude: [calls list_tasks]
📋 High priority pending:
- task-2: "finish report" (in-progress)

---

7. Error Handling Best Practices

from mcp.server.fastmcp import FastMCP
from mcp.types import McpError, ErrorCode
import logging

mcp = FastMCP("robust-server")
logger = logging.getLogger(__name__)

@mcp.tool()
async def safe_api_call(endpoint: str, params: dict = {}) -> dict:
    """Call an external API with robust error handling."""
    import httpx
    
    try:
        async with httpx.AsyncClient(timeout=10.0) as client:
            response = await client.get(endpoint, params=params)
            response.raise_for_status()
            return response.json()
            
    except httpx.TimeoutException:
        # Return structured error — don't crash the server
        raise McpError(
            ErrorCode.InternalError,
            f"Request to {endpoint} timed out after 10 seconds"
        )
    except httpx.HTTPStatusError as e:
        raise McpError(
            ErrorCode.InternalError,
            f"HTTP {e.response.status_code} from {endpoint}: {e.response.text[:200]}"
        )
    except Exception as e:
        logger.error(f"Unexpected error calling {endpoint}: {e}")
        raise McpError(ErrorCode.InternalError, f"Unexpected error: {str(e)}")

---

8. Testing Your MCP Server

# test_server.py
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def test_mcp_server():
    """Integration test for our task manager server."""
    
    server_params = StdioServerParameters(
        command="python",
        args=["server.py"]  # Your server file
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # Test 1: List tools
            tools = await session.list_tools()
            tool_names = [t.name for t in tools.tools]
            print(f"✅ Tools available: {tool_names}")
            assert "create_task" in tool_names
            
            # Test 2: Create a task
            result = await session.call_tool("create_task", {
                "title": "Test Task",
                "priority": "high"
            })
            print(f"✅ Task created: {result.content[0].text}")
            
            # Test 3: List tasks
            result = await session.call_tool("list_tasks", {"status": "todo"})
            tasks = result.content[0].text
            print(f"✅ Tasks listed: {tasks}")
            
            # Test 4: Read a resource
            resource = await session.read_resource("tasks://stats")
            print(f"✅ Stats resource: {resource.contents[0].text}")
            
            print("\n✅ All tests passed!")

asyncio.run(test_mcp_server())

Run tests:

python test_server.py

# Output:
# ✅ Tools available: ['create_task', 'list_tasks', 'update_task_status', 'delete_task']
# ✅ Task created: {"created": "task-1", "task": {...}}
# ✅ Tasks listed: [{"id": "task-1", ...}]
# ✅ Stats resource: {"total": 1, ...}
# ✅ All tests passed!

---

9. Deploying to Production

Docker Container

FROM python:3.11-slim
WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
EXPOSE 8000

# Run with HTTP+SSE transport for remote access
CMD ["python", "-m", "uvicorn", "remote_server:mcp.get_asgi_app()", "--host", "0.0.0.0", "--port", "8000"]

Environment-based Configuration Pattern

from mcp.server.fastmcp import FastMCP
from pydantic_settings import BaseSettings

class ServerConfig(BaseSettings):
    """Load config from environment variables or .env file."""
    database_url: str
    github_token: str = ""
    slack_webhook: str = ""
    max_results: int = 10
    debug: bool = False
    
    class Config:
        env_file = ".env"

config = ServerConfig()
mcp = FastMCP("production-server")

@mcp.tool()
async def get_config_info() -> dict:
    """Return non-sensitive server configuration."""
    return {
        "max_results": config.max_results,
        "has_github": bool(config.github_token),
        "has_slack": bool(config.slack_webhook),
        "debug_mode": config.debug
    }

---

10. MCP Server Checklist

Quiz

Key Takeaways

✅ Use FastMCP decorators (@mcp.tool(), @mcp.resource(), @mcp.prompt()) for clean server code.
✅ stdio transport = local subprocess; HTTP + SSE = remote/cloud deployment.
✅ Tool description and inputSchema are critical — the AI reads them to make decisions.
✅ Always validate inputs, handle errors with McpError, and return structured JSON.
✅ Test your server using a real ClientSession — it tests the full MCP protocol.
✅ Never expose API keys or secrets in tool results — they end up in the model's context.

What's Next?

You've now mastered MCP from protocol theory to production server deployment. The natural next step is combining MCP Servers inside Multi-Agent Systems — where one orchestrator agent coordinates multiple specialized agents, each backed by their own MCP tools.

Keep building! 🚀