Lesson 10 — Observability, Testing, Logging

Make the server tell you what it did. Then write tests so you can change it without fear.

MCP Servers Module 4

The Module-4 bridge

What MCP gives you for free (and doesn't)

• Free: stdio framing logs from mcp dev; Inspector's request/response view.
• Not free: request IDs in your tool code; per-tool timing; per-user attribution; error-rate metrics.
• You build all of that on top.

Structured logging primer

Stdlib logging is enough for the course. Configure once at startup; pull a logger per module; log key-value pairs (not English sentences) so you can grep and aggregate.

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s %(levelname)s %(name)s %(message)s',
)

The log_tool_call wrapper

def log_tool_call(tool_name: str):
    """Decorator that logs entry/exit/error for any tool function."""
    def deco(fn):
        @functools.wraps(fn)
        async def wrapper(*args, **kwargs):
            log = logging.getLogger(f"pkb.tool.{tool_name}")
            log.info("tool=%s phase=start", tool_name)
            t0 = time.monotonic()
            try:
                result = await fn(*args, **kwargs) if asyncio.iscoroutinefunction(fn) else fn(*args, **kwargs)
                log.info("tool=%s phase=ok dur_ms=%.1f", tool_name, (time.monotonic()-t0)*1000)
                return result
            except Exception as e:
                log.warning("tool=%s phase=err type=%s", tool_name, type(e).__name__)
                raise
        return wrapper
    return deco

One decorator, every tool gets start/ok/err lines with timing — no per-tool boilerplate.

What NOT to log

Tracing patterns for MCP

A single user prompt may fan out into several tool calls. Generate a trace id at the top of each tool entry and propagate it through composed calls (summarize_and_save → summarize_via_sampling → …).

For local dev: just log trace_id=abc123 and parent=def456. For production: OpenTelemetry. We show the trace-id idea here; OTel lands in Lesson 15.

Unit-testing tools: create_connected_server_and_client_session

from mcp.shared.memory import create_connected_server_and_client_session
from pkb.server import mcp

async def test_echo():
    async with create_connected_server_and_client_session(mcp._mcp_server) as client:
        result = await client.call_tool("echo", {"text": "hi"})
        assert result.content[0].text == "hi"

An in-process client/server pair — no subprocess, no transport overhead, fast enough for CI.

Mocking sampling in tests

• Sampling tools (summarize, summarize_and_save) need a client that can respond to sampling/create_message requests.
• Option (a): use a SamplingFnT callback when constructing the test client to return a canned response.
• Option (b): split the tool so the sampling-dependent piece is a small async function you can patch with monkeypatch.
• Option (b) is what we do for unit tests; option (a) is for integration tests.

Integration testing via mcp dev

For end-to-end checks, mcp dev pkb/server.py boots Inspector with the real server. Smoke-test from the CLI by calling each tool once; capture the request log. This is your "did anything regress" gate before shipping.

mcp dev pkb/server.py  # then click through tools in Inspector

What we'll build today

• Add pkb/observability.py with the log_tool_call decorator + a redact() helper.
• Instrument every tool in server.py.
• Create pkb/tests/ with test_tools.py (echo, search, fetch, write, summarize-with-mocked-client) and test_resources.py (about, notes).
• Run pytest pkb/tests/.

Recap & what's next

• Logging is your operational eyes — but redact by default.
• The in-memory test harness makes unit tests fast and dependency-free.
• mcp dev is the integration-test backstop.