Observability, Testing, Logging
Lesson 10 · MCP Servers · ~10 min read
Why this lesson is the bridge
Modules 1–3 grew PKB from a single echo tool to seven tools, two resources,
one prompt, and a stub auth layer. The whole thing was developed and verified by clicking
through Inspector. That workflow stops scaling the moment someone else touches the code, or
the moment you change something subtle — a Pydantic constraint, an error message shape,
the order of two awaits inside summarize_and_save. You will not notice the
regression until the next time you happen to click the right button, and by then the cause is
three commits back.
This lesson is where PKB grows the two cheap habits that let you change it later without breaking it: logs you can read and tests that fail when behaviour drifts. Neither needs new infrastructure. Both pay for themselves the first time a tool starts failing in CI before it ever reaches Inspector.
Structured logging with logging
The stdlib logging module is enough for a personal MCP server. Configure it once
in server.py (or in a _configure_logging() call from
observability.py) at INFO level, with a format that includes
timestamp, level, logger name, and message. Pull a logger per module with the standard
idiom:
log = logging.getLogger(__name__)
Inside tool functions, log key–value pairs rather than English sentences. Sentences are fine for humans reading one line; key–value pairs are what you grep, what you ship to an aggregator, and what you parse into JSON later without a regex rewrite:
log.info("tool=echo phase=ok dur_ms=1.2 text_len=5")
Grep-friendly. Aggregation-friendly. Trivial to switch to a JSON formatter when you move to
production (Lesson 15 covers production formatters and shippers). The single most useful
pattern in this style is a decorator — log_tool_call("name") — that
wraps every tool function and emits three lines max per call: phase=start,
phase=ok dur_ms=…, or phase=err type=…. One decorator,
zero per-tool boilerplate, complete coverage. Stack it underneath
@mcp.tool() on each tool and the lesson’s observability story is done in
seven lines of diff.
Tracing patterns for MCP tool calls
A single LLM turn can fan out into multiple tool calls — and PKB’s composed tools
(notably summarize_and_save) fan out further server-side. Without trace
ids you cannot reconstruct a session from logs; every line looks isolated and the order is
whatever the scheduler decided. The minimum-viable trace is three steps:
- At the top of each tool, generate a short random
trace_id(or pull one fromctxif the caller already provided it). - Log every line from that tool with
trace_id=…. - For composed tools, pass the same
trace_idinto the sub-call and log it with aparent=…field on the child lines.
For local development that is enough — you can reconstruct any single tool call with
grep trace=abc12345 server.log. For production you graduate to OpenTelemetry:
same idea, standardised wire format, support for distributed propagation across an HTTP
transport so a trace can stitch together a browser client, your MCP server, and any
downstream services it calls. We defer OTel to Lesson 15 (production hardening) where the
production HTTP server can carry trace headers across services and an OTel collector becomes
worth running.
Redaction: what NOT to log
The redaction-by-default policy
The rule for every new log line is: assume the value is sensitive until you have proved it is not. Concretely:
- Never log full tool arguments. They may contain user-typed prompts, file contents, OAuth bearer tokens that the LLM helpfully copy-pasted.
- Never log full LLM responses from sampling. The user paid tokens for that output; logging it both leaks the data and inflates log storage.
- Never log
Authorizationheaders, API keys, env-var values, or anything fromos.environdirectly. - Never log
str(exception)from third-party SDKs. The message often contains the failing URL with API keys in the query string. - DO log: tool name, phase (
start/ok/err), duration, exception type, request id, user id (the id, not their data), input lengths or counts (e.g.,text_len=5,n_notes=3).
PKB ships a redact(value) helper in pkb/observability.py that
masks any string longer than 200 chars and summarises bytes by length. Use it as the default
formatter for any field that might be sensitive — the cost of a redaction you
did not need is zero; the cost of a key in plaintext is a rotation incident.
Unit-testing tools with the in-memory client/server harness
The mcp package ships
mcp.shared.memory.create_connected_server_and_client_session. It returns an
async context manager that wires an in-process client to an in-process server with
no subprocess and no transport overhead — pure Python async over a
memory channel. That makes per-tool unit tests fast enough for CI: a full suite of nine
tests against PKB’s tools and resources runs in well under a second.
The pattern is small enough to memorise:
import pytest
from mcp.shared.memory import create_connected_server_and_client_session
from pkb.server import mcp
@pytest.mark.asyncio
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"
A few notes worth pinning to the wall:
mcp._mcp_serverreaches the lower-level server beneath FastMCP. The leading underscore is a stable-enough public access in current SDK versions; we pin viarequirements.txtso the underscore does not turn into a silent breakage.pytest-asyncioprovides the@pytest.mark.asynciomarker. Forget the marker and pytest will report the test as passing without ever running it.- The harness runs the full protocol — request framing, response framing, error mapping — so failures match what Inspector would see. The only thing it skips is the subprocess boundary.
For tools that use sampling (summarize), you have two choices. The first is to
pass a sampling_callback when creating the connected session so the client can
answer the server’s sampling/create_message request with a canned
response. The second is to factor the LLM-dependent piece into a small async function and
patch it with monkeypatch in the test. We use the second pattern in
test_tools.py because it is simpler and does not depend on the SDK’s
sampling-callback signature stability.
Integration testing with mcp dev
For an end-to-end check, mcp dev pkb/server.py boots Inspector against the real
server with the real transport. The lab walks through a manual smoke checklist: call every
tool once, watch the log file (or stdout) for phase=ok lines, confirm the exit
code is clean. This is intentionally manual — automating Inspector itself is overkill
for a hobby or teaching project, and the maintenance burden of driving a UI through code is
more than the regression risk it would catch. CI runs the unit tests; humans run the
integration smoke before each merge.
Test structure for PKB
Two test modules in pkb/tests/:
test_tools.py— one test per tool:test_echo_round_trip,test_search_returns_known_titles,test_fetch_known_id,test_fetch_unknown_id_is_error,test_write_then_fetch,test_summarize_uses_sampling_helper.test_resources.py—test_about_resource_returns_text,test_notes_resource_known,test_notes_resource_unknown.
Each test uses create_connected_server_and_client_session and is one or two
assertions long. The whole suite runs in under a second. The point is not coverage theatre
— it is a tripwire: if any of these change shape, behaviour drifted, and you want to
know before you ship the change, not after the next user opens Inspector and the
“known good” flow returns something unexpected.
Common pitfalls
Four pitfalls that bite every team
- Logging full args “just for debugging” → secret leak in your log
file. The fix: write the
redact()helper before you write the log line, not after the first incident. - Forgetting
@pytest.mark.asyncioon async test functions → pytest reports them as passing without ever running. Add a CI grep that fails the build if anyasync def test_lacks the marker. - Calling
mcp._mcp_serverfrom your test code without pinning themcppackage version → silent breakage on SDK updates. Pin inrequirements.txtand bump deliberately. - Logging
str(exception)instead oftype(exception).__name__→ API key in your log file the first time a third-party SDK throws an error that includes the failing URL.
References
- Python
loggingHOWTO — the canonical primer; covers handlers, formatters, propagation, and the per-module logger pattern used by PKB. - pytest-asyncio documentation — the
@pytest.mark.asynciomarker, mode configuration, and how to run async fixtures. - MCP Python SDK —
mcp.shared.memory— source forcreate_connected_server_and_client_sessionso you can see exactly what the in-memory harness wires together. - OpenTelemetry for Python — deferred reading for Lesson 15; the production-grade replacement for the home-grown
trace_idpattern. - PKB sample-server tree — the working server at
mcp-servers/sample-server/python/wherepkb/observability.pyand the newpkb/tests/directory land during this lesson’s lab.