Building MCP Servers

The Model Context Protocol (MCP) is the standard interface for connecting AI models to external tools, data sources, and services. An MCP server exposes three primitives: tools (functions the model can call), resources (data the model can read), and prompts (reusable prompt templates). This skill covers building production-quality MCP servers with FastMCP.

Core Mental Model

Think of an MCP server as a typed API contract between your service and any MCP-compatible client (Claude Desktop, Claude API, custom agents). Tools are for actions and computation; resources are for data access; prompts are for workflow templates. The key design principle: tools should be composable and narrowly scoped. A tool that does one thing well can be composed by the model into complex workflows. A monolithic "do everything" tool is hard to use reliably.

MCP Specification Fundamentals

MCP Architecture:
┌─────────────────┐     ┌──────────────────────┐     ┌──────────────────┐
│   MCP Client    │────▶│   MCP Server         │────▶│  External APIs   │
│ (Claude, Agent) │◀────│  (Your FastMCP app)  │◀────│  Databases, etc. │
└─────────────────┘     └──────────────────────┘     └──────────────────┘

Primitives:
  Tools     → Functions with JSON Schema input, called on demand
  Resources → URIs that expose readable content (text, JSON, binary)
  Prompts   → Named prompt templates with typed arguments

Complete FastMCP Server

# server.py — Production MCP server with tools, resources, and prompts
from fastmcp import FastMCP
from fastmcp.exceptions import ToolError
from pydantic import BaseModel, Field
from typing import Optional
import httpx
import json

mcp = FastMCP(
    name="company-tools",
    version="1.0.0",
    description="Company database and API integration tools",
)

# ============================================================
# TOOLS — Functions the model can invoke
# ============================================================

class CustomerLookupInput(BaseModel):
    customer_id: str = Field(description="Customer ID (e.g., 'cust_abc123')")
    include_history: bool = Field(default=False, description="Include purchase history")

@mcp.tool()
async def lookup_customer(params: CustomerLookupInput) -> dict:
    """
    Look up a customer by ID. Returns profile, contact info, and optionally purchase history.
    Use this when the user asks about a specific customer or needs to verify customer details.
    """
    # Input validation — fail fast with clear error
    if not params.customer_id.startswith("cust_"):
        raise ToolError(
            f"Invalid customer ID format: '{params.customer_id}'. Must start with 'cust_'"
        )
    
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(
                f"https://api.company.com/customers/{params.customer_id}",
                headers={"Authorization": f"Bearer {get_api_key()}"},
                timeout=10.0,
            )
            response.raise_for_status()
        except httpx.TimeoutException:
            raise ToolError("Customer API timed out. Try again in a moment.")
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 404:
                raise ToolError(f"Customer '{params.customer_id}' not found.")
            raise ToolError(f"API error: {e.response.status_code}")
    
    data = response.json()
    
    if params.include_history:
        # Parallel fetch for efficiency
        hist_response = await client.get(
            f"https://api.company.com/customers/{params.customer_id}/orders",
            headers={"Authorization": f"Bearer {get_api_key()}"},
        )
        data["order_history"] = hist_response.json()
    
    return data

@mcp.tool()
async def search_knowledge_base(
    query: str = Field(description="Search query"),
    max_results: int = Field(default=5, ge=1, le=20, description="Max results to return"),
    category: Optional[str] = Field(default=None, description="Filter by category"),
) -> list[dict]:
    """
    Search the internal knowledge base. Returns ranked articles with title, snippet, and URL.
    Use when answering questions about products, policies, or procedures.
    """
    results = await kb_client.search(query, limit=max_results, category=category)
    return [
        {
            "title": r.title,
            "snippet": r.snippet[:500],  # Limit snippet length
            "url": r.url,
            "relevance_score": r.score,
        }
        for r in results
    ]

@mcp.tool()
async def create_support_ticket(
    customer_id: str = Field(description="Customer ID"),
    subject: str = Field(description="Ticket subject (max 100 chars)"),
    description: str = Field(description="Detailed problem description"),
    priority: str = Field(description="Priority: low, medium, high, urgent"),
) -> dict:
    """
    Create a support ticket. Returns ticket ID and estimated response time.
    Use when a customer issue requires follow-up action beyond immediate resolution.
    """
    if priority not in ("low", "medium", "high", "urgent"):
        raise ToolError(f"Invalid priority '{priority}'. Must be: low, medium, high, urgent")
    
    if len(subject) > 100:
        raise ToolError(f"Subject too long ({len(subject)} chars). Max 100 characters.")
    
    # Sanitize description — strip potentially dangerous content
    clean_description = description[:5000]  # Hard limit
    
    ticket = await ticketing_system.create({
        "customer_id": customer_id,
        "subject": subject,
        "description": clean_description,
        "priority": priority,
    })
    
    return {
        "ticket_id": ticket.id,
        "status": "created",
        "estimated_response": ticket.sla_deadline.isoformat(),
    }

# ============================================================
# RESOURCES — Data sources the model can read
# ============================================================

@mcp.resource("company://docs/{doc_id}")
async def get_document(doc_id: str) -> str:
    """Retrieve a company document by ID. Returns markdown content."""
    doc = await doc_store.get(doc_id)
    if not doc:
        raise ValueError(f"Document '{doc_id}' not found")
    return doc.content_markdown

@mcp.resource("company://customers/{customer_id}/profile")
async def get_customer_profile_resource(customer_id: str) -> str:
    """Customer profile as formatted text for context injection."""
    customer = await db.get_customer(customer_id)
    return f"""
Customer: {customer.name}
ID: {customer.id}
Email: {customer.email}
Tier: {customer.subscription_tier}
Account Age: {customer.account_age_days} days
Total Orders: {customer.total_orders}
Lifetime Value: ${customer.lifetime_value:.2f}
"""

@mcp.resource("company://metrics/dashboard")
async def get_metrics_dashboard() -> str:
    """Current business metrics dashboard. Updates every 5 minutes."""
    metrics = await metrics_client.get_current()
    return json.dumps(metrics, indent=2)

# ============================================================
# PROMPTS — Reusable prompt templates
# ============================================================

@mcp.prompt()
def customer_escalation_prompt(
    customer_id: str,
    issue_summary: str,
    previous_attempts: str,
) -> list[dict]:
    """Template for escalating a complex customer issue to a specialist."""
    return [
        {
            "role": "user",
            "content": f"""
Please handle this escalated customer issue requiring specialist attention.

Customer ID: {customer_id}
Issue Summary: {issue_summary}
Previous Resolution Attempts: {previous_attempts}

Please:
1. Look up the customer profile
2. Review relevant knowledge base articles
3. Propose a resolution
4. Create a support ticket if needed
""",
        }
    ]

# ============================================================
# SERVER TRANSPORT
# ============================================================

if __name__ == "__main__":
    # stdio for local tools (Claude Desktop, CLI agents)
    mcp.run(transport="stdio")

# For remote/HTTP deployment:
# mcp.run(transport="sse", host="0.0.0.0", port=8080)

Server Transports

# stdio: local process communication — simplest, for Claude Desktop integration
mcp.run(transport="stdio")

# SSE (Server-Sent Events): HTTP-based, supports remote connections
# Client connects via HTTP, server streams events
mcp.run(
    transport="sse",
    host="0.0.0.0",
    port=8080,
    # Configure CORS for browser clients
)

# Streamable HTTP: newer standard (MCP 2025-03-26+)
mcp.run(transport="streamable-http", port=8080)

Claude Desktop Integration (stdio)

// ~/Library/Application Support/Claude/claude_desktop_config.json
{
  "mcpServers": {
    "company-tools": {
      "command": "python",
      "args": ["/path/to/server.py"],
      "env": {
        "COMPANY_API_KEY": "your-key-here",
        "DATABASE_URL": "postgresql://..."
      }
    }
  }
}

Authentication for Remote Servers

from fastmcp import FastMCP
from fastapi import HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt

security = HTTPBearer()

def verify_token(credentials: HTTPAuthorizationCredentials = Security(security)):
    """Validate JWT bearer token on all MCP requests."""
    try:
        payload = jwt.decode(
            credentials.credentials,
            key=os.environ["JWT_SECRET"],
            algorithms=["HS256"],
        )
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(401, "Token expired")
    except jwt.InvalidTokenError:
        raise HTTPException(401, "Invalid token")

# FastMCP with auth middleware
mcp = FastMCP(name="authenticated-server")
# Add middleware to the underlying FastAPI app
mcp.app.add_middleware(AuthMiddleware, verify_fn=verify_token)

Error Handling Patterns

from fastmcp.exceptions import ToolError
from enum import Enum

class ErrorCode(str, Enum):
    NOT_FOUND = "NOT_FOUND"
    INVALID_INPUT = "INVALID_INPUT"
    RATE_LIMITED = "RATE_LIMITED"
    UPSTREAM_ERROR = "UPSTREAM_ERROR"
    PERMISSION_DENIED = "PERMISSION_DENIED"

def raise_tool_error(code: ErrorCode, message: str) -> None:
    """Standardized error format — models can parse and respond appropriately."""
    raise ToolError(json.dumps({
        "error_code": code.value,
        "message": message,
        "recoverable": code in (ErrorCode.RATE_LIMITED, ErrorCode.UPSTREAM_ERROR),
    }))

@mcp.tool()
async def sensitive_operation(resource_id: str, user_context: dict) -> dict:
    """Example with comprehensive error handling."""
    # Authorization check first
    if not await check_permission(user_context["user_id"], "write", resource_id):
        raise_tool_error(ErrorCode.PERMISSION_DENIED, 
                         f"User lacks write permission for resource {resource_id}")
    
    # Resource exists?
    resource = await db.get(resource_id)
    if not resource:
        raise_tool_error(ErrorCode.NOT_FOUND, f"Resource '{resource_id}' not found")
    
    try:
        result = await upstream_api.operate(resource_id)
        return result
    except RateLimitError:
        raise_tool_error(ErrorCode.RATE_LIMITED, 
                         "Rate limit hit. Wait 60 seconds before retrying.")
    except UpstreamServiceError as e:
        raise_tool_error(ErrorCode.UPSTREAM_ERROR, 
                         f"Upstream service unavailable: {str(e)}")

Testing MCP Servers

# Unit tests using FastMCP test client
import pytest
from fastmcp.testing import MCPTestClient

@pytest.fixture
def client():
    return MCPTestClient(mcp)

@pytest.mark.asyncio
async def test_lookup_customer_success(client):
    result = await client.call_tool("lookup_customer", {
        "customer_id": "cust_test123",
        "include_history": False,
    })
    assert result["id"] == "cust_test123"
    assert "email" in result

@pytest.mark.asyncio
async def test_lookup_customer_invalid_id(client):
    with pytest.raises(ToolError) as exc_info:
        await client.call_tool("lookup_customer", {"customer_id": "invalid"})
    assert "Invalid customer ID format" in str(exc_info.value)

@pytest.mark.asyncio
async def test_list_tools(client):
    tools = await client.list_tools()
    tool_names = [t.name for t in tools]
    assert "lookup_customer" in tool_names
    assert "search_knowledge_base" in tool_names

# Integration test with MCP Inspector CLI:
# npx @modelcontextprotocol/inspector python server.py

Security Considerations

# 1. Input validation — always sanitize before passing to external systems
import re

def validate_customer_id(customer_id: str) -> str:
    """Strict allowlist validation — reject anything that doesn't match."""
    if not re.match(r'^cust_[a-zA-Z0-9]{8,32}
Anti-Patterns
❌ Monolithic tools that do everything  
A tool named manage_customer that handles lookup, update, and deletion is hard for models to use correctly. Split into get_customer, update_customer, delete_customer.
❌ Poor tool descriptions  
The description IS the model's documentation. Vague descriptions lead to incorrect tool use. Include: what it does, when to use it, what parameters mean, what it returns.
❌ Letting exceptions propagate raw  
Python exceptions become opaque error messages. Always catch and convert to ToolError with human-readable context.
❌ No input validation  
Models make mistakes. Always validate inputs server-side — don't trust that the model will pass correct types or formats.
❌ Mutable side effects without confirmation  
For destructive operations (delete, send email, charge card), consider requiring an explicit confirm=True parameter so models can't accidentally execute them.
Quick Reference
Tool description template:
  "[Action verb] [what it operates on]. Returns [what]. 
   Use when [trigger condition]. 
   [Important constraints or caveats]."

Transport selection:
  Local (Claude Desktop, CLI)  → stdio
  Remote (HTTP clients)        → SSE or streamable-http
  High-throughput production   → streamable-http + load balancer

Error types:
  ToolError  → Expected failure (not found, validation failed, permission denied)
  Exception  → Unexpected failure (bug) — FastMCP converts to internal error

Validation checklist:
  ☐ Allowlist validation on IDs (regex or enum)
  ☐ Length limits on string inputs
  ☐ Range limits on numeric inputs
  ☐ Secrets redacted from output
  ☐ Rate limiting on expensive/dangerous tools
  ☐ Authorization check before any data access
, customer_id):
        raise ToolError("Invalid customer ID format")
    return customer_id

# 2. Scope limiting — tools should only access what they need
# BAD: Generic database tool with raw SQL access
@mcp.tool()
async def run_sql(query: str) -> list:  # DANGEROUS — full DB access
    return await db.execute(query)

# GOOD: Specific, scoped read-only queries
@mcp.tool()  
async def get_customer_orders(customer_id: str, limit: int = 10) -> list:
    """Returns up to 20 orders for a customer. Read-only."""
    return await db.query(
        "SELECT id, status, total FROM orders WHERE customer_id = $1 LIMIT $2",
        [validate_customer_id(customer_id), min(limit, 20)],  # Hard cap on limit
    )

# 3. Rate limiting per tool
from slowapi import Limiter
limiter = Limiter(key_func=get_client_id)

@mcp.tool()
@limiter.limit("10/minute")
async def expensive_operation(params: dict) -> dict:
    ...

# 4. Secrets never in tool output
@mcp.tool()
async def get_config() -> dict:
    config = await config_store.get_all()
    # Redact secrets before returning
    return {k: "***REDACTED***" if "secret" in k.lower() or "key" in k.lower() 
            else v for k, v in config.items()}

Anti-Patterns

❌ Monolithic tools that do everything
A tool named __INLINE_CODE_0__ that handles lookup, update, and deletion is hard for models to use correctly. Split into __INLINE_CODE_1__, __INLINE_CODE_2__, __INLINE_CODE_3__.

❌ Poor tool descriptions
The description IS the model's documentation. Vague descriptions lead to incorrect tool use. Include: what it does, when to use it, what parameters mean, what it returns.

❌ Letting exceptions propagate raw
Python exceptions become opaque error messages. Always catch and convert to __INLINE_CODE_4__ with human-readable context.

❌ No input validation
Models make mistakes. Always validate inputs server-side — don't trust that the model will pass correct types or formats.

❌ Mutable side effects without confirmation
For destructive operations (delete, send email, charge card), consider requiring an explicit __INLINE_CODE_5__ parameter so models can't accidentally execute them.

Quick Reference

__CODE_BLOCK_8__