代码里的 MCP 大结果:改造前后与选库

14 分钟读完

C14
深入解析 · MCP

三十个拉取请求,修复前要花 187,521 个 token,修复后只要 5,379 个。其中 97% 是投影的功劳;按 token 预算的分页、绑定查询的游标和测试,只是确保这个数字不再涨回去。

这是实现篇。协议层面的推理见 MCP 中的大结果与分页,宿主的各种限制和运维手册见 当 MCP 结果撑爆上下文,GitHub 的来龙去脉见 这篇案例研究。这里只有一个工具,先写得很糟,再一步步修好,对着的始终是同一份固定数据:github/github-mcp-server 最近的 30 个拉取请求,抓取于 2026-10-21。下文每个数字都是在这份文件上实测的,使用 tiktoken 0.14.0(o200k_base),环境为官方 Python SDK mcp 2.3.0、TypeScript SDK @modelcontextprotocol/server 2.3.1、fastmcp 4.1.3 和 claude-agent-sdk 0.2.170。写这篇的过程中,我在自己的初稿里发现了三个 bug;它们被原样留在了故事里,因为你多半也会写出同样的 bug。

步骤 1改造前:一个直通式工具,以及它的 token 实际花在了哪里。

大多数服务端最先上线的就是这个样子:调用上游 API,原样返回它的结果。它甚至还有分页——per_page 和 page,和 REST API 接受的参数一模一样。

# 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

用默认参数调用一次,返回 656,124 个字符,也就是 187,521 个 token,是 Claude Code 25,000 token 上限的七倍半。通过 Claude Agent SDK 运行时,这个结果被写进了文件而没有进入上下文;而测试会话只被允许使用这一个工具,打不开那个文件,最终一直没有给出回答。线路上的实际数据量还要再翻一倍。由于函数标注了 -> str,MCPServer 会发布一个 {"result": string} 形式的 outputSchema,并把整份数据作为 structuredContent 再发一遍:每次调用都是 656,124 个字符的文本,外加 694,798 个字符的结构化副本。Claude Code 报告的 694,797 个字符,对应的正是第二份副本。fastmcp 的做法也一样。返回现成 JSON 字符串的工具应当用 @mcp.tool(structured_output=False) 明确声明,这样 schema 和副本都会去掉;想要结构化输出的工具,则应当有意识地只返回一次(见 STEP 3)。

分页也无济于事,因为体积出在单条记录上,而不是记录条数上。按字段拆解这份数据,就能看出钱花在哪:

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%

将近一半的数据是仓库对象:每个拉取请求都带着两份完整副本,一份在 head 下,一份在 base 下。没有哪个模型需要它们,也没有哪种页大小能把它们去掉。所以第一步修复是投影,而不是分页。

要用分词器计数,而不是 len(text) // 4。在这份数据上,字符数 ÷ 4 得到 164,031,少算了 12%。在扁平的表格式 JSON、CSV 或 TOON 上,字符数 ÷ 4 会少算三分之一到一半,因为数字和标点切分出的 token 很密。用这个经验公式算出的预算,恰恰会在最要紧的地方超标。

步骤 2先做投影:明确列出要返回的字段,并给自由文本设上限。

# 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,
    }

把 project() 应用到全部 30 条记录上,数据量从 187,521 个 token 降到 5,297(−97%)。它体现了三条规则。字段是你自己选定的白名单,这样上游 API 新增字段时,你的结果不会悄无声息地变大。嵌套对象收缩成智能体真正会用的那一个值:user 变成登录名,head 变成分支名。自由文本只给预览,并附一个标志说明它被截断了;完整文本只需再调用一次 get_pull_request(见 STEP 3)就能拿到,因此要保留后续调用所需的标识符。

如果调用方确实需要不同的字段,就让它们自己提出来,从最安全到最强大依次是:一个对照白名单校验的 fields: list[str] 参数;一个 JMESPath 表达式(Python 用 jmespath 1.1.0,JavaScript 用 @jmespath-community/jmespath),它没有副作用,可以放心接受模型传来的值;或者一个 jq 过滤器(jq 1.13.0 绑定),它是图灵完备的,需要限制 CPU 和执行时间。三种做法在生产中都有先例:Stainless 生成的服务端会给每个工具加一个 jq_filter 参数,Zscaler 的服务端则在其列表类工具上接受一个可选的 JMESPath query。

步骤 3再按 token 分页,游标绑定到它所属的查询——而且要数你实际发出去的,而不是你以为会发出去的。

# 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"]}

循环在下一条记录会超出预算时结束本页,从不按记录条数截断;而且每页至少发出一条记录,这样一条超大的记录也不会造成空页和一个永远不前进的游标。游标是不透明的 base64,但里面带着生成它的那个查询的哈希,所以一旦有人拿着游标换了过滤条件再用,调用就会失败,并给出一条告诉对方该怎么做的消息,而不是返回错误的行。这个结果分两页走完 30 条记录,分别是 3,851 和 1,528 个 token,结果本身就带着 truncated、total_matching、next_cursor 和一行说明。

这段代码里有两处之所以这样写,是因为初稿写错了,而 STEP 4 的测试把两处都抓了出来。

  • SDK 会把你的结果格式化缩进。初稿直接返回 Pydantic 模型,让 MCPServer 去序列化。它渲染文本块时用的是 indent=2(func_metadata.py 里的 pydantic_core.to_json(result, fallback=str, indent=2)),于是一页预算 4,000 个 token 的数据,发出去时变成了 4,838。改为返回一个 CallToolResult、文本由你自己序列化,问题就解决了;而 Annotated[CallToolResult, PullRequestPage] 这个返回类型,保住了发布出去的 outputSchema 以及对 structured_content 的校验。
  • 信封也是页的一部分。去掉缩进之后,这一页仍有 4,016 个 token:预算算了记录,却忘了游标、说明和各个标志。ENVELOPE 就是给它们预留的空间。
  • 只有 ToolError 能传到模型那里。初稿遇到坏游标时抛的是 ValueError。MCPServer 会把 ToolError 以外的任何异常都当作崩溃处理,只发出 Error executing tool list_pull_requests,于是消息里的补救办法根本没送到。TypeScript SDK 会把普通 Error 的消息原样透传;Python 不会。

文本块是结构化结果的紧凑副本,不是摘要。规范规定,返回结构化内容的工具 SHOULD 同时把序列化后的 JSON 作为文本返回,而各家客户端对模型读的是哪一份意见不一:langchain-mcp-adapters 把文本发给模型,把 structuredContent 留作模型永远看不到的附件;OpenAI Agents SDK 则默认 use_structured_content=False,恰恰是因为大多数服务端都会镜像一份。镜像出一个紧凑的文本块,是唯一能让每个客户端都读对的选择,也正是预算所度量的那一份。

同一份契约在 TypeScript SDK 上是逐行移植,只有工具体不同:

  }, 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 };

它同样分页走完这 30 条记录,两页分别是 3,844 和 1,524 个 token。gpt-tokenizer 4.1.0 用纯 TypeScript 提供了 o200k_base;它的 isWithinTokenLimit(text, limit) 一旦超过限制就停止编码,用来做大小检查比把整个字符串编码完更合适。

步骤 4测试契约,而不是测试工具:五条断言,页一变大就失败。

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

这些测试在进程内运行(即 测试 MCP 服务器里的 Client(server) 模式),耗时 0.3 秒。每一条守住一种不同的故障。预算测试读取的是线路上的文本块,所以它抓到了缩进 bug。遍历测试证明游标既不会漏掉、也不会重复任何一条记录。带内测试会在有人把标志挪进日志时失败。外来游标测试检查错误文本是否真的送到了客户端,所以它抓到了 ValueError。最后一条把预算设得比一行记录还小,断言仍然会发出一页。

预算测试要对着从你最大的真实账户录制的 fixture 跑,而不是合成数据。这里的 fixture 只是 30 个公开的拉取请求;真正会击穿预算的输入,是某个 PR 描述动辄 4,000 行的客户,而除非你把它录下来,否则你手里根本没有这样的数据。

步骤 5为还没人改造的工具设一个全服务端的上限。

逐个工具定契约是要花功夫的,一个有 40 个工具的服务端不可能一次全部改完。中间件里的一道上限可以兜住其余的工具:它度量每一个 tools/call 结果,把超大的那个变成模型可以据此行动的错误。

# 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.")])

用 mcp.middleware.append(result_guard) 注册到 STEP 1 的服务端上之后,那个 187,521 token 的结果就变成了一句话:"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." 有两个细节很重要。替换结果必须是 CallToolResult 模型;普通 dict 在 2026-07-28 修订版下会校验失败,因为它没有 resultType。另外,中间件 API 在 mcp 2.x 中被标为临时性的(「签名可能在 2.x 的某个次版本中改变」),所以要锁定 SDK 版本,并保留这条测试。

如果你的服务端用的是 fastmcp,内置的对应物是 ResponseLimitingMiddleware(max_size=..., tools=[...]),在依赖它之前值得先了解它的行为。它统计的是整个序列化结果的字节数,而不是 token;那份 656KB 的数据量出来是 1,389,809 字节,因为它把 STEP 1 里的文本块和重复的 structuredContent 都算进去了,而且每一份里的引号都被转义了。超过限制时,它会截断而不是拒绝:设 max_size=60_000 时,模型收到的是 16,530 个 token 的 JSON 数组,在字符串中间被切断,后面跟着 [Response truncated due to size limit]。它还会丢掉 structuredContent,并从它所限制的每个工具上移除 outputSchema。它是一道可靠的兜底,却是一份糟糕的契约;像上面那样拒绝并给出补救办法,通常好过塞给模型半份文档。

步骤 6在代理层改造一个不归你管的服务端。

智能体见到的超大结果,大多来自你无法修改的第三方服务端。fastmcp 可以作为代理挡在它前面,用塑形过的版本替换它的工具,而不必碰上游代码:

# 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

网关只暴露 list_pull_requests_slim,全部 30 个拉取请求只用 1,763 个 token 就返回了。forward() 调用原工具,disable() 把原工具从 tools/list 中隐藏,字节限制则留在后面当兜底。唯一的代价是代理仍然要接收完整的上游数据;它省下的是模型的上下文,而不是网络流量。

现成的替代品也有,成熟度却相差很大。atlassian-labs/mcp-compressor(v0.38.0,Apache-2.0,有 Python、TypeScript 和 Rust 版本)维护得最好:它主要压缩工具目录,可选的 --shrinkfit 标志会缩减超过 20,000 个字符的结果,同时保留顶层结构,并允许调用方用 full=true 请求完整结果。mcp-trunc-proxy(0.2.2,GPL-3.0)以通用方式实现了句柄模式,把大结果卸载到存储里,返回一份预览外加一个取回工具,但它只是个小项目。搜索能找到的其余项目(mcp-rtk、mcpslim、mcpsqueeze、mcp-context-proxy)在 2026-10-21 时 GitHub star 数只有零到四个。这些项目宣称的节省量都没有在这里复现过,请把它们当作作者自己的说法。

步骤 7宿主侧钩子:各个智能体框架允许你在哪里改写结果——以及那个触发得太晚的钩子。

如果你构建的是智能体而不是服务端,就可以在结果到达时对它塑形。在 Claude Agent SDK 中,PostToolUse 钩子可以在模型读取之前替换 MCP 工具的输出:

# 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])]},
)

在一个两条 PR 的结果(约 43,000 个字符)上,钩子的表现与文档一致:问模型第一条记录有哪些键,它答的是 number, title, state, merged_at,正是钩子保留的那四个。在 30 条 PR 的结果上,它什么也没做,而原因文档里没写。大小限制在钩子之前就已生效。等到 slim_mcp_result 触发时,tool_response 已经是字符串 Error: result (694,797 characters) exceeds maximum allowed tokens. Output has been saved to …,根本没东西可瘦身了。宿主钩子能改造那些仅仅是浪费的结果,却改不了已经溢出的结果;后者必须在服务端或代理层修。钩子还得应付它收到的数据形态:一个 JSON 字符串,其中结构化结果被包成 {"result": …}。

每个主流框架都有对应的拦截点。在依赖其中任何一个之前,都要先确认两件事:它在宿主自己的限制之前还是之后运行,以及它按 token 还是按字符计数。

  • OpenAI Agents SDK(openai-agents 0.24.0):把 ToolOutputTrimmer(recent_turns=2, max_output_chars=500, preview_chars=200, trimmable_tools=...) 用作 call_model_input_filter。它只裁剪较早的轮次,并按字符计数。它只有 Python 版;在 JavaScript 里,你得自己写 callModelInputFilter。
  • LangChain:在 MultiServerMCPClient 上用 tool_interceptors 改写每个结果,用 ContextEditingMiddleware(edits=[ClearToolUsesEdit(...)]) 做驱逐。两点注意:langchain-mcp-adapters 0.3.2 仍然要求 mcp<2,而且默认的近似计数器就是字符数 ÷ 4。
  • Vercel AI SDK(ai 7.x):工具上的 toModelOutput 给模型一份塑形后的结果,应用自己保留完整结果;prepareStep 配合 pruneMessages 在步骤之间丢弃旧的工具结果。
  • Pydantic AI:MCPToolset(process_tool_call=...) 包住每一次调用。Google ADK:after_tool_callback 返回一个替换用的响应 dict,对 McpToolset 工具同样有效。

对那些到达时并无问题的结果做驱逐——无论是通过 Anthropic 的 clear_tool_uses_20250919、压缩,还是 LangChain 的 SummarizationMiddleware——属于运行时策略,而不是代码改动;它的配置以及它对提示缓存的影响,见运维手册。

步骤 8库的评估:每一个是干什么的,以及什么时候用哪个。

先说计数,因为上面所有的预算都依赖它。

  • tiktoken 0.14.0(Python):用 o200k_base 编码一个 34KB 的 JSON 字符串约需 1.5ms,便宜到可以对每个结果都跑一遍。它只对 OpenAI 模型是精确的;最新的模型名没有对应条目,所以要显式调用 get_encoding("o200k_base")。
  • gpt-tokenizer 4.1.0(JavaScript):纯 TypeScript,发布活跃,带有可提前退出的 isWithinTokenLimit。js-tiktoken 1.0.21 也能用,但自 2025 年 8 月以来就没有发布过新版本。
  • Anthropic messages.count_tokens:免费,且与创建消息分开限流,但它是一次网络调用,文档把它的结果称为估算值,而且它会拒绝使用了 MCP connector 的请求。目前没有可用的本地 Claude 分词器:@anthropic-ai/tokenizer 0.0.4 是 2023 年的东西,它的 README 也说它已经不准确了。Anthropic 还指出,Claude 4.7 及之后的模型对同样的文本会多产生约 30% 的 token。实用的做法是用本地 o200k_base 的计数乘一个安全系数,离线用 count_tokens 校准,再对照 usage.input_tokens 核验。

然后是编码格式。下面是同样 30 条投影后的记录在四种格式下的结果(labels 拼成一个字符串,让每种格式承载的信息完全相同):

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

格式是二阶杠杆。压缩空白不花任何代价,而且永远是对的。TOON(Python 用 toon-format 1.2.0,TypeScript 用 @toon-format/toon 4.4.0)在这里又省了 16%,并在表头里声明行数,这有助于模型注意到某一页不完整。它自己的 README 说,在深度嵌套或不规整的数据上,紧凑 JSON 往往更胜一筹;而对纯扁平数据,CSV 更小——这与这次测量吻合。Datadog 从自家 MCP 服务端报告了同样的规律:每条记录用 CSV 只需「大约一半的 token」,YAML 在嵌套数据上省了「约 20%」,而最大的收益来自让智能体用 SQL 做聚合、干脆不分页,评测跑起来「便宜了约 40%」(Datadog engineering)。拿这 16–21% 和投影带来的 97% 比一比:格式要在选好字段之后再换,而不是用它来代替选字段。

综合起来:

  • 你拥有一个 Python 服务端:用上面的 mcp 2.x MCPServer 模式,用 tiktoken 算预算,再加一道中间件上限。没有哪个 SDK 会替你给工具结果分页;这部分永远是你自己的代码。至于列表方法的分页,fastmcp 提供了 FastMCP(list_page_size=N),而 MCPServer 会忽略列表游标。
  • 你拥有一个 TypeScript 服务端:用 @modelcontextprotocol/server 2.3.x,配一个 zod outputSchema 和 gpt-tokenizer。v1 的 @modelcontextprotocol/sdk 系列没有实现 2026-07-28 修订版。
  • 服务端不归你管:用带 Tool.from_tool 的 fastmcp 代理;想要现成的东西,就用 mcp-compressor --shrinkfit。
  • 你拥有智能体:对付浪费,用 STEP 7 里的框架钩子;对付溢出,在服务端或代理层修,因为在 Claude Code 中,钩子是在限制之后才触发的。
  • 无论如何:用真正的分词器,而不是字符数 ÷ 4;再加一条对着从你最大账户录制的 fixture 运行的预算测试。

本页每一段代码都按所示方式实际跑过:这些摘录来自通过了测试的文件,合起来约 340 行,小到可以直接复制进仓库当模板。每个选择背后的设计推理见 MCP 中的大结果与分页;更通用、与协议无关的结果塑形规则见 塑形工具返回值。