mcp-brasil源码深度解析：Async全链路、指数退避重试与BM25工具过滤如何实现

mcp-brasil是一个连接 AI 与70+ 巴西政府公开 API的 MCP Server（Model Context Protocol 服务器），覆盖 IBGE、巴西央行、众议院、参议院、透明度门户、DataJud 等权威数据源。对于想读懂它的开发者，本文从源码层面拆解三大核心机制：贯穿全链路的Async 异步架构、应对网络抖动的指数退避重试，以及解决 154+ 工具选择难题的BM25 工具过滤，帮你快速建立对这套项目的整体认知。

👆 上图是 mcp-brasil 的整体架构：上层是 MCP Client（Claude、Cursor、Open WebUI），中间是挂载了数十个 Feature 子服务器的根服务器，底部是分散的政府 API。本文就沿着这张图，把每个关键模块的源码讲透。

项目架构总览：70+ API 如何组织

mcp-brasil 采用"约定优于配置"的插件式设计。每个政府数据源都是一个Feature（功能模块），存放在src/mcp_brasil/data/下，目录结构高度统一：

新增一个数据源时，你不需要手动改根服务器。只要按约定放好文件，注册器就会自动发现并挂载。完整架构文档见 docs/concepts/architecture.md。

核心源码分布：

Async 全链路：从请求到响应

mcp-brasil 的"全链路"指一次工具调用从入口到出口全程异步，没有任何阻塞点。它由三层协同构成。

共享 HTTP 客户端：lifespan 如何管理连接

传统做法是每个请求临时建一个连接、用完即弃，开销大且浪费。mcp-brasil 用生命周期钩子在服务器启动时创建一个共享的httpx.AsyncClient，所有工具共用它，关闭时再统一释放。

关键在 src/mcp_brasil/_shared/lifespan.py：

@lifespan async def http_lifespan(server: FastMCP[Any]) -> AsyncIterator[dict[str, Any] | None]: client = httpx.AsyncClient( timeout=httpx.Timeout(HTTP_TIMEOUT), headers={"User-Agent": USER_AGENT, "Accept": "application/json"}, follow_redirects=True, ) try: yield {"http_client": client} # 工具通过 ctx.lifespan_context 取用 finally: await client.aclose()

这种"启动时建、关闭时拆"的模式，让连接池、超时、默认头都能全局复用，是 Async 架构的地基。

自动注册：FeatureRegistry 的发现机制

FeatureRegistry通过pkgutil.iter_modules()扫描mcp_brasil/data/下的每个子包，遵循一套约定：

1. 是带__init__.py的子包；
2. __init__.py导出一个FEATURE_META（声明名称、描述、是否需鉴权）；
3. server.py导出一个mcp（FastMCP 实例）。

满足条件即被挂载，并以 Feature 名做命名空间前缀（工具名变成ibge_buscar_*）。任一模块校验失败只会被跳过并记录日志，绝不会让整个服务器崩溃——见 src/mcp_brasil/_shared/feature.py。

异步批处理：asyncio.gather 并行执行

当一次任务需要多个相互独立的数据时，mcp-brasil 提供executar_lote工具，把多条查询用asyncio.gather()并行跑完，而不是串行等待。核心在 src/mcp_brasil/_shared/batch.py：

results = await asyncio.gather(*[_run_one(q) for q in queries])

单个查询失败不会影响其他查询——每个都会返回带错误信息的独立结果，最后拼成 Markdown。这让"对比两个议员的 2023/2024 年开销"这类任务能一次性并发完成。

指数退避重试：让 API 调用更稳健

政府 API 偶尔会抖动（5xx、限流 429、超时）。mcp-brasil 用"只重试可恢复错误 + 指数退避"策略来兜底，核心在 src/mcp_brasil/_shared/http_client.py。

重试哪些错误？_RETRYABLE_STATUS_CODES

代码里用一个冻结集合精确界定"值得重试"的状态码：

_RETRYABLE_STATUS_CODES = frozenset({429, 500, 502, 503, 504})

设计意图很清晰：4xx（除 429）是客户端自己的错，重试也救不了，直接抛错；只有 5xx 和限流这类"暂时性故障"才进入重试循环。

退避公式与 RateLimiter 限流

重试等待时间遵循经典的指数退避公式：

wait = HTTP_BACKOFF_BASE * (2 ** attempt) # 1s → 2s → 4s → 8s ... await asyncio.sleep(wait)

默认值来自 src/mcp_brasil/settings.py：HTTP_MAX_RETRIES=3（共 4 次尝试）、HTTP_BACKOFF_BASE=1.0，且都能用环境变量覆盖。指数增长避免了"雪崩式"地对故障端点反复轰炸，给上游留出恢复时间。

除了对外请求的退避，项目还有一个滑动窗口限流器src/mcp_brasil/_shared/rate_limiter.py，用asyncio.Lock+deque保证并发下的公平性：

limiter = RateLimiter(max_requests=80, period=60.0) async with limiter: await do_request()

当窗口内请求数达到上限时，它会精确算出"最老一条多久过期"，然后asyncio.sleep等待——绝不阻塞事件循环，是 Async 友好的限流实现。

BM25 工具过滤：154+ 工具如何精准匹配

mcp-brasil 挂了 70+ 个 API，展开后是154+ 个工具。如果一次性全塞给大模型，会撑爆上下文、也让模型"选择困难"。解决方案是BM25 工具过滤——一种经典的全文检索算法，按相关性给工具打分排序。

为什么需要工具过滤

在 src/mcp_brasil/server.py 里，默认TOOL_SEARCH="bm25"（见 settings.py）。它把list_tools替换成两个更省心的工具：

• search_tools：按关键词检索，返回Top 10最相关工具；
• call_tool：用检索结果里挑出来的工具去执行。

模型不再需要"看到全部再挑"，而是"先搜到相关的，再调用"。

BM25SearchTransform 工作原理

实现挂在根服务器上，通过 FastMCP 的 Transform 机制动态改写工具列表：

if TOOL_SEARCH == "bm25": from fastmcp.server.transforms.search import BM25SearchTransform mcp.add_transform( BM25SearchTransform( max_results=10, always_visible=_always_visible, ) )

两个设计要点值得注意：

• max_results=10：只暴露最相关的 10 个，把上下文压力降到最低；
• always_visible白名单：listar_features、recomendar_tools、planejar_consulta、executar_lote、listar_datasets_disponiveis这 5 个"导航型"元工具永远可见，保证模型任何时候都能"发现工具"，不会因为过滤而迷路。

除了 BM25，项目还支持实验性的CodeMode（search+get_tags+get_schemas）与关闭过滤（none）三种模式，用MCP_BRASIL_TOOL_SEARCH一键切换。配套的 LLM 推荐能力（recomendar_tools、planejar_consulta）分别位于 src/mcp_brasil/_shared/discovery.py 与 src/mcp_brasil/_shared/planner.py，可作为 BM25 的补充手段。

总结：三大机制如何协同

mcp-brasil 用一套克制而成熟的设计，把一个"多 API 聚合服务"做稳了：

1. Async 全链路——lifespan共享客户端 +asyncio.gather并行，全程无阻塞；
2. 指数退避重试—— 只重试 429/5xx，退避公式给上游留恢复时间，滑动窗口限流防雪崩；
3. BM25 工具过滤—— Top 10 检索 + 元工具白名单，让 154+ 工具不再"难选"。

想深入更多细节，可直接阅读 docs/concepts/architecture.md 与 docs/guide/development.md。这套"自动注册 + 异步 + 稳健重试 + 智能检索"的组合，也为其他多数据源 MCP 项目提供了不错的参考范本。