Large MCP Results in Code: Before, After & Libraries

Step 1Before: a pass-through tool, and where its tokens actually go.

This is the shape most servers ship first: call the upstream API and return what it returns. It even has pagination — per_page and page, the same parameters the REST API takes.

# before.py — the tool as most servers first ship it
import json
from mcp.server import MCPServer

mcp = MCPServer("prs-before")
PRS = json.load(open("prs.json"))          # stand-in for GET /repos/{o}/{r}/pulls

@mcp.tool()
def list_pull_requests(state: str = "all", per_page: int = 30, page: int = 1) -> str:
    """List pull requests in the repository."""
    rows = [p for p in PRS if state == "all" or p["state"] == state]
    start = (page - 1) * per_page
    return json.dumps(rows[start:start + per_page])   # every field, every nested object

One call with the defaults returns 656,124 characters, or 187,521 tokens. That is seven and a half times Claude Code's 25,000-token cap. Run through the Claude Agent SDK, the result was written to a file instead of the context, and the test session, which was allowed only this one tool, could not open the file and never answered. And the wire figure is twice that. Because the function is annotated -> str, MCPServer publishes an outputSchema of {"result": string} and sends the payload again as structuredContent: 656,124 characters of text plus 694,798 of structured copy, on every call. The 694,797 characters Claude Code reported match that second copy. fastmcp does the same. A tool that returns a ready-made JSON string should say so with @mcp.tool(structured_output=False), which removes both the schema and the copy; a tool that wants structured output should return it once, deliberately (STEP 3).

Paging does not help either, because the size is in the record, not the count. A per-field breakdown of the payload shows where:

field            tokens    share
head              56,157   29.9%   ┐ each embeds a full repo object:
base              54,376   29.0%   ┘ 91,207 tokens (48.6%) of copies
body              44,016   23.5%
user               9,504    5.1%
_links             6,540    3.5%
other *_url        5,700    3.0%
everything else    ~11,000  ~6%

Nearly half the payload is repository objects: every pull request carries two full copies, one under head and one under base. No model needs it, and no page size removes it. That is why the first fix is projection, not pagination.

Step 2Project first: name the fields you return, and cap the free text.

# after.py — same data, a contract the context window can survive
import base64, hashlib, json
from typing import Annotated, Any
from pydantic import BaseModel
from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from mcp.types import CallToolResult, TextContent
import tiktoken

mcp = MCPServer("prs-after")
PRS = json.load(open("prs.json"))
ENC = tiktoken.get_encoding("o200k_base")
PAGE_BUDGET = 4_000          # tokens per page, sized for the strictest host you support
ENVELOPE = 150               # cursor, note, flags: reserve it, or the page overshoots
BODY_CHARS = 280             # preview only; full body via get_pull_request

def tokens(obj: Any) -> int:
    return len(ENC.encode(json.dumps(obj, separators=(",", ":"))))

# 1. Projection: decide the fields, don't inherit them from the upstream API.
def project(p: dict) -> dict:
    body = p.get("body") or ""
    return {
        "number": p["number"], "title": p["title"], "state": p["state"],
        "draft": p["draft"], "author": p["user"]["login"],
        "labels": [l["name"] for l in p["labels"]],
        "created_at": p["created_at"], "merged_at": p["merged_at"],
        "head": p["head"]["ref"], "base": p["base"]["ref"],
        "body_preview": body[:BODY_CHARS],
        "body_truncated": len(body) > BODY_CHARS,
    }

Applied to all 30 records, project() brings the payload from 187,521 tokens to 5,297 (−97%). Three rules are encoded in it. Fields are an allow-list you choose, so an upstream API that adds a field does not silently grow your result. Nested objects collapse to the one value an agent uses: user becomes the login and head becomes the branch name. Free text gets a preview and a flag saying it was cut, and the full text is one call away in get_pull_request (STEP 3), which keeps the identifier a follow-up call needs.

When callers genuinely need different fields, let them ask, from safest to most powerful: a fields: list[str] argument checked against an allow-list; a JMESPath expression (jmespath 1.1.0 in Python, @jmespath-community/jmespath in JavaScript), which has no side effects and is safe to accept from a model; or a jq filter (jq 1.13.0 bindings), which is Turing-complete and needs a CPU and time limit. All three have precedents in production: Stainless-generated servers add a jq_filter parameter to every tool, and Zscaler's server takes an optional JMESPath query on its list tools.

Step 3Then page by tokens, with a cursor bound to its query — and count what you send, not what you meant to send.

# 2. Opaque cursor bound to the query that produced it.
def query_key(state: str) -> str:
    return hashlib.sha256(state.encode()).hexdigest()[:12]

def encode_cursor(offset: int, state: str) -> str:
    raw = json.dumps({"o": offset, "q": query_key(state)}).encode()
    return base64.urlsafe_b64encode(raw).decode()

def decode_cursor(cursor: str, state: str) -> int:
    try:
        data = json.loads(base64.urlsafe_b64decode(cursor.encode()))
    except Exception:
        raise ToolError("Invalid cursor. Call again without a cursor to restart.")
    if data.get("q") != query_key(state):
        raise ToolError("This cursor belongs to a different query. Restart without a cursor.")
    return int(data["o"])        # offset is fine for a frozen list; use a keyset (sort key) for live data

class PullRequestPage(BaseModel):
    items: list[dict]
    next_cursor: str | None
    truncated: bool
    total_matching: int
    note: str | None = None

# 3. Page by token budget, not by record count.
@mcp.tool()
def list_pull_requests(state: str = "all", cursor: str | None = None) -> Annotated[CallToolResult, PullRequestPage]:
    """List pull requests, newest first, as compact summaries (no diffs, no full bodies).
    If truncated is true, call again with next_cursor to continue.
    Use get_pull_request(number) for one PR's full body."""
    rows = [p for p in PRS if state == "all" or p["state"] == state]
    offset = decode_cursor(cursor, state) if cursor else 0
    items, used = [], 0
    for p in rows[offset:]:
        row = project(p)
        cost = tokens(row)
        if items and used + cost > PAGE_BUDGET - ENVELOPE:
            break
        items.append(row); used += cost
    end = offset + len(items)
    more = end < len(rows)
    page = PullRequestPage(
        items=items, total_matching=len(rows), truncated=more,
        next_cursor=encode_cursor(end, state) if more else None,
        note=f"Showing {offset + 1}-{end} of {len(rows)}. Pass next_cursor to continue." if more else None,
    ).model_dump(mode="json")
    # 4. Ship the compact text you counted, not the SDK's indent=2 rendering.
    text = json.dumps(page, separators=(",", ":"))
    return CallToolResult(content=[TextContent(type="text", text=text)], structured_content=page)

# 5. Drill-down instead of a bigger page.
@mcp.tool()
def get_pull_request(number: int, body_max_chars: int = 4_000) -> dict:
    """Get one pull request with its body (capped at body_max_chars)."""
    p = next((p for p in PRS if p["number"] == number), None)
    if p is None:
        raise ToolError(f"No pull request #{number}. Use list_pull_requests to find numbers.")
    body = p.get("body") or ""
    return {**project(p), "body": body[:body_max_chars],
            "body_truncated": len(body) > body_max_chars, "url": p["html_url"]}

The loop stops a page when the next record would cross the budget, never on a record count, and it always ships at least one record, so an oversized row cannot produce an empty page and a cursor that never moves. The cursor is opaque base64, but it carries a hash of the query that produced it, so a cursor reused with different filters fails with a message that says what to do instead of returning the wrong rows. The result walks the 30 records in two pages of 3,851 and 1,528 tokens, with truncated, total_matching, next_cursor and a one-line note in the result itself.

Two things in that code are there because the first draft was wrong, and the tests in STEP 4 caught both.

• The SDK pretty-prints your result. The first draft returned the Pydantic model and let MCPServer serialize it. It renders the text block with indent=2 (pydantic_core.to_json(result, fallback=str, indent=2) in func_metadata.py), so a page budgeted at 4,000 tokens went out at 4,838. Returning a CallToolResult with text you serialized yourself fixes it, and the Annotated[CallToolResult, PullRequestPage] return type keeps the published outputSchema and the validation of structured_content.
• The envelope is part of the page. With the indent gone, the page still came out at 4,016: the budget counted the records and forgot the cursor, the note and the flags. ENVELOPE reserves room for them.
• Only ToolError reaches the model. The first draft raised ValueError for a bad cursor. MCPServer treats any exception other than ToolError as a crash and sends only Error executing tool list_pull_requests, so the remedy in the message never arrived. The TypeScript SDK passes a plain Error's message through; Python does not.

The text block is a compact copy of the structured result, not a summary. The spec says a tool returning structured content SHOULD also return the serialized JSON as text, and clients disagree about which one the model reads: langchain-mcp-adapters sends the text to the model and keeps structuredContent as an artifact the model never sees, while the OpenAI Agents SDK defaults use_structured_content=False precisely because most servers mirror. A mirrored, compact text block is the one choice every client reads correctly, and it is the one the budget measures.

The same contract on the TypeScript SDK is a line-for-line port; only the tool body differs:

  }, async ({ state, cursor }) => {
    const rows = PRS.filter(p => state === 'all' || p.state === state);
    const offset = cursor ? decodeCursor(cursor, state) : 0;
    const items: ReturnType<typeof project>[] = [];
    let used = 0;
    for (const p of rows.slice(offset)) {
      const row = project(p), cost = tokens(row);
      if (items.length && used + cost > PAGE_BUDGET - ENVELOPE) break;
      items.push(row); used += cost;
    }
    const end = offset + items.length, more = end < rows.length;
    const page = {
      items, total_matching: rows.length, truncated: more,
      next_cursor: more ? encodeCursor(end, state) : null,
      note: more ? `Showing ${offset + 1}-${end} of ${rows.length}. Pass next_cursor to continue.` : null,
    };
    return { content: [{ type: 'text', text: JSON.stringify(page) }], structuredContent: page };

It walks the same 30 records in pages of 3,844 and 1,524 tokens. gpt-tokenizer 4.1.0 provides o200k_base in pure TypeScript; its isWithinTokenLimit(text, limit) stops encoding as soon as the limit is crossed, which suits a size check better than encoding the whole string.

Step 4Test the contract, not the tool: five assertions that fail when a page grows.

async def walk(client, **args):
    cursor, pages = None, []
    while True:
        r = await client.call_tool("list_pull_requests", {**args, **({"cursor": cursor} if cursor else {})})
        pages.append(r)
        cursor = r.structured_content["next_cursor"]
        if cursor is None:
            return pages

@pytest.mark.anyio
async def test_every_page_fits_the_budget():
    async with Client(after.mcp) as c:
        for r in await walk(c):
            assert len(ENC.encode(r.content[0].text)) <= after.PAGE_BUDGET

@pytest.mark.anyio
async def test_cursor_walk_returns_every_record_once():
    async with Client(after.mcp) as c:
        numbers = [i["number"] for r in await walk(c) for i in r.structured_content["items"]]
    assert numbers == [p["number"] for p in after.PRS]

@pytest.mark.anyio
async def test_truncation_is_in_band():
    async with Client(after.mcp) as c:
        first = (await walk(c))[0].structured_content
    assert first["truncated"] is True and first["next_cursor"] and "next_cursor" in first["note"]

@pytest.mark.anyio
async def test_cursor_from_another_query_is_rejected():
    async with Client(after.mcp) as c:
        cur = (await c.call_tool("list_pull_requests", {"state": "all"})).structured_content["next_cursor"]
        r = await c.call_tool("list_pull_requests", {"state": "open", "cursor": cur})
    assert r.is_error and "different query" in r.content[0].text

@pytest.mark.anyio
async def test_one_record_larger_than_the_budget_still_ships(monkeypatch):
    monkeypatch.setattr(after, "PAGE_BUDGET", 10)   # smaller than any single row
    async with Client(after.mcp) as c:
        r = await c.call_tool("list_pull_requests", {})
    assert len(r.structured_content["items"]) == 1   # progress over purity: never an empty page

These run in-process (the Client(server) pattern from Testing MCP Servers) and take 0.3 seconds. Each guards a different failure. The budget test reads the text block from the wire, which is how it caught the indent bug. The walk test proves the cursor neither skips nor repeats a record. The in-band test fails if someone moves the flag into logs. The foreign-cursor test checks that the error text actually reaches the client, which is how it caught the ValueError. The last test sets the budget below the size of one row and asserts that a page still ships.

Point the budget test at a fixture recorded from your largest real account, not a synthetic one. These fixtures are 30 public pull requests; a customer with 4,000-line PR descriptions is the input that breaks a budget, and it is the one you do not have unless you record it.

Step 5A server-wide ceiling for the tools nobody has reshaped yet.

Per-tool contracts take work, and a server with 40 tools will not get them all at once. A ceiling in middleware covers the rest: it measures every tools/call result and turns an oversized one into an error the model can act on.

# guard.py — one ceiling for every tool, including the ones nobody re-shaped yet
import logging
import tiktoken
from mcp.types import CallToolResult, TextContent

ENC = tiktoken.get_encoding("o200k_base")
HARD_CAP = 20_000   # below Claude Code's 25,000-token cap
log = logging.getLogger("result-guard")

async def result_guard(ctx, call_next):
    result = await call_next(ctx)
    if ctx.method != "tools/call" or not isinstance(result, dict):
        return result
    text = "".join(b.get("text", "") for b in result.get("content", []))
    n = len(ENC.encode(text))
    if n <= HARD_CAP:
        return result
    tool = (ctx.params or {}).get("name")
    log.warning("tool %s returned %d tokens (cap %d)", tool, n, HARD_CAP)
    return CallToolResult(is_error=True, content=[TextContent(type="text", text=
        f"{tool} returned about {n:,} tokens, over this server's {HARD_CAP:,}-token limit. "
        "Narrow the query (filters, a smaller page) or fetch one item at a time.")])

Registered on the STEP 1 server with mcp.middleware.append(result_guard), the 187,521-token result becomes one sentence: "list_pull_requests returned about 187,521 tokens, over this server's 20,000-token limit. Narrow the query (filters, a smaller page) or fetch one item at a time." Two details matter. The replacement must be a CallToolResult model; a plain dict fails validation under the 2026-07-28 revision because it has no resultType. And the middleware API is marked provisional in mcp 2.x ("the signature may change in a 2.x minor release"), so pin the SDK version and keep the test.

If your server is on fastmcp, ResponseLimitingMiddleware(max_size=..., tools=[...]) is the built-in equivalent, and its behaviour is worth knowing before you rely on it. It counts bytes of the whole serialized result, not tokens; the 656KB payload measured 1,389,809 bytes, because it counts the text block and the doubled structuredContent from STEP 1, each with its quotes escaped. Over the limit it truncates instead of refusing: with max_size=60_000 the model received 16,530 tokens of a JSON array cut off mid-string, followed by [Response truncated due to size limit]. It also drops structuredContent and removes outputSchema from every tool it limits. It is a sound backstop and a poor contract, and refusing with a remedy, as above, is usually better than handing the model half a document.

Step 6Reshape a server you don't own, at a proxy.

Most of the oversized results an agent sees come from third-party servers you cannot edit. fastmcp can front one as a proxy and replace its tools with shaped versions, without touching the upstream code:

# wrap.py — reshape a server you don't own, without touching its code (fastmcp 4.1.3)
import json
from fastmcp import FastMCP
from fastmcp.server import create_proxy
from fastmcp.server.middleware.response_limiting import ResponseLimitingMiddleware
from fastmcp.tools import Tool
from fastmcp.tools.tool_transform import forward
from fastmcp.tools.base import ToolResult

# Stand-in for the third-party server: one fat tool.
upstream = FastMCP("third-party")
PRS = json.load(open("prs.json"))
@upstream.tool
def list_pull_requests(state: str = "all") -> str:
    """List pull requests."""
    return json.dumps([p for p in PRS if state == "all" or p["state"] == state])

KEEP = ("number", "title", "state", "created_at", "merged_at")

async def build() -> FastMCP:
    proxy = create_proxy(upstream, name="prs-gateway")
    fat = await proxy.get_tool("list_pull_requests")

    async def slim(state: str = "all", limit: int = 20) -> ToolResult:
        raw = await forward(state=state)
        rows = json.loads(raw.content[0].text)
        out = [{k: r[k] for k in KEEP} | {"author": r["user"]["login"]} for r in rows[:limit]]
        return ToolResult(content=json.dumps({"items": out, "truncated": len(rows) > limit,
                                              "total_matching": len(rows)}, separators=(",", ":")))

    proxy.add_tool(Tool.from_tool(fat, name="list_pull_requests_slim", transform_fn=slim,
        description="List pull requests as compact summaries. Raise limit for more."))
    proxy.disable(names={"list_pull_requests"})                      # hide the fat original
    proxy.add_middleware(ResponseLimitingMiddleware(max_size=60_000))  # bytes, a backstop
    return proxy

The gateway exposes only list_pull_requests_slim, and all 30 pull requests come back in 1,763 tokens. forward() calls the original tool, disable() hides it from tools/list, and the byte limit stays behind it as a backstop. The one cost is that the proxy still receives the full upstream payload; it saves the model's context, not the network.

Packaged alternatives exist, at very different levels of maturity. atlassian-labs/mcp-compressor (v0.38.0, Apache-2.0, in Python, TypeScript and Rust) is the most maintained: it mainly compresses the tool catalog, and its opt-in --shrinkfit flag shrinks results over 20,000 characters while keeping the top-level shape and letting the caller ask for the full result with full=true. mcp-trunc-proxy (0.2.2, GPL-3.0) implements the handle pattern generically, offloading large results to a store and returning a preview plus a retrieval tool, but it is a small project. The rest of what a search turns up (mcp-rtk, mcpslim, mcpsqueeze, mcp-context-proxy) had zero to four GitHub stars on 2026-10-21. None of these measured savings have been reproduced here; treat them as the authors' claims.

Step 7Host-side hooks: where each agent framework lets you rewrite a result — and the one that fires too late.

If you build the agent rather than the server, you can shape results as they arrive. In the Claude Agent SDK, a PostToolUse hook can replace an MCP tool's output before the model reads it:

# host.py — shape MCP results in the host, before the model reads them (claude-agent-sdk 0.2.170)
import asyncio, json, sys
from claude_agent_sdk import ClaudeAgentOptions, HookMatcher, query

KEEP = ("number", "title", "state", "merged_at")

async def slim_mcp_result(input_data, tool_use_id, context):
    resp = input_data["tool_response"]          # a string by the time the hook runs
    try:
        data = json.loads(resp)
        if isinstance(data, dict) and "result" in data:   # structured {"result": ...} wrapper
            data = json.loads(data["result"])
    except (TypeError, ValueError):
        return {}                               # e.g. Claude Code's "saved to file" notice
    slim = [{k: r.get(k) for k in KEEP} for r in data]
    return {"hookSpecificOutput": {
        "hookEventName": "PostToolUse",
        "updatedMCPToolOutput": json.dumps(slim, separators=(",", ":")),
    }}

options = ClaudeAgentOptions(
    model="haiku",
    mcp_servers={"prs": {"type": "stdio", "command": sys.executable, "args": ["before.py"]}},
    allowed_tools=["mcp__prs__list_pull_requests"],
    hooks={"PostToolUse": [HookMatcher(matcher="mcp__prs__.*", hooks=[slim_mcp_result])]},
)

On a two-PR result (about 43,000 characters), the hook worked as documented: asked for the keys of the first record, the model answered number, title, state, merged_at, the four the hook kept. On the 30-PR result it did nothing, and the reason is not in the documentation. The size limit runs before the hook. By the time slim_mcp_result fired, tool_response was already the string Error: result (694,797 characters) exceeds maximum allowed tokens. Output has been saved to …, so there was nothing left to slim. A host hook can reshape results that are merely wasteful, but not the ones that overflow; those have to be fixed at the server or the proxy. The hook also has to cope with the shape it receives: a JSON string, with a structured result wrapped as {"result": …}.

Every major framework has an equivalent interception point. In each, check two things before relying on it: whether it runs before or after the host's own limit, and whether it counts in tokens or characters.

• OpenAI Agents SDK (openai-agents 0.24.0): ToolOutputTrimmer(recent_turns=2, max_output_chars=500, preview_chars=200, trimmable_tools=...) as a call_model_input_filter. It trims older turns only and counts characters. It is Python-only; in JavaScript you write your own callModelInputFilter.
• LangChain: tool_interceptors on MultiServerMCPClient to rewrite each result, and ContextEditingMiddleware(edits=[ClearToolUsesEdit(...)]) for eviction. Two caveats: langchain-mcp-adapters 0.3.2 still requires mcp<2, and the default approximate counter is characters ÷ 4.
• Vercel AI SDK (ai 7.x): toModelOutput on a tool gives the model a shaped result while the app keeps the full one; prepareStep with pruneMessages drops old tool results between steps.
• Pydantic AI: MCPToolset(process_tool_call=...) wraps every call. Google ADK: after_tool_callback returns a replacement response dict, including for McpToolset tools.

Evicting results that were fine when they arrived, through Anthropic's clear_tool_uses_20250919, compaction or LangChain's SummarizationMiddleware, is a run-time policy rather than a code change; its settings, and what it does to the prompt cache, are covered in the runbook.

Step 8The library evaluation: what each one is for, and what to use when.

Counting first, because every budget above depends on it.

• tiktoken 0.14.0 (Python): encoding a 34KB JSON string with o200k_base takes about 1.5ms, cheap enough to run on every result. It is exact only for OpenAI models; there is no entry for the newest model names, so call get_encoding("o200k_base") explicitly.
• gpt-tokenizer 4.1.0 (JavaScript): pure TypeScript, actively released, with the early-exit isWithinTokenLimit. js-tiktoken 1.0.21 works but has not had a release since August 2025.
• Anthropic messages.count_tokens: free and rate-limited separately from message creation, but it is a network call, the docs describe its result as an estimate, and it rejects requests that use the MCP connector. There is no current local Claude tokenizer: @anthropic-ai/tokenizer 0.0.4 dates from 2023, and its README says it is no longer accurate. Anthropic also notes that Claude 4.7 and later models produce about 30% more tokens for the same text. The practical pattern is a local o200k_base count times a safety factor, calibrated offline with count_tokens and checked against usage.input_tokens.

Then the encoding. These are the same 30 projected records in four formats (labels joined into a string so every format carries the same information):

format            tokens   vs compact JSON
JSON, indent=2     6,361   +21%
JSON, compact      5,267   baseline
TOON               4,443   −16%
CSV                4,165   −21%

Format is a second-order lever. Minifying is free and always right. TOON (toon-format 1.2.0 in Python, @toon-format/toon 4.4.0 in TypeScript) saves a further 16% here and declares the row count in its header, which helps a model notice a short page. Its own README says compact JSON often wins on deeply nested or non-uniform data, and that CSV is smaller for purely flat data, which matches this measurement. Datadog reported the same pattern from its MCP server: CSV used "about half as many tokens" as JSON per record, YAML saved "around 20%" on nested data, and the biggest gain came from letting agents aggregate with SQL rather than page at all, with evals running "about 40% cheaper" (Datadog engineering). Compare that 16–21% with the 97% from projection: change the format after you have chosen the fields, not instead.

Putting it together:

• You own a Python server: the mcp 2.x MCPServer pattern above, with tiktoken for the budget and a middleware ceiling. No SDK paginates tool results for you; that part is always your code. For list-method pagination, fastmcp offers FastMCP(list_page_size=N), while MCPServer ignores the list cursor.
• You own a TypeScript server: @modelcontextprotocol/server 2.3.x with a zod outputSchema and gpt-tokenizer. The v1 @modelcontextprotocol/sdk line does not implement the 2026-07-28 revision.
• You don't own the server: a fastmcp proxy with Tool.from_tool, or mcp-compressor --shrinkfit if you want something packaged.
• You own the agent: the framework hook from STEP 7 for waste, and the server or proxy fix for overflow, because in Claude Code the hook fires after the limit.
• Always: a real tokenizer instead of characters ÷ 4, and a budget test against a fixture recorded from your largest account.