ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

六站合一MCP服务实践:从统一检索建模到工程落地

六站合一MCP服务实践:从统一检索建模到工程落地 上个月终于把憋了大半年的活儿给落地了把我手上六个独立站点——技术博客、产品文档、社区问答、资源下载、周刊归档、图片素材库——全部塞进了一个 MCP 服务。现在我在任何支持 MCP 的客户端里都能用一句自然语言同时查这六个站的内容再也不用逐个站点开搜索框翻半天了。这篇不写那些花哨的 Hello World Demo就讲讲这个“六合一”MCP 从建模、编码到排坑的完整过程。尤其是工具粒度和数据标准化这两块我前后推翻过两版设计踩了不少坑写出来给准备做类似内容聚合服务的同学一个参考。1. 六站塞进一个MCP到底图什么1.1 起点六个站的内容管理越来越失控先交代一下背景。我维护的这六个站技术栈完全是各搞各的博客是自研的 Python 应用文档站挂在 VuePress 上问答社区用的现成论坛程序资源下载站是纯静态页面加 JSON 索引周刊归档以前是手工维护的 Markdown 文件图片素材库则是对象存储加一张数据库表。内容加起来有几千篇文章、几万条问答、上千个下载资源。过去想找某个东西我的路径是先回忆它在哪个站再打开那个站的站内搜索翻两页列表点进去看详情。如果是一个跨站主题比如“某某框架在部署时的兼容性问题”我得在博客、文档、问答三个站各搜一遍然后自己在脑子里把结果拼起来。我一直想做个统一的站群搜索但一直没动手。原因很简单我只是想让 AI 在对话里顺便帮我把这些内容调出来并不想再维护一个搜索界面、做一堆前端页面。我需要的不是一个“搜索产品”而是一个能让 AI 直接访问内容的接口层。1.2 为什么不是搜索引擎、不是RSS而是MCP有人可能说你直接写一个聚合搜索页不行吗我认真考虑过有几个痛点让我放弃了六个站的数据分散在 MySQL、PostgreSQL、对象存储和纯静态文件里做聚合搜索页意味着要把所有数据同步到一个专门的搜索索引里等于又多了一个要维护的基础设施。我真正的诉求是配合 AI 使用。比如在 Cursor 里写文章时想引证一篇自己博客里两年前的观点或者在 Claude 里回答用户问题前想先查一下问答社区是否有人讨论过。这些场景都需要把搜索结果直接“递”给 AI而不是让 AI 再去打开一个网页。站内搜索各自为政搜索质量参差不齐。问答社区的搜索是第三方插件纯静态资源站干脆没有搜索功能。MCP 只需要我写一套工具封装所有站点的搜索能力就统一了。MCP 的价值不在于它是“最新技术”而在于它是目前接入面最广的协议。Claude Desktop、Cursor、Trae、Cherry Studio 这些客户端都原生支持我只要实现一次服务端所有客户端通吃。相比之下RSS 只能解决“订阅更新”解决不了“按关键词检索历史内容”的问题。1.3 目标定位一个只读的站群知识接口动手前我把目标收敛得很清楚这个 MCP 服务先做成只读的不提供任何写入能力。它能做到三件事在指定站点内搜索内容返回标题、摘要和链接获取某一篇内容的全文供 AI 后续分析、引用给出一份站群数据总览让 AI 知道当前各站的内容规模和最近更新情况。至于“AI 写文章自动发布到博客”“AI 回答问答社区的提问”这些读写闭环我放在第二阶段做。先解决读的问题把检索链路跑通再谈别的。一上来就做全双工调试成本会翻好几倍。2. 动手前想清楚MCP聚合服务怎么建模2.1 六站数据形态盘点建服务之前我先把六个站的数据访问方式盘了一遍画成一张表贴在工位旁边站点数据载体内容类型是否有现成搜索访问频率技术博客MySQL 单表文章正文、标签、发布时间有LIKE 查询高产品文档VuePress 编译后的 Markdown 文件文档章节、锚点、更新日志无中社区问答PostgreSQL 三张表问题、回答、标签有插件实现高资源下载静态 JSON 索引 OSS文件名称、描述、大小、下载链接无中周刊归档Markdown 文件夹每期主题、链接汇总、导读无低图片素材MySQL 单表 OSS图片标题、标签、缩略图地址无只按分类浏览中这步很重要。不是因为“要写文档”而是它直接决定了我要为每个站写什么样的适配器。博客和问答有现成数据库可以走 SQL 查询文档站和周刊其实是文件直接读文件系统做内容截取资源站和图片素材库本质上是“元数据存库里、文件本体在 OSS 上”MCP 只需返回元数据和访问链接。2.2 工具粒度设计一个站一套工具还是按能力横切工具粒度是这次设计里我改动最大的地方。第一版我把六个站拆成六组独立工具比如blog_search、docs_search、qa_search、res_search…… 听起来很直观。但实际用起来问题很大工具列表太长AI 在选择时要费劲翻而且当用户问一个跨站问题时AI 不知道先调哪个经常连续调用好几个工具试错既慢又费 token。后来我把粒度改成“按能力横切”只保留三类工具search_content(site, keyword, limit)在指定站点内搜索site 参数限定范围search_all(keyword, limit)跨六个站统一搜索每个站各返回少量结果get_content(site, item_id)获取指定站点某条内容的完整正文。工具少了AI 的决策路径就短了。用户说“帮我搜一下博客里关于 FastAPI 部署的文章”AI 会直接调用search_content(siteblog, keywordFastAPI 部署)。用户说“看看这些站里有哪些讲缓存策略的”AI 就会调用search_all。这个设计的教训是MCP 工具不是越多越好每个工具都会增加 AI 的决策成本。能用参数区分的就不要拆成独立工具。2.3 内容标准化从六套数据结构到统一Schema六站的数据结构各不相同如果直接返回原始结构AI 每次都要重新理解字段含义还容易解析出错。我给所有搜索结果定义了一个统一的输出 Schema{ site: blog, type: article, id: 2024-0912-fastapi-deploy, title: FastAPI 部署踩坑记录, summary: 总结 FastAPI 在 Docker 部署时的常见问题包括静态文件路径、环境变量传递……, url: https://blog.example.com/p/2024-0912, updated_at: 2024-09-12 }具体实现时我让六个适配器各自把原始数据映射成这个结构。博客的id是主键数字文档站的id是文件相对路径问答站的id是问题 ID资源站的id是 JSON 数组下标加文件名——这些细节都在适配器内部消化AI 看到的永远是同一套结构。摘要字段我花了点心思。对于正文很长的文章直接截取开头 200 字效果并不好因为很多文章的前言就是套话。我用了最朴素的办法优先取正文里匹配到关键词的句子前后文各 100 字如果没有命中关键词再退化为正文开头。实践证明这个方法在大多数内容场景下都够用还省去了做文本摘要模型的成本和延迟。3. 服务端落地基于FastMCP搭建聚合服务器3.1 选型与依赖实现上我没有从零写 MCP 协议栈直接用了mcpPython SDK 里的 FastMCP 封装。选它是因为异步支持好、装饰器注册工具很顺手而且自带 stdio 和 HTTP/SSE 两种传输模式调试时用 stdio部署后走 SSE切换成本极低。安装依赖pip install mcp[cli] fastmcp httpx aiomysql redis如果用 uv 管理项目也可以把依赖写进pyproject.toml后统一安装。我个人习惯在服务器上用 systemd 托管这个服务部署路径是固定的客户端配置里不需要写复杂命令。3.2 服务骨架与配置加载服务的骨架很简单。我建议把所有配置数据库连接、OSS 端点、站点白名单都通过环境变量注入不要硬编码在代码里。这样同一个服务代码在本地调试、线上运行都能用也方便以后再加一个站点只需要改配置和新增一个适配器。import os import json import asyncio from typing import Any from mcp.server.fastmcp import FastMCP mcp FastMCP( hex-site-hub, instructions( 聚合六个站点的统一检索入口blog(技术博客)、docs(产品文档)、 qa(社区问答)、res(资源下载)、weekly(周刊归档)、gallery(图片素材)。 调用工具时 site 参数必须使用这些代码。 ), ) MYSQL_DSN os.environ[MYSQL_DSN] REDIS_URL os.environ[REDIS_URL]instructions参数是个不起眼但很重要的细节。它会被注入到系统提示词里告诉 AI 有哪些站点、site 参数用什么枚举值。实测下来写好instructions之后AI 传非法站名的概率大幅下降。3.3 统一工具的实现逻辑核心的几个工具注册代码如下为了篇幅我做了精简去掉了具体的数据库查询细节重点展示数据流结构SITES { blog: {name: 技术博客, type: article}, docs: {name: 产品文档, type: doc}, qa: {name: 社区问答, type: qa}, res: {name: 资源下载, type: file}, weekly: {name: 周刊归档, type: newsletter}, gallery: {name: 图片素材, type: image}, } mcp.tool() async def search_content(site: str, keyword: str, limit: int 10) - str: 在指定站点内搜索内容。site 必须是 blog/docs/qa/res/weekly/gallery 之一。 if site not in SITES: return f未知站点: {site}, 可选站点: {, .join(SITES.keys())} results await search_in_site(site, keyword, limit) return json.dumps(results, ensure_asciiFalse, indent2) mcp.tool() async def search_all(keyword: str, limit: int 3) - str: 跨所有站点统一搜索, 每个站点最多返回 limit 条结果。 out [] for site in SITES: try: results await search_in_site(site, keyword, limit) out.extend(results) except Exception as e: out.append({ site: site, error: str(e), }) return json.dumps(out, ensure_asciiFalse, indent2) mcp.tool() async def get_content(site: str, item_id: str) - str: 获取指定站点某个条目的完整正文。item_id 由搜索接口返回。 if site not in SITES: return f未知站点: {site} item await load_full_content(site, item_id) if item is None: return 未找到对应内容 return json.dumps(item, ensure_asciiFalse, indent2)注意search_content的返回类型是str而不是dict或自定义 Pydantic 模型。一开始我也试过直接返回结构化对象FastMCP 也能处理但实测下来返回 JSON 字符串对 AI 更友好——它可以按自己习惯的节奏解析不容易因为字段类型校验失败而报错。这个取舍不一定适合所有人但在我这个只读聚合场景里实测最稳。3.4 六站适配器封装与缓存每个站一个适配器函数负责把站内原始数据映射成统一 Schema。这里以博客站为例async def search_in_site(site: str, keyword: str, limit: int) - list[dict]: if site blog: sql SELECT id, title, summary, url, updated_at FROM posts WHERE title LIKE %s OR summary LIKE %s OR content LIKE %s ORDER BY updated_at DESC LIMIT %s params (f%{keyword}%, f%{keyword}%, f%{keyword}%, limit) rows await query_mysql(sql, params) return [ { site: blog, type: article, id: str(r[id]), title: r[title], summary: r[summary], url: r[url], updated_at: r[updated_at].isoformat(), } for r in rows ] # 其他站点分支类似只是 SQL 或文件读取方式不同Redis 缓存我加在了search_in_site外层。同一个关键词在短时间内被重复搜索的场景很常见AI 经常会在一次对话里对同一话题多次追问缓存一下能明显降低数据库压力import redis.asyncio as aioredis redis_client aioredis.from_url(REDIS_URL) async def search_in_site_with_cache(site: str, keyword: str, limit: int) - list[dict]: cache_key fmcp:search:{site}:{keyword}:{limit} cached await redis_client.get(cache_key) if cached: return json.loads(cached) results await search_in_site(site, keyword, limit) await redis_client.set(cache_key, json.dumps(results, ensure_asciiFalse), ex300) return results缓存时间我设了 300 秒。对站点的内容检索来说5 分钟内的结果基本可以接受如果你的站点更新非常频繁可以缩短到 60 秒。注意缓存 key 一定要包含 limit否则不同条数请求会互相污染。3.5 传输模式与启动方式开发调试我用 stdio 模式配合客户端配置直接跑线上则用 SSE 模式方便运维管理和权限控制。if __name__ __main__: # 本地调试默认 stdio; 需要远程服务时改为 mcp.run(sse, host0.0.0.0, port8090) mcp.run()用 stdio 启动时客户端要负责拉起这个 Python 进程用 SSE 启动时客户端只需要连到 HTTP 地址。我在本地开发时习惯开 stdio因为可以直接在终端看到日志排查问题方便。4. 客户端接入从Claude Desktop到Cursor的实际配置4.1 Claude Desktop 配置Claude Desktop 的 MCP 配置写在claude_desktop_config.json里。macOS 路径是~/Library/Application Support/Claude/claude_desktop_config.jsonWindows 是%APPDATA%\Claude\claude_desktop_config.json。我的配置{ mcpServers: { hex-site-hub: { command: python, args: [/opt/hex-site-hub/server.py], env: { MYSQL_DSN: mysql://reader:password127.0.0.1:3306/site_hub, REDIS_URL: redis://127.0.0.1:6379/0 } } } }配置完成后重启 Claude Desktop在设置里能看到 MCP 服务连接状态。如果报错先看日志路径~/Library/Logs/Claude/mcp.log。4.2 Cursor 与 Trae 的接入方式Cursor 里在 Settings 搜索 “MCP”选 “Add new MCP Server”可以用 command 类型填同样的 stdio 命令。配置完在 Agent 对话里输入 “查询所有站点最近更新的内容”如果工具被正常加载AI 会自动调用search_all。Trae 的接入类似也支持自定义 MCP Server。我试用下来这些 IDE 类工具走的都是 stdio 模式所以只要你的命令能拉起 Python 进程就成。唯一要注意的是IDE 里的终端环境变量和系统 launchd 不一样PATH里没有你本地 Python 虚拟环境的路径时命令要写成python的绝对路径或者用uv run --directory /opt/hex-site-hub server.py。我踩过这个坑本地终端能跑一配进 Cursor 就报“spawn python ENOENT”。后来我把配置里的command改成了 Python 解释器的绝对路径{ mcpServers: { hex-site-hub: { command: /usr/bin/python3, args: [/opt/hex-site-hub/server.py] } } }4.3 权限控制谁可以读什么因为我做的是只读服务权限控制相对简单但不代表没内容要注意。数据库连接账号建议单独建一个只读账号只授权SELECT不要用 root 或主账号。我的 DSN 里用户名就叫reader密码也单独设置跟写入账号隔离。服务部署在只有本机客户端能访问的环境里。如果要用 SSE 模式开放给远程客户端必须在前面挂一层认证推荐用 OAuth 或 API Key。目前 MCP 对认证还没有完全统一的标准但我个人经验是生产环境远程调用一定要上认证否则等于把数据库的只读权限裸奔在网络上。5. 实测过程与踩坑记录5.1 第一次把六个站跑通的场景第一次整体跑通我用的测试问题是“博客里有没有聊过数据库连接池的坑顺便看看问答社区有人问过类似问题吗”AI 的过程大致是先调用search_all(keyword数据库连接池, limit3)返回里博客有一篇相关文章问答社区也有两条历史提问。它接着调用get_content(siteblog, item_id2023-0615-db-pool)拿到了全文然后整合成一段回答还附带了原文链接。让我比较惊喜的是AI 没有像第一版那样频繁地在工具之间试错基本是一两次调用就拿到了所有信息。这说明统一工具粒度加上清晰的instructions对降低调用错误率确实有效。第一版六组工具的时候同样的提问它常常先调用docs_search发现文档站没有相关内容再去调blog_search白白多花了几轮时间。5.2 三个绕不开的坑编码、超时、返回体过大坑一中文乱码。这个主要在 Windows 下出现。MCP 服务通过 stdio 和客户端通信默认编码不一致时AI 收到的 JSON 里中文全部变成乱码搜索功能基本不可用。解决方式是在脚本最开始强制 stdout 使用 UTF-8import sys sys.stdout.reconfigure(encodingutf-8) sys.stderr.reconfigure(encodingutf-8)或者在启动命令前加上环境变量PYTHONIOENCODINGutf-8。macOS 和 Linux 下一般没有这个问题但我建议还是写上算是防御性操作。坑二空关键词把数据库打满。有几次 AI 在用户没给明确关键词时直接把空字符串传进了搜索工具。LIKE %%这种查询会返回全表数据不仅慢还把上下文窗口直接塞爆。我在工具入口做了参数校验关键词为空时直接返回提示并且限制limit最大值为 20防止 AI 一次请求太多内容。如果某个站点内容量大这个限制可以再收紧。坑三返回体过大导致上下文截断。get_content一开始返回完整正文结果一篇文章动辄上万字AI 的上下文窗口很容易被撑满后面的回答质量明显下降。我调整成了搜索接口只返回摘要get_content默认只返回正文前 3000 字如果 AI 确实需要更长的内容再通过一个可选参数max_len增加。这个改动让整个服务的 usable 程度提升了一个台阶。5.3 性能优化连接池、信号量与合理缓存服务刚上线时我遇到过并发高的场景下 MySQL 连接数飙升。后来做了三件事用aiomysql的连接池限制最大连接数不超过 10用信号量控制同时进行的搜索任务数量防止 AI 一次性并发调用十几个搜索把查询压垮缓存穿透的场景单独处理热门关键词短时间被反复查询时直接走 Redis不落库。具体到连接池我在模块初始化时创建了一个全局池import aiomysql pool None async def get_pool(): global pool if pool is None or pool.closed: pool await aiomysql.create_pool( host127.0.0.1, userreader, passwordpassword, dbsite_hub, autocommitTrue, maxsize10, ) return pool这轮优化做完服务在 AI 密集调用场景下稳定多了。实际上我的访问量并不大但这个经验对任何接 MCP 的数据库型服务都有参考价值不设连接池上限迟早会因为某个异常查询把数据库打崩。6. 这套方案现在的形态和下一步的扩展空间6.1 当前运行状态和实际收益服务上线运行了一个多月目前六个站的检索请求都能稳定处理。对我自己来说最直观的收益是写东西的时候引证旧内容快了很多以前要翻好几个站现在一句话就能让 AI 把相关的旧文章、旧问答、旧文档都找出来还能直接引用原文链接。对其它项目来说这个模式还能复用。我后来又给一个内部使用的知识库做了类似的 MCP 服务只不过数据源从六个站变成了内部 Wiki 和需求文档代码结构完全不用大改只换了数据源适配器。6.2 后续计划从只读到可写从单服务到多智能体下一步我打算做两件事。第一件是把写入能力加上让 AI 可以调用create_weekly_draft之类的工具把从各站检索到的内容整理成周刊初稿写到一个固定的 GitHub 仓库。这个工具会严格限制输入输出且发布前仍然需要人审一遍。第二件是尝试把多个 MCP 服务串起来做多智能体场景让一个 MCP 负责站群检索另一个 MCP 负责外部信息获取再建立一个调度层让 AI 根据任务自动选择调哪个。目前我还在调研阶段因为智能体之间的状态传递和上下文隔离还没有完全想清楚。但至少这次“六合一”的经验已经把地基打好了。无论是数据标准化、工具粒度设计还是踩坑后的防御性编码这些经验在后续扩展时都能直接用上。最后再分享一个小技巧每加一个新站点时先手动测试search_content能否返回统一 Schema再交给 AI 去调用。这一步检查做扎实了后续基本不会出现返工的情况。
返回列表