Logging and Observability

Servers in production need three classes of observability. Each is cheap to add early and painful to retrofit later.

Server-side structured logs. JSON-lines on stderr (or a sidecar file) with at minimum: timestamp, request id, method, latency, success/error. This is your incident-investigation source of truth.
Protocol-level logging. MCP defines a logging capability: when both sides advertise it, the client can subscribe to logs and the server emits them as notifications/message events. Useful when debugging integration issues from the host side.
Metrics and tracing. Counter for tools/call by name and outcome; histogram for latency; tracing span per request that propagates through any HTTP calls your tools make. OpenTelemetry is the safe default; the per-tool counter is the single most useful chart you will build.

The piece that catches new operators is the audit log from the security track. Audit logs are distinct from observability logs: they are append-only, immutable, and meant for compliance / forensics rather than debugging. Keep the two streams separate (different files, different rotation policies) so you don't lose audit data to a routine log-purge cron.

Code

Structured per-request logging in a tool wrapper·python

import logging, time, json, sys

logger = logging.getLogger("mcp")
handler = logging.StreamHandler(sys.stderr)
handler.setFormatter(logging.Formatter('%(message)s'))
logger.addHandler(handler)

def trace(method):
    async def wrapped(*args, **kwargs):
        start = time.perf_counter()
        ok, err = True, None
        try:
            return await method(*args, **kwargs)
        except Exception as e:
            ok, err = False, repr(e); raise
        finally:
            logger.info(json.dumps({
                "ts": time.time(), "tool": method.__name__,
                "latency_ms": round((time.perf_counter()-start)*1000, 2),
                "ok": ok, "error": err,
            }))
    return wrapped

Subscribing to MCP protocol logs from a client·python

async with ClientSession(read, write) as s:
    init = await s.initialize()
    if init.capabilities.get("logging"):
        await s.set_logging_level("info")
        # The session emits notifications/message events; pipe them to a sink.

Code

External links

Exercise

Progress

Comments 0