AI智能体项目实战记录（23）：MCP 使用教程之 创建 MCP Server，把知识库调用暴露成 MCP Tool

AI智能体项目实战记录（23）：MCP 使用教程之 创建 MCP Server，把知识库调用暴露成 MCP Tool

1. 引言

前两篇做的是MCP Client：

• 第 21 篇：stdio / Streamable HTTP 客户端；
• 第 22 篇：把 MCP 工具动态注册到主管智能体。

这一篇反过来：把当前系统里的search_knowledge_base暴露成一个MCP Server，让 IDE、Claude Desktop、其它 Agent 或业务系统都能通过 MCP 调用知识库。

外部客户端（IDE / Claude / 其它 Agent） │ ▼ MCP Server: knowledge-base │ ▼ RAGTool.search_knowledge_base │ ├─ bge embedding ├─ Milvus 检索 └─ DeepSeek 生成答案

目标：

1. 创建一个 MCP Server；
2. 把search_knowledge_base注册成 MCP Tool；
3. 支持 stdio 和 SSE 两种运行方式；
4. 验证工具发现与调用；
5. 说明鉴权、超时和部署注意事项。

2. 设计

2.1 对外暴露的工具

MCP 客户端看到的 schema：

{"type":"object","properties":{"question":{"type":"string","title":"Question"}},"required":["question"]}

2.2 传输方式

mcp 1.4.1的FastMCP支持：

FastMCP 1.4.1还没有 Streamable HTTP 服务端；需要该传输时要升级 SDK，但会牵动pydantic版本，当前不采用。

2.3 安全边界

search_knowledge_base会读取内部知识库，属于敏感能力：

• stdio 只在本机进程间通信，风险最低；
• SSE 默认只绑定127.0.0.1，不要直接暴露公网；
• FastMCP 1.4.1 的 SSE 没有内置鉴权，远程访问要在前面加 Nginx / API 网关做认证和限流；
• 建议给 MCP Server 单独设置资源限制和超时。

3. 目录与配置

3.1 目录

app/ ├── mcp/ │ ├── client.py # MCP 客户端（第 21 篇） │ ├── tool.py # MCP 工具包装（第 22 篇） │ └── knowledge_server.py # 本文新增：知识库 MCP Server

3.2 依赖

FastMCP已经包含在mcp==1.4.1中，不需要新增依赖。

知识库能力依赖项目现有配置：

MILVUS_URL / MILVUS_USER / MILVUS_PASSWORD MILVUS_COLLECTION_DOC EMBEDDING_MODEL_PATH DEEPSEEK_API_KEY

3.3 新增配置

# 对外 MCP Server：把知识库检索能力暴露出去（stdio / sse） MCP_KNOWLEDGE_SERVER_NAME="knowledge-base" MCP_KNOWLEDGE_SERVER_HOST="127.0.0.1" MCP_KNOWLEDGE_SERVER_PORT=18765

对应settings.py：

mcp_knowledge_server_name:Optional[str]='knowledge-base'mcp_knowledge_server_host:Optional[str]='127.0.0.1'mcp_knowledge_server_port:int=18765

注意区分：

• MCP_TRANSPORT是本项目作为客户端连接外部 Server 的方式；
• --transport是本文的 Server 进程使用哪种传输。

两者互不影响。

4. 编写 MCP Server

文件：app/mcp/knowledge_server.py

4.1 创建 FastMCP 实例

frommcp.server.fastmcpimportFastMCPfromapp.configimportsetting mcp=FastMCP(name=setting.mcp_knowledge_server_nameor"knowledge-base",instructions=("提供本地知识库检索问答能力。""当用户询问公司内部文档、政策法规、产品手册、业务流程等非公开知识时，""调用 search_knowledge_base 获取答案。"),host=setting.mcp_knowledge_server_hostor"127.0.0.1",port=setting.mcp_knowledge_server_portor18765,)

host/port只在 SSE 模式生效；stdio 模式不使用。

4.2 注册工具

@mcp.tool(name="search_knowledge_base",description=("检索本地文档知识库并生成答案。""适用于招商政策、公司产品、业务流程、内部制度等专业问题。"),)asyncdefsearch_knowledge_base(question:str)->str:""" 检索本地知识库并回答用户问题。 Args: question: 用户原始问题 """# 延迟导入：避免 stdio 启动时就加载 embedding 模型fromapp.tools.rag_toolimportRAGToolreturnawaitasyncio.to_thread(RAGTool()._run,question)

两个关键点：

1. 延迟导入RAGTool
rag_tool.py导入时会加载embedding_service，进而加载 bge 模型。放在函数内部，可以让tools/list快速返回，不必等模型加载。

2. asyncio.to_thread
RAGTool._run是同步方法，内部有 embedding、Milvus、DeepSeek 调用。放到线程里执行，避免阻塞 MCP Server 的事件循环，尤其是 SSE 模式下多个客户端并发时。

4.3 入口与传输选择

defmain()->None:parser=argparse.ArgumentParser(description="Knowledge base MCP Server")parser.add_argument("--transport",choices=["stdio","sse"],default="stdio",help="stdio：本地子进程；sse：HTTP SSE 服务",)args=parser.parse_args()logger.info(f"[mcp-server] starting name={mcp.name}transport={args.transport}"f"host={setting.mcp_knowledge_server_host}port={setting.mcp_knowledge_server_port}")mcp.run(transport=args.transport)if__name__=="__main__":main()

启动日志走loguru的 stderr，不会污染 stdio 的协议输出。

5. 运行 Server

5.1 stdio 模式

python-mapp.mcp.knowledge_server--transportstdio

这个进程不会主动输出内容，而是等待 MCP Client 通过 stdin/stdout 发 JSON-RPC。

适合：

• Claude Desktop；
• Cursor / VS Code 等 IDE；
• 本项目的MCPClient。

5.2 SSE 模式

python-mapp.mcp.knowledge_server--transportsse

默认监听：

http://127.0.0.1:18765 GET /sse POST /messages/

修改地址：

MCP_KNOWLEDGE_SERVER_HOST="0.0.0.0" MCP_KNOWLEDGE_SERVER_PORT=18765

0.0.0.0只建议在内网或前面有网关认证时使用。

5.3 预热（可选）

首次tools/call会加载 embedding 模型，可能比较慢。可以启动时加--preload：

python-mapp.mcp.knowledge_server--transportstdio--preloadpython-mapp.mcp.knowledge_server--transportsse--preload

--preload会先导入rag_tool，触发 embedding 模型和 Milvus 初始化。启动会变慢，但第一次工具调用会快很多。

6. 客户端接入

6.1 Claude Desktop / IDE（stdio）

配置示例：

{"mcpServers":{"knowledge-base":{"command":"<python解释器绝对路径>","args":["-m","app.mcp.knowledge_server","--transport","stdio"],"cwd":"<chat_py_server 项目根目录>"}}}

要点：

• command必须是能导入项目依赖的 Python；
• cwd必须是app.py所在目录，否则-m app.mcp...找不到模块；
• 不建议用python这种不确定的解释器，写绝对路径最稳。

6.2 MCP Inspector（SSE）

先启动：

python-mapp.mcp.knowledge_server--transportsse

Inspector 连接：

http://127.0.0.1:18765/sse

6.3 本项目 MCPClient（stdio）

如果想让当前项目作为客户端连这个 Server，把.env的客户端配置改成：

MCP_TRANSPORT="stdio" MCP_SERVER_COMMAND="<python解释器绝对路径>" MCP_SERVER_ARGS='["-m","app.mcp.knowledge_server","--transport","stdio"]' MCP_WORKSPACE_DIR="mcp_workspace"

注意：这会替换掉原来的文件系统 Server 配置。两者只能选一个；如果要同时接多个 Server，需要把客户端扩展成多 Server 配置。

7. 验证

7.1 tools/list（stdio）

tool count: 1 search_knowledge_base schema: {"properties":{"question":{"type":"string"}},"required":["question"]}

7.2 tools/list（SSE）

tool count: 1 search_knowledge_base

说明 Server 启动、协议握手、工具发现都正常。

7.3 使用项目内 SSE 测试脚本

项目已提供：

scripts/test_mcp_knowledge_sse.py

#!/usr/bin/env python# -*- coding: utf-8 -*-# @File : test_mcp_knowledge_sse.py# @Desc : SSE 方式测试知识库 MCP Server## 使用前先启动 Server：# python -m app.mcp.knowledge_server --transport sse --preload## 只测 tools/list：# python scripts/test_mcp_knowledge_sse.py## 测试 tools/call：# python scripts/test_mcp_knowledge_sse.py --question "知识库里有哪些招商政策？"importargparseimportasyncioimportsysfrommcpimportClientSessionfrommcp.client.sseimportsse_clientdef_unwrap_exception(exc:BaseException)->BaseException:"""展开 anyio/ExceptionGroup，拿到真正的错误"""current=excwhilegetattr(current,"exceptions",None):current=current.exceptions[0]returncurrentdef_content_to_text(result)->str:"""把 MCP 工具返回的 content 拼成可读文本"""parts=[]foriteminresult.contentor[]:text=getattr(item,"text",None)iftextisnotNone:parts.append(text)else:parts.append(str(item))return"\n".join(parts)asyncdefrun(url:str,question:str,timeout:int)->None:asyncwithsse_client(url,timeout=15,sse_read_timeout=timeout,)as(read,write):asyncwithClientSession(read,write)assession:awaitsession.initialize()tools=awaitsession.list_tools()print(f"[tools] count={len(tools.tools)}")fortoolintools.tools:print(f"-{tool.name}:{tool.description}")ifnotquestion:print("[call] 未传 --question，仅测试 tools/list")returnprint(f"[call] search_knowledge_base question={question}")result=awaitsession.call_tool("search_knowledge_base",{"question":question},)print(f"[call] isError={result.isError}")print(_content_to_text(result))defmain(question:str="")->None:parser=argparse.ArgumentParser(description="Knowledge base MCP Server SSE 测试")parser.add_argument("--url",default="http://127.0.0.1:18765/sse",help="SSE 地址，默认 http://127.0.0.1:18765/sse",)parser.add_argument("--question",default=question,help="要测试的问题；不传则只测试 tools/list",)parser.add_argument("--timeout",type=int,default=600,help="工具调用读取超时（秒），首次加载模型建议 600",)args=parser.parse_args()ifsys.platform=="win32":asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())try:asyncio.run(run(args.url,args.question,args.timeout))exceptExceptionase:# noqa: BLE001root=_unwrap_exception(e)print(f"[error]{type(root).__name__}:{root}",file=sys.stderr)sys.exit(1)if__name__=="__main__":main(question="知识库里有没记录当前项目使用了什么技术栈，是怎么样的架构？")

先启动 Server：

cd e:\myProject\my-langchain\agent\chat_py_server D:\anaconda3\envs\test01\python.exe-m app.mcp.knowledge_server--transport sse--preload

只测工具发现：

D:\anaconda3\envs\test01\python.exe scripts\test_mcp_knowledge_sse.py

预期：

[tools] count=1 - search_knowledge_base: 检索本地文档知识库并生成答案。适用于招商政策、公司产品、业务流程、内部制度等专业问题。 [call] 未传 --question，仅测试 tools/list

测试工具调用：

D:\anaconda3\envs\test01\python.exe scripts\test_mcp_knowledge_sse.py--question"知识库里有没记录当前项目使用了什么技术栈，是怎么样的架构？"

预期：

[tools] count=1 - search_knowledge_base: ... [call] search_knowledge_base question=知识库里有没记录当前项目使用了什么技术栈，是怎么样的架构？ [call] isError=False 根据知识库内容……

脚本参数：

脚本做了三件事：

1. 建立 SSE 连接并initialize；
2. 调用tools/list，打印工具名和描述；
3. 如果传了--question，调用search_knowledge_base并打印结果。

脚本只适用于 SSE；stdio 用前面的MCPClient脚本或 MCP Inspector。

7.4 tools/call

调用参数：

{"question":"知识库里有哪些公司产品？"}

执行链路：

MCP tools/call → search_knowledge_base → RAGTool._run → embedding_service 加载 bge 模型 → milvus_service.search → DeepSeek 生成答案

注意首次调用耗时：

• RAGTool延迟导入，第一次调用才加载 embedding 模型；
• 模型加载 + Milvus 连接 + DeepSeek 生成，首次可能达到数分钟；
• 客户端超时（如本项目MCP_TIMEOUT_SECONDS=60）可能先触发，表现为TimeoutError或ClosedResourceError；
• 验证时把超时调到600秒，或先在主服务里预热 embedding/Milvus，再调用 MCP。

8. 安全与部署

8.1 不要直接暴露公网

FastMCP 1.4.1 的 SSE 没有内置鉴权。推荐：

外部客户端 → Nginx / API 网关（TLS + 认证 + 限流） → 127.0.0.1:18765

认证可以做在网关层：

• Basic Auth；
• JWT；
• IP 白名单；
• 内部服务间 mTLS。

8.2 资源隔离

MCP Server 是独立进程，会加载：

• bge embedding 模型；
• Milvus 客户端；
• DeepSeek HTTP 客户端。

内存占用比单纯转发请求高。建议：

• 单独设置容器内存上限；
• 限制并发；
• 给 RAG 调用设置超时；
• 监控 Milvus 和 DeepSeek 的错误率。

8.3 权限最小化

• stdio 模式只给本地可信客户端；
• SSE 模式不要绑定0.0.0.0直接上公网；
• 知识库内容如果有租户/部门隔离，需要在search_knowledge_base内部加过滤条件，不能只靠网络隔离。

8.4 日志

建议记录：

• 调用时间；
• 客户端标识；
• question 的脱敏摘要；
• Milvus 检索耗时；
• DeepSeek 耗时；
• 是否成功。

不要把完整知识库内容写进日志。

9. 常见问题

Q1：客户端报ModuleNotFoundError: No module named 'app'？

cwd不对。必须把工作目录设为chat_py_server项目根目录，或者显式设置PYTHONPATH。

Q2：tools/list很快，但tools/call超时？

首次调用要加载 embedding 模型，耗时较长。把客户端超时调到 600 秒，或先预热。超时后常见TimeoutError、ClosedResourceError。

Q3：SSE 端口被占用？

修改：

MCP_KNOWLEDGE_SERVER_PORT=18765

Q4：stdio 模式下客户端收不到响应？

检查是否有代码往 stdout 打印日志。MCP stdio 的 stdout 只能传协议消息；日志必须走 stderr（loguru默认是 stderr）。

Q5：为什么不用 Streamable HTTP 作为 Server 传输？

mcp 1.4.1的FastMCP只支持stdio和sse。要上 Streamable HTTP 需要升级 SDK，但会牵动pydantic版本；当前先用 SSE。

Q6：SSE 没有鉴权，安全吗？

默认只绑定127.0.0.1，本机使用是安全的。远程访问必须加网关认证、TLS 和限流，不要直接暴露公网。

Q7：多个客户端同时调用会不会互相阻塞？

工具函数用asyncio.to_thread执行同步 RAG，不会阻塞事件循环；但 embedding/Milvus/DeepSeek 本身的并发能力有限，建议在网关或进程内做并发限制。

Q8：知识库检索不到内容？

检查：

1. MILVUS_COLLECTION_DOC是否有数据；
2. EMBEDDING_MODEL_PATH是否可加载；
3. Milvus 连接是否正常；
4. 知识库文档是否已经通过上传接口写入。

10. 小结

• MCP Server 的本质是“把已有能力按 MCP 协议暴露”；
• 本文用FastMCP把search_knowledge_base注册成一个 Tool；
• 支持 stdio 和 SSE 两种传输；
• RAGTool延迟导入，保证tools/list快速返回；
• asyncio.to_thread避免同步 RAG 阻塞 SSE 事件循环；
• 首次调用会加载 embedding 模型，客户端要设置足够大的超时；
• SSE 默认只绑定本机，远程访问必须加认证、TLS 和限流；
• 后续如果需要多知识库、多租户，可以在 Tool 入参加collection/tenant_id，并在检索层做过滤。