fluxio-mcp 开发手记:从需求到开源#
这篇文章记录 fluxio-mcp 从想法到开源的完整过程——一个给 LLM 提供"可编程信息获取能力"的 MCP Server。为什么做它、怎么设计、测试怎么覆盖、开源踩了什么坑。
一、为什么做#
背景:做信息流工具(Fluxio)时,我发现一个真实痛点——
LLM 的通用搜索能力(如果提供)和"信息流"差很远:
需求成型:
定位:不是搜索,是"读取器"——LLM 已经有链接,需要的是把链接变成可读的正文。
二、架构设计#
设计决策:
| 决策 |
原因 |
| FastMCP |
装饰器极简、中间件、Python 生态 |
| httpx 并发 |
批量场景 IO 密集,并发是核心收益 |
| 可配置 UA |
部分站点反爬,UA 是基本尊重 |
| 超时 + 重试 |
网络不可靠,工具不能挂 |
| 输出纯 Markdown |
LLM 消费友好(HTML 会污染上下文) |
三、核心实现#
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
@mcp.tool()
async def fetch_urls(urls: list[str]) -> list[dict]:
"""批量抓取 URL 内容,返回标题+正文(Markdown)。
Args:
urls: 要抓取的 URL 列表(最多 10 个)
"""
async with httpx.AsyncClient(
timeout=15,
headers={"User-Agent": UA},
follow_redirects=True,
) as client:
results = await asyncio.gather(
*(fetch_one(client, u) for u in urls[:10]),
return_exceptions=True,
)
return [normalize(r) for r in results]
|
关键点:
- 并发:
asyncio.gather 10 个 URL 并发,总耗时 ≈ 最慢的一个
- 容错:单个失败不影响整体(
return_exceptions)
- 上限:
urls[:10] 防滥用(工具级限流)
- 降级:正文提取失败 → 返回元数据(title/url),不让 LLM 拿空结果
四、测试怎么覆盖#
开源项目没有测试 = 没有可信度。fluxio-mcp 的测试设计:
1
2
3
4
5
6
7
8
|
# tests/test_fetch.py
class TestFetchUrls:
async def test_single_url(self): ...
async def test_multiple_urls(self): ... # 并发正确性
async def test_invalid_url(self): ... # 容错
async def test_timeout_fallback(self): ... # 超时降级
async def test_max_limit(self): ... # 数量上限
async def test_markdown_extraction(self): ... # 正文提取质量
|
测试策略:
CI:GitHub Actions,push 自动跑 pytest——开源项目的门面是 CI 绿。
五、开源的坑#
坑 1:README 是门面,别随便写#
坑 2:License 必须先定#
坑 3:示例比文档有用#
1
2
3
|
# README 里给"10 秒跑起来"的示例
uvx fluxio-mcp 或 clone 后 python -m fluxio_mcp
# 让用户 30 秒内看到效果,比 1000 字文档有用
|
坑 4:版本与发布#
六、收获#
fluxio-mcp 的核心认知:
- 开源项目的起点是"自己的真实痛点"——不是追热点,是解决问题
- 设计决策要写出来:为什么 FastMCP、为什么并发、为什么 Markdown 输出
- 测试是开源的门面:CI 绿 + 边界覆盖 = 可信
- README/License/示例是开源的"交付物":代码好但没人用 = 白写
开源不是"把代码放网上",是"把问题、决策、过程一起交付"。 过程的价值不比代码低。
微信公众号「福清而不淡」
后端架构 · AI 工程 · 云原生的一线实践,扫码关注,不错过更新。
本文已同步发布到公众号,欢迎留言交流。