MCP(Model Context Protocol,模型上下文协议)用于统一大模型与外部工具、数据源之间的连接方式。本文从零实现一个可以搜索技术文档的 MCP Server,并使用 Python Client 和大模型完成工具调用。
MCP解决了什么问题
在 MCP 出现之前,不同模型平台的 Function Calling 格式并不完全一致。接入文件系统、数据库或第三方 API 时,开发者通常需要为每个平台重复编写适配代码。
MCP 将模型应用与外部能力拆分为两个角色:
- MCP Server:提供 Tools、Resources 和 Prompts
- MCP Client:连接 Server,发现能力并发起调用
可以把 MCP 理解为 AI 应用领域的 USB-C。Server 只需要按照协议暴露能力,Cursor、Codex、自建 Agent 等 Client 就能以统一方式使用。
一次完整调用大致如下:
用户问题
-> MCP Client读取工具列表
-> 大模型判断是否调用工具
-> Client调用MCP Server
-> Server执行搜索或读取数据
-> Client把结果交给大模型
-> 大模型生成最终回答
初始化项目
项目使用 Python 3.11 和 uv 管理依赖。
git clone https://github.com/gobinfan/python-mcp-server-client.git
cd python-mcp-server-client
uv sync
cp .env.example .env
主要依赖如下:
dependencies = [
"bs4>=0.0.2",
"httpx>=0.28.1",
"mcp[cli]>=1.28.1,<2",
"openai>=1.66.3",
]
在 .env 中配置搜索服务和模型信息:
SERPER_API_KEY=your-serper-api-key
OPENAI_API_KEY=your-openai-api-key
OPENAI_BASE_URL=https://api.openai.com/v1
OPENAI_MODEL=your-model-name
密钥不要写入代码或提交到 Git 仓库。
使用FastMCP构建Server
FastMCP 是 Python MCP SDK 提供的高层 API。创建 Server 时开启无状态 HTTP 和 JSON Response,更方便后续通过容器或多个实例部署。
from mcp.server.fastmcp import FastMCP
mcp = FastMCP(
"Agentdocs",
host="0.0.0.0",
port=8020,
stateless_http=True,
json_response=True,
)
项目实现了一个 get_docs 工具。它先把技术框架映射到官方文档域名,再通过 Serper 搜索相关页面,最后使用 BeautifulSoup 提取网页文本。
docs_urls = {
"langchain": "python.langchain.com/docs",
"llama-index": "docs.llamaindex.ai/en/stable",
"openai-agents-sdk": "openai.github.io/openai-agents-python",
"mcp-doc": "modelcontextprotocol.io",
"crew-ai": "docs.crewai.com",
}
@mcp.tool()
async def get_docs(query: str, library: str):
"""搜索指定框架的最新官方文档。"""
if library not in docs_urls:
raise ValueError(f"Library {library} not supported by this tool")
search_query = f"site:{docs_urls[library]} {query}"
results = await search_web(search_query)
if not results["organic"]:
return "No results found"
text = ""
for result in results["organic"]:
text += await fetch_url(result["link"])
return text
@mcp.tool() 会根据函数名、类型注解和 Docstring 自动生成工具名称、说明和输入 Schema。因此,清晰的参数类型和文档说明非常重要,它们会直接影响大模型选择工具的准确性。
选择传输协议
项目支持三种启动方式:
# 本地进程通信,适合桌面客户端
uv run main.py --transport stdio
# 兼容已有SSE客户端
uv run main.py --transport sse --host 0.0.0.0 --port 8020
# 推荐的HTTP方式
uv run main.py --transport streamable-http --host 0.0.0.0 --port 8020
Stdio 通过标准输入输出通信,Client 负责启动 Server 进程;SSE 和 Streamable HTTP 则允许 Client 通过网络连接。当前项目中 SSE 地址为 http://127.0.0.1:8020/sse,Streamable HTTP 地址为 http://127.0.0.1:8020/mcp。
对于新项目优先使用 Streamable HTTP;只有本地工具或需要兼容旧 Client 时,再选择 Stdio 或 SSE。
实现Python MCP Client
Client 首先建立传输层连接,然后创建 ClientSession 并初始化协议会话。
from mcp import ClientSession
from mcp.client.sse import sse_client
from mcp.client.streamable_http import streamable_http_client
async def connect_to_server(self, server_url: str):
if server_url.rstrip("/").endswith("/sse"):
context = sse_client(url=server_url)
streams = await context.__aenter__()
else:
context = streamable_http_client(server_url)
streams = (await context.__aenter__())[:2]
self.session = await ClientSession(*streams).__aenter__()
await self.session.initialize()
response = await self.session.list_tools()
print([tool.name for tool in response.tools])
连接成功后,Client 通过 list_tools() 获取 Server 能力,并转换成大模型可以识别的工具格式:
available_tools = [
{
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.inputSchema,
},
}
for tool in response.tools
]
如果模型返回 tool_calls,Client 解析工具名称和参数,通过 MCP Session 发起调用:
tool_name = tool_call.function.name
tool_args = json.loads(tool_call.function.arguments)
result = await self.session.call_tool(tool_name, tool_args)
工具结果需要连同原始 tool_call_id 追加到消息列表,再请求一次模型,才能得到面向用户的最终回答。
启动自建 Client:
uv run client.py http://127.0.0.1:8020/mcp
进入交互模式后可以提问:
查询MCP Python SDK中streamable-http的使用方法
在Codex中配置MCP Server
如果 Server 已通过 Streamable HTTP 启动,可以在 .codex/config.toml 中配置:
[mcp_servers.agentdocs_http]
url = "http://127.0.0.1:8020/mcp"
也可以由 Codex 直接以 Stdio 方式启动 Server:
[mcp_servers.agentdocs_stdio]
command = "/path/to/python-mcp-server-client/.venv/bin/python"
args = ["/path/to/python-mcp-server-client/main.py", "--transport", "stdio"]
配置完成并重启 Client 后,就可以查看和调用 get_docs 工具。Stdio 模式下不要向标准输出随意写日志,否则可能污染 MCP 的 JSON-RPC 消息;日志应写入标准错误或文件。
Tools、Resources和Prompts
MCP 不只是工具调用。项目 example/01_full_feature_server.py 演示了三种核心能力:
@mcp.tool()
def add(a: int, b: int) -> int:
return a + b
@mcp.resource("greeting://{name}")
def greeting(name: str) -> str:
return f"Hello, {name}!"
@mcp.prompt()
def summarize_city(city: str) -> str:
return f"Please summarize the current status of {city}."
- Tool 表示可以执行的动作,例如搜索、计算和写入数据库
- Resource 表示可以读取的上下文,例如文件、文档和配置
- Prompt 表示可复用的提示词模板
Tool 还可以直接返回 Pydantic Model。Client 能从 structuredContent 获取结构化结果,比解析普通文本更加稳定。
使用lifespan管理共享资源
数据库连接、HTTP Client 和知识库对象不应该在每次工具调用时重复创建。FastMCP 的 lifespan 可以在 Server 启动时初始化共享资源,并在关闭时统一释放。
@asynccontextmanager
async def lifespan(_: FastMCP):
kb = await FakeKnowledgeBase.create()
try:
yield AppContext(kb=kb)
finally:
await kb.close()
mcp = FastMCP("LifespanServer", lifespan=lifespan)
工具函数可以通过 Context 获取共享状态。这种方式适合连接池、缓存和需要复用的第三方 SDK Client。
常见问题
Client连接成功但没有工具
确认函数使用了 @mcp.tool(),并在 Session 初始化后调用 list_tools()。同时检查连接地址:SSE 通常以 /sse 结尾,Streamable HTTP 使用 /mcp。
工具经常被错误调用
完善函数类型注解和 Docstring,明确参数含义、可选值和返回内容。对 library 这类枚举参数,应在 Server 端再次校验,不能只依赖模型生成正确参数。
请求超时或返回内容太长
为外部 HTTP 请求设置超时和异常处理,并限制搜索结果数量。生产环境还应清洗 HTML、限制返回字符数,避免无关网页内容消耗模型上下文。
如何保证安全
MCP Tool 本质上可以执行代码和访问数据。部署时应增加身份认证、参数校验、访问控制和审计日志;文件、Shell、数据库写入等高风险工具还应限制作用域,并在执行前要求用户确认。
总结
MCP 的价值不在于替代大模型,而在于为模型访问外部世界提供统一协议。使用 FastMCP,只需要普通 Python 函数、类型注解和装饰器,就能快速构建 Server;Client 则负责发现工具、让模型选择工具、执行调用并回传结果。
实际项目建议从一个边界清晰的只读 Tool 开始,优先使用 Streamable HTTP,补齐超时、鉴权和日志后,再逐步接入数据库、内部 API 或自动化任务。
参考
- https://modelcontextprotocol.io
- https://github.com/modelcontextprotocol/python-sdk
- https://github.com/gobinfan/python-mcp-server-client
文档信息
- 本文作者:onefeng
- 本文链接:https://me.onefeng.xyz/2026/07/21/%E6%9E%84%E5%BB%BAmcp-server-client/
- 版权声明:自由转载-非商用-非衍生-保持署名(创意共享3.0许可证)