Lab 12 — Token accounting, latency budgets, and an MCP-bridge demo
Lesson 12 · Module 6 · Choosing & Shipping · ~45 min hands-on, ~$0.10 budget
Goal: by the end of the lab you will have (a) instrumented the L04 research-assistant agent with per-turn token + cost logging, (b) computed a realistic latency budget for it, and (c) connected the same agent (or a slimmed-down version) to the stub MCP server shipped with this course and seen one successful tool call routed through MCP stdio. The lab has two halves — the accounting half is mechanical; the MCP-bridge half is the more interesting one.
Prereqs
-
L04 lab completed and working —
python agents/lesson04-tool-heavy-agent/agent.py "…"should run end-to-end without errors. If it’s broken in your checkout, fix it first; we’re going to modify and re-run it repeatedly here. -
agents/requirements.txtinstalled (it pinsmcp>=0.6,<0.7— you need that for Part 2):pip install -r agents/requirements.txt ANTHROPIC_API_KEYexported in your shell.- About $0.10 of API budget for the lab’s runs.
- Working directory at the repo root (paths below assume that).
Part 1 — Token accounting
Step 1 — Instrument agent.py with per-turn token counts
Open agents/lesson04-tool-heavy-agent/agent.py. Find the
log_step callback. Replace it with the version below, which adds
per-turn input/output token deltas alongside the running cost:
def log_step(step):
tools = [c.name for c in step.tool_calls]
turn_in = step.usage_delta.input_tokens
turn_out = step.usage_delta.output_tokens
print(
f"[step {step.turn}] stop_reason={step.stop_reason} "
f"tools={tools} cost=${step.cost_so_far_usd:.4f} "
f"this_turn=(in:{turn_in}, out:{turn_out})"
)
The key new fields are step.usage_delta.input_tokens and
step.usage_delta.output_tokens — the delta for this turn,
not the cumulative total. That’s what you want for budgeting: a per-turn
histogram tells you whether you have a long-tail problem.
Step 2 — Run it and read the per-turn numbers
python agents/lesson04-tool-heavy-agent/agent.py "What changed in Linux kernel 6.14's networking stack?"
Expected: each [step N] line now also shows the input/output token
counts for that turn. Typical numbers:
- Input tokens per turn: ~800–1500 (climbs as the conversation grows — tool results stack into history).
- Output tokens per turn: ~200–600 (relatively flat — the agent doesn’t suddenly start writing essays).
- Cost climbs roughly linearly across turns.
✓ Self-check: each step prints both an in: and out: number, and the cost is non-zero on the final step.
Part 1 — Latency budgeting
Step 3 — Time each turn
Now we want per-turn wall-clock latency. Replace log_step again, this
time tracking elapsed time between callback invocations:
import time
_last_t = [None]
def log_step(step):
now = time.perf_counter()
elapsed = (now - _last_t[0]) if _last_t[0] else 0.0
_last_t[0] = now
tools = [c.name for c in step.tool_calls]
print(
f"[step {step.turn}] tools={tools} cost=${step.cost_so_far_usd:.4f} "
f"turn_s={elapsed:.2f}"
)
Re-run the same question. Each step now shows its wall-clock latency
(turn_s=). Typical: ~1.5–4 seconds per turn for Sonnet on a
reasonable connection. The first turn shows 0.00 because there’s no
prior timestamp to subtract from — that’s fine, you only need
deltas.
Step 4 — Verify the latency formula
Sum the per-turn latencies. For a 6-turn run at 2.5s/turn you should see roughly ~15 seconds wall-clock total. Confirm:
total_latency ≈ turns × per_turn_latency
If your actual total is much higher than that formula predicts, the bottleneck is
almost certainly tool execution (the L04 tools are in-process dict lookups,
so they’re free — but a real network tool could easily dominate the
turn). If actual is much lower, you’ve probably got fewer turns than expected;
check step.turn on the last line.
✓ Self-check: you can state a latency budget for this agent in the form “~N turns × ~Ts per turn = ~Xs total.”
Part 2 — MCP-bridge sidebar
Now the more interesting half. The repo ships a stub MCP server at
agents/_eval/mcp_stub/server.py that exposes a single tool
(pkb_search) over the standard Model Context Protocol stdio transport.
We’re going to (a) confirm the stub boots, (b) call it from a minimal client,
and (c) wire that same tool into a Claude SDK agent. The takeaway is that the agent
code is decoupled from the tool implementation — the tool lives in a separate
process and could in principle live on another machine or be written in another
language.
Step 5 — Inspect the stub server and confirm it boots
Open agents/_eval/mcp_stub/server.py and skim it. Note the single
pkb_search tool and the in-process PKB dict (the same
shape used by the L04 agent’s tools). Then run it standalone to confirm it
starts cleanly:
python -m agents._eval.mcp_stub.server
Expected: the process starts and silently waits on stdin (it’s a stdio MCP
server — it speaks to a client over its own stdin/stdout, not to your
terminal). No output is normal. Kill it with Ctrl-C — we’ll
launch it from the client in the next step.
✓ Self-check: the standalone server starts without a traceback. If it crashes on import, fix that before continuing — the client will silently hang on a broken server.
Step 6 — Write a minimal MCP client
Create a new file mcp_client_demo.py at the repo root with the contents
below. It launches the stub server as a subprocess, lists its tools, and calls
pkb_search once:
"""Connect to the stub MCP server and call pkb_search once."""
import asyncio
from mcp import ClientSession
from mcp.client.stdio import stdio_client, StdioServerParameters
async def demo():
params = StdioServerParameters(
command="python",
args=["-m", "agents._eval.mcp_stub.server"],
)
async with stdio_client(params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# List the tools the server exposes.
tools = await session.list_tools()
print("Tools:", [t.name for t in tools.tools])
# Call pkb_search.
result = await session.call_tool(
"pkb_search", {"query": "BIG TCP"}
)
for content in result.content:
print(content.text)
asyncio.run(demo())
Run it:
python mcp_client_demo.py
Expected:
Tools: ['pkb_search']
kernel-6.14-net-notes: Personal notes on Linux 6.14 networking - io_uring zero-copy TX, BIG TCP for IPv4 by default.
You just called a tool hosted in a separate process via the standard Model Context Protocol. No agent is involved yet — that’s the next step.
✓ Self-check: the client prints the tool list and the search hit. If it hangs, kill it and re-run Step 5 standalone — the hang almost always means a server-side import error.
Step 7 — Wire the MCP tool into a Claude SDK agent
This is the bridge moment: the same MCP-hosted tool is now exposed to a Claude SDK
agent via a thin adapter. Create agent_with_mcp.py at the repo root:
"""Claude SDK agent that calls pkb_search via the MCP stub server."""
import asyncio
from anthropic import Anthropic
from anthropic.agents import Agent, tool
from mcp import ClientSession
from mcp.client.stdio import stdio_client, StdioServerParameters
async def main(question: str) -> None:
params = StdioServerParameters(
command="python",
args=["-m", "agents._eval.mcp_stub.server"],
)
async with stdio_client(params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# Adapter: wrap the MCP tool as a Claude SDK @tool.
@tool
async def pkb_search(query: str) -> str:
"""Search the personal knowledge base (MCP-hosted)."""
result = await session.call_tool(
"pkb_search", {"query": query}
)
return "\n".join(c.text for c in result.content)
agent = Agent(
client=Anthropic(),
model="claude-3-5-sonnet-latest",
system="Use pkb_search to find what you need. Be concise.",
tools=[pkb_search],
max_turns=5, max_cost_usd=0.05,
)
result = await agent.run_async(question)
print(result.final_text)
if __name__ == "__main__":
import sys
q = sys.argv[1] if len(sys.argv) > 1 else "What's BIG TCP?"
asyncio.run(main(q))
Run it:
python agent_with_mcp.py "What changed in Linux 6.14 networking?"
Expected: the agent calls pkb_search, gets the kernel-6.14 entry back,
and answers the question in one or two sentences. The key thing:
the tool implementation lives in agents/_eval/mcp_stub/server.py, not
in this file. You could host that server on a different machine, in a different
language, behind a different VCS — and the agent code wouldn’t change.
That’s the MCP value proposition in three files.
✓ Self-check: the agent’s final answer references BIG TCP or io_uring (the actual content of the PKB entry), and stays within the 5-turn / $0.05 budget.
What you should leave with
- First-hand experience of per-turn token + latency logging on a real agent — not just cumulative cost, but the histogram you need to spot a long-tail turn.
-
A working
agent_with_mcp.pythat calls an MCP-hosted tool from a Claude SDK agent. Keep this around — it’s a 40-line reference for every future MCP integration. - The instinct that MCP isn’t a “bigger framework” — it’s an interop layer between agents and tools. None of the four agent frameworks from Modules 2–5 compete with it; all four can consume MCP-hosted tools.
- A pointer to the Creating MCP Servers course for the deeper dive on authoring servers (auth, transports beyond stdio, hosting at scale).
Troubleshooting
Common stumbles
-
ModuleNotFoundError: No module named 'mcp'— install:pip install -r agents/requirements.txt. Themcppackage is pinned at>=0.6,<0.7in the Agents course’s requirements file. -
MCP client hangs forever — almost always a server-side
import error. Re-run the standalone server (Step 5) to see the real traceback;
stdio_clientswallows the server’s stderr by default. -
agent.run_asyncnot found — the Claude SDK’s async name may differ in your installed version. Checkdir(Agent)for the right method. The synchronousagent.run(…)works too if you drop theawaits and the surroundingasync with. -
MCP tool returns empty — confirm your query substring
matches something in the stub’s
PKBdict. “BIG TCP” hits; “ZFS” does not. The stub uses naive substring matching — not semantic search. -
Cost budget tripped early — if the agent loops on
pkb_searchwith slightly varying queries, you can burn throughmax_cost_usd=0.05in 3–4 turns. Lowermax_turnsfurther (trymax_turns=3) to surface the loop fast.