Building MCP Servers

Now that you've built individual tools, it's time to package them into production-ready MCP servers. This lesson covers architecture, security, deployment, and best practices.

MCP Server Architecture

┌──────────────────────────────────────────────────────────────────┐
│                     MCP SERVER                                   │
├──────────────────────────────────────────────────────────────────┤
│  ┌─────────────────┐  ┌─────────────────┐  ┌─────────────────┐  │
│  │   Transport     │  │   Protocol      │  │   Business      │  │
│  │   Layer         │  │   Handler       │  │   Logic         │  │
│  │                 │  │                 │  │                 │  │
│  │  • stdio        │  │  • tools/list   │  │  • Payment      │  │
│  │  • HTTP+SSE     │  │  • tools/call   │  │  • Inventory    │  │
│  │  • WebSocket    │  │  • resources/   │  │  • Shipping     │  │
│  │                 │  │  • prompts/     │  │  • CRM          │  │
│  └─────────────────┘  └─────────────────┘  └─────────────────┘  │
├──────────────────────────────────────────────────────────────────┤
│  ┌─────────────────┐  ┌─────────────────┐  ┌─────────────────┐  │
│  │   Auth          │  │   Rate          │  │   Logging &     │  │
│  │   Middleware    │  │   Limiting      │  │   Monitoring    │  │
│  └─────────────────┘  └─────────────────┘  └─────────────────┘  │
└──────────────────────────────────────────────────────────────────┘

Basic MCP Server Structure

Python Server

import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
import json
import logging

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("ecommerce-mcp")

# Initialize server
app = Server("ecommerce-server")

# Declare capabilities
@app.list_capabilities()
async def list_capabilities():
    return types.ServerCapabilities(
        tools=types.ToolsCapability(listChanged=True),
        resources=types.ResourcesCapability(subscribe=True),
        prompts=types.PromptsCapability(listChanged=True)
    )

# Define tools
@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="check_inventory",
            description="Check product inventory levels",
            inputSchema={
                "type": "object",
                "properties": {
                    "product_id": {"type": "string"}
                },
                "required": ["product_id"]
            }
        ),
        types.Tool(
            name="process_payment",
            description="Process a customer payment",
            inputSchema={
                "type": "object",
                "properties": {
                    "amount": {"type": "integer", "minimum": 50},
                    "customer_id": {"type": "string"},
                    "idempotency_key": {"type": "string"}
                },
                "required": ["amount", "customer_id", "idempotency_key"]
            }
        ),
        types.Tool(
            name="calculate_shipping",
            description="Calculate shipping rates",
            inputSchema={
                "type": "object",
                "properties": {
                    "destination_zip": {"type": "string"},
                    "weight_oz": {"type": "number"}
                },
                "required": ["destination_zip", "weight_oz"]
            }
        )
    ]

# Handle tool calls
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    logger.info(f"Tool called: {name} with args: {arguments}")

    try:
        if name == "check_inventory":
            result = await inventory_service.check(arguments["product_id"])
            return [types.TextContent(type="text", text=json.dumps(result))]

        elif name == "process_payment":
            result = await payment_service.charge(
                amount=arguments["amount"],
                customer_id=arguments["customer_id"],
                idempotency_key=arguments["idempotency_key"]
            )
            return [types.TextContent(type="text", text=json.dumps(result))]

        elif name == "calculate_shipping":
            result = await shipping_service.get_rates(
                destination=arguments["destination_zip"],
                weight=arguments["weight_oz"]
            )
            return [types.TextContent(type="text", text=json.dumps(result))]

        else:
            raise ValueError(f"Unknown tool: {name}")

    except Exception as e:
        logger.error(f"Tool error: {e}")
        return [types.TextContent(
            type="text",
            text=json.dumps({"error": str(e)})
        )]

# Main entry point
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

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

TypeScript Server

import { Server } from "@modelcontextprotocol/sdk/server";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio";
import { Tool, TextContent, ServerCapabilities } from "@modelcontextprotocol/sdk/types";

// Initialize server
const server = new Server(
  { name: "ecommerce-server", version: "1.0.0" },
  {
    capabilities: {
      tools: { listChanged: true },
      resources: { subscribe: true }
    }
  }
);

// Tool definitions
const tools: Tool[] = [
  {
    name: "check_inventory",
    description: "Check product inventory levels across warehouses",
    inputSchema: {
      type: "object",
      properties: {
        product_id: { type: "string", description: "Product SKU" },
        warehouse_id: { type: "string", description: "Optional warehouse filter" }
      },
      required: ["product_id"]
    }
  },
  {
    name: "process_payment",
    description: "Process a customer payment via Stripe",
    inputSchema: {
      type: "object",
      properties: {
        amount: { type: "integer", minimum: 50, description: "Amount in cents" },
        customer_id: { type: "string" },
        idempotency_key: { type: "string" }
      },
      required: ["amount", "customer_id", "idempotency_key"]
    }
  },
  {
    name: "calculate_shipping",
    description: "Get shipping rates for a package",
    inputSchema: {
      type: "object",
      properties: {
        destination_zip: { type: "string" },
        weight_oz: { type: "number", minimum: 0.1 }
      },
      required: ["destination_zip", "weight_oz"]
    }
  }
];

// Handle tools/list
server.setRequestHandler("tools/list", async () => ({
  tools
}));

// Handle tools/call
server.setRequestHandler("tools/call", async (request) => {
  const { name, arguments: args } = request.params;

  console.log(`[${new Date().toISOString()}] Tool called: ${name}`);

  try {
    let result: any;

    switch (name) {
      case "check_inventory":
        result = await inventoryService.check(args.product_id, args.warehouse_id);
        break;

      case "process_payment":
        result = await paymentService.charge({
          amount: args.amount,
          customerId: args.customer_id,
          idempotencyKey: args.idempotency_key
        });
        break;

      case "calculate_shipping":
        result = await shippingService.getRates({
          destinationZip: args.destination_zip,
          weightOz: args.weight_oz
        });
        break;

      default:
        throw new Error(`Unknown tool: ${name}`);
    }

    return {
      content: [{ type: "text", text: JSON.stringify(result) }]
    };
  } catch (error) {
    console.error(`Tool error: ${error}`);
    return {
      content: [{ type: "text", text: JSON.stringify({ error: String(error) }) }]
    };
  }
});

// Start server with stdio transport
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.log("E-commerce MCP server running on stdio");
}

main().catch(console.error);

HTTP Server with Authentication

For remote access, deploy an HTTP server with proper security:

from fastapi import FastAPI, HTTPException, Depends, Header
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from mcp.server.sse import sse_server
import jwt
import os

app = FastAPI()
security = HTTPBearer()

# JWT validation
def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    try:
        payload = jwt.decode(
            token,
            os.environ["JWT_SECRET"],
            algorithms=["HS256"]
        )
        return payload
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="Invalid token")

# Rate limiting
from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@app.post("/mcp")
@limiter.limit("100/minute")
async def mcp_endpoint(
    request: Request,
    user: dict = Depends(verify_token)
):
    """Main MCP endpoint with SSE for streaming responses"""
    # Parse JSON-RPC request
    body = await request.json()

    # Log request for audit
    logger.info(f"MCP request from {user['sub']}: {body['method']}")

    # Route to MCP server
    response = await mcp_server.handle_request(body)

    return response

# SSE endpoint for streaming
@app.get("/mcp/sse")
async def sse_endpoint(
    request: Request,
    user: dict = Depends(verify_token)
):
    """Server-Sent Events endpoint for MCP"""
    async def event_generator():
        async for event in mcp_server.stream_events():
            yield f"data: {json.dumps(event)}\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream"
    )

TypeScript HTTP Server

import express from "express";
import { Server } from "@modelcontextprotocol/sdk/server";
import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse";
import jwt from "jsonwebtoken";
import rateLimit from "express-rate-limit";

const app = express();
const mcpServer = new Server({ name: "ecommerce-server", version: "1.0.0" });

// Rate limiting
const limiter = rateLimit({
  windowMs: 60 * 1000, // 1 minute
  max: 100 // 100 requests per minute
});

// JWT authentication middleware
function authenticateToken(req: express.Request, res: express.Response, next: express.NextFunction) {
  const authHeader = req.headers["authorization"];
  const token = authHeader && authHeader.split(" ")[1];

  if (!token) {
    return res.status(401).json({ error: "No token provided" });
  }

  jwt.verify(token, process.env.JWT_SECRET!, (err, user) => {
    if (err) {
      return res.status(403).json({ error: "Invalid token" });
    }
    (req as any).user = user;
    next();
  });
}

app.use(express.json());
app.use(limiter);

// MCP SSE endpoint
app.get("/mcp/sse", authenticateToken, async (req, res) => {
  const transport = new SSEServerTransport("/mcp/messages", res);
  await mcpServer.connect(transport);
});

// MCP message endpoint
app.post("/mcp/messages", authenticateToken, async (req, res) => {
  const response = await mcpServer.handleRequest(req.body);
  res.json(response);
});

app.listen(3000, () => {
  console.log("MCP HTTP server running on port 3000");
});

Claude Desktop Configuration

For local development with Claude Desktop:

{
  "mcpServers": {
    "ecommerce": {
      "command": "python",
      "args": ["/path/to/ecommerce_server.py"],
      "env": {
        "STRIPE_API_KEY": "sk_test_...",
        "DATABASE_URL": "postgresql://...",
        "LOG_LEVEL": "INFO"
      }
    },
    "ecommerce-node": {
      "command": "node",
      "args": ["/path/to/ecommerce_server.js"],
      "env": {
        "STRIPE_API_KEY": "sk_test_..."
      }
    },
    "stripe": {
      "command": "npx",
      "args": ["-y", "@stripe/mcp", "--api-key", "sk_test_..."]
    }
  }
}

Docker Deployment

# Python MCP Server
FROM python:3.11-slim

WORKDIR /app

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

# Copy application
COPY . .

# Health check
HEALTHCHECK --interval=30s --timeout=10s \
  CMD python -c "import requests; requests.get('http://localhost:8000/health')"

# Run server
CMD ["python", "-m", "uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
version: '3.8'

services:
  mcp-server:
    build: .
    ports:
      - "8000:8000"
    environment:
      - STRIPE_API_KEY=${STRIPE_API_KEY}
      - DATABASE_URL=${DATABASE_URL}
      - JWT_SECRET=${JWT_SECRET}
      - LOG_LEVEL=INFO
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3
    deploy:
      replicas: 3
      resources:
        limits:
          cpus: '0.5'
          memory: 512M

  redis:
    image: redis:alpine
    volumes:
      - redis-data:/data

volumes:
  redis-data:

Security Best Practices

1. Authentication

# OAuth 2.0 for production
from authlib.integrations.starlette_client import OAuth

oauth = OAuth()
oauth.register(
    name='auth0',
    client_id=os.environ['AUTH0_CLIENT_ID'],
    client_secret=os.environ['AUTH0_CLIENT_SECRET'],
    authorize_url='https://your-domain.auth0.com/authorize',
    access_token_url='https://your-domain.auth0.com/oauth/token',
)

@app.middleware("http")
async def auth_middleware(request: Request, call_next):
    if request.url.path.startswith("/mcp"):
        token = request.headers.get("Authorization")
        if not token or not await validate_oauth_token(token):
            return JSONResponse(status_code=401, content={"error": "Unauthorized"})
    return await call_next(request)

2. Input Validation

from pydantic import BaseModel, validator
from typing import Optional

class PaymentRequest(BaseModel):
    amount: int
    customer_id: str
    idempotency_key: str
    metadata: Optional[dict] = None

    @validator('amount')
    def amount_must_be_positive(cls, v):
        if v < 50:
            raise ValueError('Amount must be at least 50 cents')
        if v > 10000000:  # $100,000 max
            raise ValueError('Amount exceeds maximum')
        return v

    @validator('customer_id')
    def customer_id_format(cls, v):
        if not v.startswith('cus_'):
            raise ValueError('Invalid customer ID format')
        return v

3. Rate Limiting Per User

from redis import Redis
import time

redis = Redis()

class RateLimiter:
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window_seconds = window_seconds

    async def check(self, user_id: str, tool_name: str) -> bool:
        key = f"rate:{user_id}:{tool_name}"
        current = redis.incr(key)
        if current == 1:
            redis.expire(key, self.window_seconds)
        return current <= self.max_requests

# Different limits for different tools
rate_limits = {
    "process_payment": RateLimiter(10, 60),      # 10/min
    "check_inventory": RateLimiter(100, 60),     # 100/min
    "get_sales_summary": RateLimiter(30, 60),    # 30/min
}

4. Audit Logging

import structlog
from datetime import datetime

logger = structlog.get_logger()

async def audit_log(
    user_id: str,
    tool_name: str,
    arguments: dict,
    result: dict,
    duration_ms: float
):
    await logger.info(
        "tool_invocation",
        user_id=user_id,
        tool=tool_name,
        arguments=sanitize_arguments(arguments),  # Remove sensitive data
        success=result.get("success", True),
        duration_ms=duration_ms,
        timestamp=datetime.utcnow().isoformat()
    )

def sanitize_arguments(args: dict) -> dict:
    """Remove sensitive fields from logs"""
    sensitive_fields = {"card_number", "cvv", "password", "api_key"}
    return {
        k: "***" if k in sensitive_fields else v
        for k, v in args.items()
    }

Monitoring and Observability

from prometheus_client import Counter, Histogram, start_http_server
import time

# Metrics
TOOL_CALLS = Counter(
    'mcp_tool_calls_total',
    'Total MCP tool calls',
    ['tool_name', 'status']
)

TOOL_DURATION = Histogram(
    'mcp_tool_duration_seconds',
    'Tool call duration',
    ['tool_name']
)

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    start = time.time()
    try:
        result = await execute_tool(name, arguments)
        TOOL_CALLS.labels(tool_name=name, status='success').inc()
        return result
    except Exception as e:
        TOOL_CALLS.labels(tool_name=name, status='error').inc()
        raise
    finally:
        duration = time.time() - start
        TOOL_DURATION.labels(tool_name=name).observe(duration)

# Start metrics server
start_http_server(9090)

Error Handling Patterns

from dataclasses import dataclass
from typing import Optional

@dataclass
class ToolError:
    code: str
    message: str
    details: Optional[dict] = None
    retryable: bool = False

class ToolErrors:
    INVALID_INPUT = ToolError("INVALID_INPUT", "Invalid input parameters", retryable=False)
    RATE_LIMITED = ToolError("RATE_LIMITED", "Too many requests", retryable=True)
    EXTERNAL_SERVICE = ToolError("EXTERNAL_SERVICE", "External service unavailable", retryable=True)
    INSUFFICIENT_FUNDS = ToolError("INSUFFICIENT_FUNDS", "Payment failed", retryable=False)

def format_error_response(error: ToolError, details: dict = None) -> dict:
    return {
        "success": False,
        "error": {
            "code": error.code,
            "message": error.message,
            "details": details or error.details,
            "retryable": error.retryable
        }
    }

Key Takeaways

Next up: Error Handling & Fallbacks - Building resilient tools that handle failures gracefully.