Production MCP Systems
Monitoring and Observability
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
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. :::
Sign in to rate