Production MCP Systems

Monitoring and Observability

5 min read

MCP servers fail in a way that is unusually easy to miss. When a tool returns a bad result, the model does not crash — it incorporates the bad result and answers confidently. There is no stack trace and no error page. The only place that failure is visible is in what you chose to record.

The Three Pillars

What to record, and what each layer answers

Logs — what happened in this one call
Metrics — what is happening across all calls
Traces — where the time went

One metric here is worth more than the rest: call count by tool. It tells you which tools the model actually uses. A tool called far more than expected usually has a description that overlaps another tool's. A tool never called at all is either badly described or shouldn't have been written — and either way, you cannot learn that from logs alone.

Structured Logging

import logging
import json
from datetime import datetime, timezone

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
            "function": record.funcName,
        }
        if hasattr(record, "extra"):
            log_data.update(record.extra)
        return json.dumps(log_data)

# Configure logging
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger = logging.getLogger("mcp")
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# Usage
logger.info("Tool called", extra={"tool": "search", "user": "user123"})

Metrics with Prometheus

import time
from prometheus_client import Counter, Histogram, generate_latest
from starlette.responses import Response
from mcp.server import MCPServer
from mcp.shared.exceptions import MCPError

TOOL_CALLS = Counter(
    "mcp_tool_calls_total", "Total tool calls", ["tool_name", "status"]
)
TOOL_LATENCY = Histogram(
    "mcp_tool_latency_seconds", "Tool call latency", ["tool_name"]
)


async def metrics_middleware(ctx, call_next):
    """Record every tools/call, whether it succeeds or raises."""
    if ctx.method != "tools/call":
        return await call_next(ctx)

    tool = (ctx.params or {}).get("name", "unknown")
    start = time.perf_counter()
    try:
        result = await call_next(ctx)
        TOOL_CALLS.labels(tool_name=tool, status="success").inc()
        return result
    except MCPError:
        TOOL_CALLS.labels(tool_name=tool, status="error").inc()
        raise
    finally:
        TOOL_LATENCY.labels(tool_name=tool).observe(time.perf_counter() - start)


mcp = MCPServer(name="observable-server", middleware=[metrics_middleware])

Instrumenting with middleware rather than inside each tool is what makes the numbers trustworthy. A tool you forget to wrap is a tool that silently reports zero calls, and a metric that is missing looks exactly like a metric that is healthy.

Two things to get right, both of which the finally above is doing deliberately:

  • Record the failures. A middleware that only counts on the success path produces an error rate of zero no matter how badly the server is doing.
  • Record the latency of failures too. Timeouts are the slowest calls you have, and excluding them makes your tail latency look best exactly when it is worst.

Expose the registry from the same ASGI app that serves MCP:

app = mcp.streamable_http_app()


async def metrics(request):
    return Response(generate_latest(), media_type="text/plain")


app.add_route("/metrics", metrics)

Health Checks

@app.route("/health")
async def health(request):
    checks = {
        "database": await check_database(),
        "redis": await check_redis(),
        "external_api": await check_external_api(),
    }

    all_healthy = all(checks.values())
    status_code = 200 if all_healthy else 503

    return JSONResponse(
        {"status": "healthy" if all_healthy else "unhealthy", "checks": checks},
        status_code=status_code
    )

async def check_database():
    try:
        await db.execute("SELECT 1")
        return True
    except:
        return False

Alerting Rules

Configure alerts in Prometheus/Grafana:

groups:
  - name: mcp-alerts
    rules:
      - alert: HighErrorRate
        expr: rate(mcp_tool_calls_total{status="error"}[5m]) > 0.1
        for: 5m
        annotations:
          summary: "High error rate in MCP server"

      - alert: SlowToolCalls
        expr: histogram_quantile(0.95, mcp_tool_latency_seconds) > 5
        for: 5m
        annotations:
          summary: "Tool calls taking too long"

Next: the test level that catches what unit tests structurally cannot. :::

Quiz

Module 5 Quiz: Production MCP Systems

Take Quiz
Was this lesson helpful?

Sign in to rate