基于MCP协议构建AI Agent工具调用服务器：从原理到Python实战
在AI应用开发中，Function Calling（函数调用）是连接大语言模型与外部工具、数据源的核心机制。其原理在于将自然语言指令转化为结构化请求，使模型能够执行搜索、计算或数据查询等操作。这项技术的价值在于突破了模型自身知识库的局限，极大地扩展了AI的实际应用能力，广泛应用于智能客服、代码助手、数据分析等场景。然而，不同模型厂商（如OpenAI、Anthropic）的Function Call
1. 项目概述
最近在折腾AI Agent开发，发现一个挺头疼的问题：不同的大模型厂商，比如OpenAI、Anthropic、DeepSeek，它们的Function Calling格式都不太一样。更麻烦的是，各种外部工具和API的封装方式也是五花八门，每次想给Agent加个新能力，都得重新写一遍适配代码，效率实在太低。
直到我接触到了MCP（Model Context Protocol），感觉像是找到了救星。简单来说，MCP就像是为AI世界定义了一个“USB-C”接口。它把大模型调用工具的流程标准化了，无论是哪种模型，哪种工具，只要遵循MCP协议，就能即插即用。这大大简化了Agent的扩展和集成工作。
这个项目，就是我基于Python，从零开始构建一个MCP Server和Client的实战记录。我实现了一个具体的功能：让AI能够查询市面上主流Agent框架（比如LangChain、LlamaIndex、AutoGen等）的技术文档。Server端支持Stdio（本地）和SSE（远程）两种传输协议，Client端则演示了如何通过OpenAI API来调用这些工具。如果你也在做AI应用开发，或者对如何让大模型更“懂”你的业务数据感兴趣，这篇内容应该能给你一些直接的参考。
2. MCP协议核心价值与设计思路
2.1 为什么我们需要MCP？
在没有MCP之前，AI调用外部功能主要依靠各厂商自家的Function Calling机制。这带来了两个核心痛点：
首先是格式不统一。 OpenAI有OpenAI的function calling格式，Anthropic的Claude有它自己的工具调用格式，国内的一些大模型可能又是另一套。这意味着，如果你开发了一个工具，想让它同时被GPT-4和Claude使用，你就得为每个模型单独写一套适配层。这不仅仅是多写几行代码的问题，更是维护的噩梦。
其次是工具封装混乱。 一个工具，比如“查询数据库”，它的输入参数、输出格式、错误处理，都没有一个社区公认的标准。A开发者可能返回JSON，B开发者可能返回纯文本。当你的Agent需要集成几十个这样的工具时，光是把这些各异的接口“翻译”成模型能理解的样子，就是一项巨大的工程。
MCP的出现，就是为了解决这两个问题。它定义了一套标准的、与模型无关的协议，用于在AI模型（Client）和工具、数据源（Server）之间进行通信。你可以把MCP Server想象成一个标准的“插座”，而MCP Client就是“插头”。只要插头和插座都符合MCP规范，不管背后的电器（模型）和电源（工具）是什么品牌，都能正常工作。
2.2 项目目标与架构选型
我这个项目的目标很明确：构建一个能查询特定技术文档的MCP Server，并配套一个能调用它的Client。我选择了Python生态，主要是因为 mcp 这个官方SDK对Python的支持非常友好，能极大降低开发门槛。
在架构上，我决定同时实现两种传输协议，以适应不同的部署场景：
- Stdio（标准输入输出）协议 ：这是为本地开发或单机部署设计的。Server作为一个独立的进程启动，通过标准输入（stdin）接收请求，通过标准输出（stdout）返回响应。它的优点是简单、零网络依赖，非常适合集成到Cursor、Cline这类本地IDE插件中。
- SSE（Server-Sent Events）协议 ：这是为远程服务设计的。基于HTTP长连接，Server可以部署在云服务器上，多个Client可以通过网络连接上来。这为构建集中式的工具服务、微服务架构提供了可能。
工具本身的功能设计为：接收用户关于某个Agent框架的技术问题（query）和指定的框架名称（library），然后自动去该框架的官方文档站点进行站内搜索，抓取相关页面的文本内容并返回。这样，AI在回答技术问题时，就能直接引用最新、最准确的官方文档，而不是依赖可能过时或不全的预训练知识。
3. 环境准备与项目初始化
3.1 包管理工具UV的安装
项目的第一步是搭建开发环境。我强烈推荐使用 uv 作为Python的包管理器和项目工具链。它由Astral团队（Ruff的创建者）开发，速度极快，并且一体化地解决了虚拟环境、依赖安装和脚本运行的问题，体验比传统的 pip + venv 组合要好得多。
安装过程非常简单：
对于MacOS或Linux系统，打开终端，运行以下命令：
curl -LsSf https://astral.sh/uv/install.sh | sh
这条命令会从Astral的官方服务器下载安装脚本并执行。安装完成后，通常需要重启终端或者执行 source ~/.bashrc （或 source ~/.zshrc ）来让 uv 命令生效。
对于Windows系统，建议在PowerShell（管理员模式）中运行：
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
命令中的 -ExecutionPolicy ByPass 是为了允许执行远程脚本， irm 是 Invoke-RestMethod 的别名，用于下载安装脚本， iex 是 Invoke-Expression 的别名，用于执行下载的脚本。
注意 ：安装完成后，可以通过运行
uv --version来验证是否安装成功。如果系统提示找不到命令，请检查你的终端配置文件（如.bashrc,.zshrc），确保UV的安装路径（通常是$HOME/.cargo/bin或$HOME/.local/bin）已经添加到PATH环境变量中。
3.2 创建项目与核心依赖安装
环境准备好后，就可以创建我们的MCP项目了。
首先，创建一个项目目录并初始化：
# 使用uv初始化一个名为mcp-docs-server的新项目
uv init mcp-docs-server
cd mcp-docs-server
uv init 命令会创建一个包含基本项目结构（如 pyproject.toml 文件）的目录。
接着，创建并激活Python虚拟环境。UV使得这一步非常优雅：
# 创建虚拟环境，UV会默认使用.venv作为环境目录名
uv venv
# 激活虚拟环境
# 在MacOS/Linux上：
source .venv/bin/activate
# 在Windows上：
.venv\Scripts\activate
激活后，你的命令行提示符前通常会显示 (.venv) ，表示你已经在这个独立的Python环境中了。
现在安装项目所需的核心依赖：
uv add "mcp[cli]" httpx beautifulsoup4 python-dotenv uvicorn
让我解释一下每个包的作用：
mcp[cli]：这是MCP协议的官方Python SDK。[cli]是一个“额外”依赖，它包含了运行MCP命令行工具所需的包，对于开发和调试非常有用。httpx：一个现代、异步的HTTP客户端库。我们将用它来发起网络请求，例如调用搜索API和抓取网页。它比传统的requests库对异步编程的支持更好。beautifulsoup4：强大的HTML解析库。当抓取到文档网页后，我们需要用它来提取纯净的文本内容，剔除HTML标签、脚本和样式。python-dotenv：用于从.env文件加载环境变量。像API密钥这样的敏感信息，我们绝不会硬编码在代码里。uvicorn：一个轻量级、快速的ASGI服务器。它是运行我们基于SSE协议的MCP Server所必需的。
最后，创建我们的主服务器文件：
touch main.py
至此，一个干净的、依赖明确的项目骨架就搭建完成了。 pyproject.toml 文件记录了所有依赖，团队其他成员或者部署时，只需要 uv sync 一下就能复现完全一致的环境。
4. 核心工具函数实现与原理剖析
4.1 定义文档源与搜索策略
工具的核心能力是“精准搜索特定框架的文档”。我们不能让AI去全网搜索，那样噪音太多。正确的做法是进行“站内搜索”（site search）。这就需要我们预先知道每个框架官方文档的域名。
我在代码中定义了一个字典 docs_urls 来维护这个映射关系：
docs_urls = {
"langchain": "python.langchain.com/docs",
"llama-index": "docs.llamaindex.ai/en/stable",
"autogen": "microsoft.github.io/autogen/stable",
"agno": "docs.agno.com",
"openai-agents-sdk": "openai.github.io/openai-agents-python",
"mcp-doc": "modelcontextprotocol.io",
"camel-ai": "docs.camel-ai.org",
"crew-ai": "docs.crewai.com"
}
为什么这么设计？
- 键（Key）是框架的标识符 ：如
langchain、llama-index。这个标识符应该简单、无歧义，并且最好和框架的通用称呼一致，方便用户记忆和输入。 - 值（Value）是文档的站点限定符 ：格式通常是
{域名}{可能的具体路径}。例如python.langchain.com/docs，这将在后续构建搜索查询时，作为site:参数的值，确保搜索引擎只在该域名及其子路径下查找。
实操心得 ：维护这个映射表是个持续的过程。框架的文档地址可能会变（比如从
docs.llamaindex.ai迁移到docs.llamaindex.com），或者有新的框架需要支持。一个好的实践是将这个字典放在一个单独的配置文件（如config.py）或甚至一个小型数据库中，方便动态更新，而无需修改核心代码。
4.2 集成搜索引擎API
为了执行搜索，我们需要一个搜索引擎。自己从零搭建一个爬虫和索引系统不现实，所以我选择了 Serper API 。它是一个专门为AI应用设计的谷歌搜索API，价格低廉，响应速度快，并且返回结构化的JSON数据，非常适合程序处理。
使用前，你需要去 Serper官网 注册一个账号，获取免费的API额度（每天足够用于开发和测试）。拿到API Key后，在项目根目录创建一个 .env 文件来保存它：
# .env 文件
SERPER_API_KEY=你的_serper_api_key_在这里
OPENAI_API_KEY=你的_openai_api_key_在这里 # Client端会用到
OPENAI_BASE_URL=https://api.openai.com/v1 # 如果你用的是官方OpenAI
重要安全提示 ：务必把 .env 文件添加到 .gitignore 中，避免将API密钥意外提交到公开的代码仓库。
搜索函数 search_web 的实现如下：
import json
import os
import httpx
SERPER_URL = "https://google.serper.dev/search"
async def search_web(query: str) -> dict | None:
# 构造请求体，这里我限制返回3条结果，平衡信息量和响应速度
payload = json.dumps({"q": query, "num": 3})
headers = {
"X-API-KEY": os.getenv("SERPER_API_KEY"), # 从环境变量读取密钥
"Content-Type": "application/json",
}
async with httpx.AsyncClient() as client:
try:
# 发起异步POST请求，设置30秒超时
response = await client.post(
SERPER_URL, headers=headers, data=payload, timeout=30.0
)
response.raise_for_status() # 如果HTTP状态码不是2xx，抛出异常
return response.json() # 返回解析后的JSON数据
except httpx.TimeoutException:
# 处理超时，返回一个空结果结构，避免上层函数崩溃
return {"organic": []}
except Exception as e:
# 可以在这里添加更细致的错误日志
print(f"Search error: {e}")
return {"organic": []}
关键点解析 ：
- 异步(async/await) ：使用
async def和await是现代Python网络编程的推荐做法。它允许服务器在等待网络I/O（如API响应）时去处理其他请求，极大提高了并发性能。我们的MCP Server底层也是异步的，所以工具函数也必须是异步的。 - 超时处理 ：网络请求永远不可靠。设置
timeout=30.0是必须的，防止一个慢速请求阻塞整个服务器。超时后，我们返回一个空列表，让工具优雅降级。 - 错误处理 ：
response.raise_for_status()会在API返回错误（如401密钥无效、429频率限制）时抛出HTTPStatusError。在更完善的生产代码中，应该捕获这个异常并返回更友好的错误信息给调用方。
4.3 网页内容抓取与清洗
搜索API返回的是链接和摘要，我们需要进一步抓取目标网页的完整内容。这里用到了 httpx 和 BeautifulSoup 。
from bs4 import BeautifulSoup
async def fetch_url(url: str) -> str:
async with httpx.AsyncClient() as client:
try:
response = await client.get(url, timeout=30.0)
# 使用html.parser解析器，它是Python标准库的一部分，无需额外依赖
soup = BeautifulSoup(response.text, "html.parser")
# 获取所有文本，并自动合并连续的空白字符
text = soup.get_text(separator=' ', strip=True)
return text
except httpx.TimeoutException:
return "[Error: Timeout while fetching the page]"
except Exception as e:
return f"[Error: Failed to fetch page - {str(e)}]"
内容清洗的注意事项 ：
soup.get_text(separator=' ', strip=True)：这个调用会移除所有HTML标签，只保留文本。separator=' '确保原本被标签分隔的文本用一个空格连接，而不是挤在一起。strip=True会去掉文本开头和结尾的空白。- 局限性 ：这种方法会丢失所有的格式（如加粗、标题层级）和结构（如代码块、表格）。对于技术文档，这可能会影响可读性。更高级的方案可以尝试只提取
<article>、<main>标签内的内容，或者使用专门的文档解析库（如readability、trafilatura），但复杂度会大大增加。当前方案在简单性和效果上取得了不错的平衡。 - 礼貌抓取 ：在循环中抓取多个URL时，最好在请求间添加短暂的延迟（如
await asyncio.sleep(1)），并设置合理的User-Agent头，以示对目标网站的尊重，避免被反爬机制屏蔽。
4.4 组装MCP工具
最后，我们将上述组件组装成一个MCP工具。 mcp 库的 @tool() 装饰器是这个过程的核心。
from mcp import tool
@tool()
async def get_docs(query: str, library: str) -> str:
"""
搜索给定查询和库的最新文档。
支持 langchain、llama-index、autogen、agno、openai-agents-sdk、mcp-doc、camel-ai 和 crew-ai。
参数:
query: 要搜索的查询 (例如 "如何创建Agent")
library: 要搜索的库 (例如 "langchain")
返回:
文档中的文本
"""
# 1. 校验库是否支持
if library not in docs_urls:
raise ValueError(f"Library '{library}' is not supported. Supported libraries are: {list(docs_urls.keys())}")
# 2. 构建站内搜索查询
site_domain = docs_urls[library]
search_query = f"site:{site_domain} {query}"
# 3. 执行搜索
results = await search_web(search_query)
if not results.get("organic"):
return f"No documentation found for '{query}' in {library}."
# 4. 并发抓取结果页面内容
fetched_texts = []
for result in results["organic"]:
link = result["link"]
text = await fetch_url(link)
# 简单过滤，避免返回完全空或错误的内容
if text and len(text) > 50 and not text.startswith("[Error"):
fetched_texts.append(f"--- From: {link} ---\n{text}\n")
# 5. 合并并返回结果
if not fetched_texts:
return f"Found links but failed to retrieve meaningful content for '{query}' in {library}."
# 限制总返回长度，避免超出模型上下文限制
combined_text = "\n".join(fetched_texts)
# 简单截断，生产环境可以考虑更智能的摘要
max_length = 8000
if len(combined_text) > max_length:
combined_text = combined_text[:max_length] + "\n\n[Content truncated due to length limit]"
return combined_text
工具设计要点 ：
- 文档字符串（Docstring）至关重要 ：大模型（Client）正是通过这个文档字符串来理解这个工具是干什么的、需要什么参数。描述必须清晰、准确。参数名和描述要对应。
- 错误处理与用户体验 ：工具不能因为一个链接失效或搜索无结果就崩溃。我们进行了多层校验和友好的错误信息返回。例如，当库不支持时，抛出一个清晰的
ValueError，并列出所有支持的库，这对调试非常有帮助。 - 输出长度管理 ：大模型有上下文窗口限制。无节制地返回抓取到的所有文本，很快就会撑爆上下文。这里我设置了一个简单的截断机制（8000字符）。更优的方案是引入文本摘要（Summarization）步骤，或者使用嵌入模型（Embedding）进行语义检索，只返回最相关的片段。
5. 构建Stdio协议的MCP Server
5.1 Server核心代码实现
有了工具函数，构建Server就水到渠成了。Stdio协议的Server是最简单的一种，它直接运行一个进程，通过标准输入输出流与Client通信。
# main_stdio.py
from mcp.server.fastmcp import FastMCP
from dotenv import load_dotenv
import httpx
import json
import os
from bs4 import BeautifulSoup
# 加载环境变量
load_dotenv()
# 创建FastMCP实例，给它起个名字
mcp = FastMCP("AgentDocsServer")
# 这里省略之前定义的 docs_urls, SERPER_URL, search_web, fetch_url 等变量和函数...
# 使用 mcp 实例的 tool 装饰器注册工具
@mcp.tool()
async def get_docs(query: str, library: str):
"""
搜索给定查询和库的最新文档。
支持 langchain、llama-index、autogen、agno、openai-agents-sdk、mcp-doc、camel-ai 和 crew-ai。
参数:
query: 要搜索的查询 (例如 "React Agent")
library: 要搜索的库 (例如 "agno")
返回:
文档中的文本
"""
# 工具实现逻辑与上一节完全相同
if library not in docs_urls:
raise ValueError(f"Library {library} not supported by this tool")
query = f"site:{docs_urls[library]} {query}"
results = await search_web(query)
if len(results["organic"]) == 0:
return "No results found"
text = ""
for result in results["organic"]:
text += await fetch_url(result["link"])
return text
# 程序入口
if __name__ == "__main__":
# 以 stdio 传输模式运行服务器
mcp.run(transport="stdio")
代码非常简洁。核心是 FastMCP 类，它是对底层MCP Server的高级封装。我们用 @mcp.tool() 装饰器注册了 get_docs 函数。最后调用 mcp.run(transport="stdio") ，服务器就会启动并监听标准输入。
启动服务器：
# 确保在项目目录下，且虚拟环境已激活
uv run main_stdio.py
启动后，你会发现程序似乎“挂起”了，没有输出。这是正常的，因为它正在等待从 stdin 接收符合MCP协议的消息。
5.2 在IDE中配置与使用（Cursor/Cline）
Stdio Server的主要应用场景是与支持MCP的IDE（如Cursor、Cline）集成，让AI助手能直接调用我们自定义的工具。
在Cursor中配置：
- 在你的项目根目录下，创建一个名为
.cursor的文件夹。 - 在
.cursor文件夹内，创建一个mcp.json文件。 - 编辑
mcp.json，内容如下：
{
"mcpServers": {
"agent-docs-server": {
"command": "uv",
"args": [
"--directory",
"/你的/绝对/项目/路径/mcp-docs-server",
"run",
"main_stdio.py"
],
"env": {
"SERPER_API_KEY": "你的_serper_api_key"
}
}
}
}
配置详解 ：
command: 指定启动Server的命令。我们使用uv。args: 传递给命令的参数。--directory: 告诉uv在哪个目录下执行。 这里必须使用绝对路径 。run main_stdio.py: 让uv运行我们的主脚本。
env: （可选）可以在这里直接设置环境变量，避免依赖外部的.env文件。对于团队共享配置或不想创建.env文件的情况很有用。
- 保存文件，重启Cursor（或重新加载窗口）。如果配置正确，你会在Cursor的Activity Bar（活动栏）看到一个“MCP”图标，点击可以看到
agent-docs-server显示为已连接（绿色）。
在Cline中配置： Cline是VSCode的插件，配置方式类似，但配置文件的位置不同。
- 在VSCode中，打开设置（
Ctrl+,或Cmd+,）。 - 搜索
mcp，找到MCP Servers配置项。 - 点击“在settings.json中编辑”，添加如下配置：
{
"mcpServers": {
"agent-docs-server": {
"command": "uv",
"args": [
"--directory",
"/你的/绝对/项目/路径/mcp-docs-server",
"run",
"main_stdio.py"
],
"env": {
"SERPER_API_KEY": "你的_serper_api_key"
}
}
}
}
配置成功后的使用体验 ： 配置成功后，你就可以在Cursor或Cline的聊天框中直接使用了。例如，你可以问：“用LangChain怎么创建一个能使用工具的Agent？” AI助手在理解你的问题后，会自动识别出需要调用 get_docs 工具，并传入 query="create tool using agent" 和 library="langchain" 参数。随后，它会在后台启动（或连接）我们的Server，执行搜索和抓取，并将获取到的文档片段作为上下文来生成更精准的回答。
避坑指南 ：最常见的配置失败原因是路径错误或环境问题。务必使用绝对路径。如果Server启动失败，可以尝试在终端手动运行
uv run main_stdio.py命令，看是否有Python包导入错误或API密钥缺失的报错。此外，确保你的uv和python都在系统PATH中，或者使用完整的可执行文件路径。
6. 构建SSE协议的MCP Server
6.1 SSE Server架构解析
Stdio协议适合本地集成，但如果我想把这个文档查询服务部署到云端，供多个客户端或者一个团队使用，就需要SSE协议了。SSE（Server-Sent Events）是一种基于HTTP的服务器向客户端推送数据的技术。在MCP中，它被用来建立Client和Server之间的双向通信通道。
与简单的Stdio Server相比，SSE Server需要处理HTTP请求、管理连接，因此代码结构稍复杂一些。我们需要一个Web服务器（这里用 uvicorn + starlette ）来承载SSE端点。
# main_sse.py
from mcp.server.fastmcp import FastMCP
from mcp.server.sse import SseServerTransport
from mcp.server import Server
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.routing import Route, Mount
import uvicorn
import argparse
# 省略之前重复的导入和工具函数定义...
# 创建FastMCP实例和工具（与Stdio版本完全相同）
mcp = FastMCP("AgentDocsServer")
# ... 注册 get_docs 工具 ...
# 关键函数：创建Starlette应用，并将MCP Server绑定到SSE路由
def create_starlette_app(mcp_server: Server, *, debug: bool = False) -> Starlette:
"""创建一个Starlette应用，用于通过SSE服务MCP Server。"""
# 初始化SSE传输层
sse_transport = SseServerTransport("/messages/")
async def handle_sse(request: Request) -> None:
"""
处理SSE连接请求。
这个函数是SSE连接的生命周期管理器。
"""
# 1. 建立SSE连接，获取读写流
async with sse_transport.connect_sse(
request.scope, # 请求的ASGI scope
request.receive, # 用于接收客户端消息的通道
request._send, # 用于向客户端发送消息的通道
) as (read_stream, write_stream):
# 2. 在这个连接上下文中，运行MCP Server的主循环
await mcp_server.run(
read_stream, # Server从Client读取消息
write_stream, # Server向Client写入消息
mcp_server.create_initialization_options(), # 初始化选项
)
# 3. 当连接断开（跳出with块），异步上下文管理器会自动清理资源
# 定义路由：
# - /sse: 客户端通过这个端点建立SSE长连接。
# - /messages/: 客户端通过这个端点向服务器发送具体的请求消息。
return Starlette(
debug=debug,
routes=[
Route("/sse", endpoint=handle_sse),
Mount("/messages/", app=sse_transport.handle_post_message),
],
)
if __name__ == "__main__":
# 获取底层的标准MCP Server对象
mcp_server = mcp._mcp_server
# 解析命令行参数，方便指定主机和端口
parser = argparse.ArgumentParser(description='Run MCP SSE-based server')
parser.add_argument('--host', default='0.0.0.0', help='Host to bind to')
parser.add_argument('--port', type=int, default=8020, help='Port to listen on')
args = parser.parse_args()
# 创建Web应用
starlette_app = create_starlette_app(mcp_server, debug=True)
# 使用uvicorn启动ASGI服务器
print(f"Starting MCP SSE server on http://{args.host}:{args.port}")
uvicorn.run(starlette_app, host=args.host, port=args.port)
代码流程剖析 ：
- 创建MCP Server ：
FastMCP创建了一个高级封装，我们通过._mcp_server属性获取到底层的标准Server对象，供SSE传输层使用。 - 创建SSE传输层 ：
SseServerTransport是MCP SDK提供的，它封装了SSE协议与MCP消息格式之间的转换逻辑。它需要知道客户端发送消息的端点路径（这里是/messages/）。 - 定义SSE端点处理器 ：
handle_sse函数是核心。当一个客户端（如我们后面要写的Python Client）连接到/sse时，这个函数被调用。sse_transport.connect_sse(...)：建立SSE连接，并返回一对异步流（read_stream,write_stream）。这相当于在HTTP长连接上建立了一个双向通信的“管道”。await mcp_server.run(...)：MCP Server的主循环开始运行。它会从read_stream读取客户端发来的请求（如“调用get_docs工具”），处理请求（执行我们的工具函数），然后将结果通过write_stream写回给客户端。
- 创建Web应用并路由 ：我们将
/sse路径映射到handle_sse函数，将/messages/路径挂载到sse_transport.handle_post_message。后者负责接收客户端通过HTTP POST发送的具体消息。 - 启动服务器 ：使用
uvicorn这个ASGI服务器来运行我们创建的Starlette应用。
启动SSE Server：
uv run main_sse.py --host 0.0.0.0 --port 8020
看到 Starting MCP SSE server on http://0.0.0.0:8020 的输出，说明服务器已经在8020端口监听所有网络接口的连接了。
6.2 部署与运维考量
将SSE Server部署到生产环境（如云服务器）时，需要考虑以下几点：
-
进程管理 ：不能简单地用
uv run在终端前台运行。需要使用进程管理器，如systemd（Linux）、supervisord或pm2，来保证服务在后台稳定运行，崩溃后自动重启。# 一个简单的systemd服务文件示例 (/etc/systemd/system/mcp-docs.service) [Unit] Description=MCP Docs Search Server After=network.target [Service] Type=simple User=ubuntu WorkingDirectory=/path/to/your/mcp-docs-server Environment="PATH=/home/ubuntu/.local/bin:/usr/local/bin:/usr/bin" Environment="SERPER_API_KEY=your_key_here" ExecStart=/home/ubuntu/.cargo/bin/uv run main_sse.py --host 127.0.0.1 --port 8020 Restart=on-failure [Install] WantedBy=multi-user.target -
网络与安全 ：
--host 0.0.0.0意味着监听所有IP，这在公网服务器上可能有风险。在生产环境，更安全的做法是绑定到127.0.0.1，然后通过Nginx这样的反向代理对外暴露，Nginx可以处理SSL/TLS加密（HTTPS）、负载均衡和访问控制。- 考虑在Server端增加简单的认证，例如要求Client在请求头中携带一个API Token。
-
性能与扩展 ：单个Python进程处理能力有限。如果工具调用很耗时（比如网络请求多），或者并发请求量大，需要考虑：
- 使用
uvicorn的--workers选项启动多个工作进程。 - 将工具函数中的IO密集型操作（如网络请求）放到异步任务队列（如
celery+redis）中执行，避免阻塞主事件循环。 - 对搜索结果或抓取到的文档内容进行缓存，减少对搜索引擎和文档站点的重复请求。
- 使用
7. 构建Python MCP Client
7.1 Client端的设计与连接
Server准备好了，我们需要一个能与之对话的Client。这个Client将扮演“大模型”的角色，负责理解用户意图，决定何时调用工具，并整合工具结果生成最终回复。这里我们用OpenAI的API来模拟这个大模型，但实际上任何能处理OpenAI格式的function calling的模型或框架都可以。
# client.py
import asyncio
import json
import os
import sys
from typing import Optional
from contextlib import AsyncExitStack
from mcp import ClientSession
from mcp.client.sse import sse_client
from openai import AsyncOpenAI
from dotenv import load_dotenv
load_dotenv()
class MCPClient:
def __init__(self):
self.session: Optional[ClientSession] = None
self.exit_stack = AsyncExitStack() # 用于管理异步上下文
self.openai = AsyncOpenAI(
api_key=os.getenv("OPENAI_API_KEY"),
base_url=os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
)
self._streams_context = None
self._session_context = None
async def connect_to_sse_server(self, server_url: str):
"""连接到运行SSE传输的MCP服务器"""
print(f"Connecting to MCP server at {server_url}...")
# 1. 建立SSE连接上下文
self._streams_context = sse_client(url=server_url)
streams = await self._streams_context.__aenter__()
# 2. 创建MCP客户端会话上下文
self._session_context = ClientSession(*streams)
self.session = await self._session_context.__aenter__()
# 3. 执行MCP初始化握手
await self.session.initialize()
# 4. 列出服务器提供的工具，验证连接
response = await self.session.list_tools()
tools = response.tools
print(f"✓ Connected successfully!")
print(f"Available tools: {[tool.name for tool in tools]}")
async def cleanup(self):
"""清理资源，关闭连接"""
print("\nCleaning up resources...")
if self._session_context:
await self._session_context.__aexit__(None, None, None)
if self._streams_context:
await self._streams_context.__aexit__(None, None, None)
print("Cleanup complete.")
连接过程详解 ：
- 初始化 ：创建
AsyncOpenAI客户端，用于后续与OpenAI API通信。 sse_client(url)：这是MCP SDK提供的客户端SSE传输层工厂函数。它接收Server的SSE端点URL（例如http://localhost:8020/sse），并返回一个异步上下文管理器。进入这个上下文（__aenter__）会建立HTTP长连接。ClientSession(*streams)：用上一步建立的连接流（一个读流，一个写流）创建MCP客户端会话。会话是进行所有MCP操作（如初始化、列出工具、调用工具）的入口。session.initialize()：执行MCP协议要求的初始化握手。Server和Client会交换版本、能力等信息。session.list_tools()：向Server请求当前可用的工具列表。这是验证连接是否正常、工具是否注册成功的好方法。
7.2 工具调用与对话循环
连接建立后，Client的核心工作就是处理用户查询，协调大模型和工具。
async def process_query(self, user_query: str) -> str:
"""处理用户查询：使用OpenAI决定是否/如何调用工具，并整合结果。"""
# 初始化对话历史
messages = [{"role": "user", "content": user_query}]
# 1. 获取Server端可用的工具列表，并格式化成OpenAI API要求的格式
response = await self.session.list_tools()
available_tools = []
for tool in response.tools:
# MCP工具描述转换为OpenAI function calling格式
available_tools.append({
"type": "function",
"function": {
"name": tool.name,
"description": tool.description or f"Calls the {tool.name} tool.",
"parameters": tool.inputSchema # MCP工具本身就定义了输入参数的JSON Schema
}
})
final_response_parts = []
max_iterations = 5 # 防止无限循环
iteration = 0
while iteration < max_iterations:
iteration += 1
# 2. 调用OpenAI，传入当前对话历史和可用工具
completion = await self.openai.chat.completions.create(
model=os.getenv("OPENAI_MODEL", "gpt-3.5-turbo"),
messages=messages,
tools=available_tools if available_tools else None, # 只在第一轮或需要时传递？
tool_choice="auto", # 让模型自己决定是否调用工具
)
assistant_message = completion.choices[0].message
# 3. 判断模型是否决定调用工具
if assistant_message.tool_calls:
print(f"\n[Iteration {iteration}] Model decided to use tools.")
for tool_call in assistant_message.tool_calls:
tool_name = tool_call.function.name
try:
tool_args = json.loads(tool_call.function.arguments)
except json.JSONDecodeError:
tool_args = {}
print(f"Warning: Failed to parse arguments for {tool_name}")
# 记录工具调用
final_response_parts.append(f"[Calling tool: {tool_name} with args: {tool_args}]")
# 4. 执行实际的MCP工具调用
try:
result = await self.session.call_tool(tool_name, tool_args)
# result.content 是一个列表，通常第一个元素是文本
tool_result_text = result.content[0].text if result.content else "Tool executed with no text output."
print(f"Tool '{tool_name}' returned: {tool_result_text[:200]}...") # 打印前200字符
except Exception as e:
tool_result_text = f"Error calling tool {tool_name}: {str(e)}"
print(f"Tool call error: {tool_result_text}")
# 5. 将工具调用和结果追加到对话历史，让模型进行下一轮思考
messages.append({
"role": "assistant",
"content": None,
"tool_calls": [tool_call]
})
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": tool_result_text
})
final_response_parts.append(f"[Tool result]: {tool_result_text}")
# 本轮有工具调用，继续循环，让模型基于工具结果生成最终回答
continue
else:
# 6. 模型没有调用工具，直接生成最终回答
final_answer = assistant_message.content
if final_answer:
final_response_parts.append(final_answer)
break # 退出循环
# 7. 返回整合后的最终文本
return "\n\n".join(final_response_parts)
async def chat_loop(self):
"""运行一个交互式聊天循环"""
print("\n" + "="*50)
print("MCP Client Started! Connect to your AI assistant.")
print("Type your query (e.g., 'How to create an agent in LangChain?')")
print("Type 'quit' or 'exit' to stop.")
print("="*50)
while True:
try:
user_input = input("\nYou: ").strip()
if user_input.lower() in ['quit', 'exit', 'q']:
print("Goodbye!")
break
if not user_input:
continue
print("\nThinking...")
response = await self.process_query(user_input)
print(f"\nAssistant:\n{response}")
except KeyboardInterrupt:
print("\n\nInterrupted by user.")
break
except Exception as e:
print(f"\nAn error occurred: {type(e).__name__}: {e}")
核心交互逻辑分析 ： 这是一个典型的 多轮对话（Multi-turn） 和 工具调用（Tool Calling） 的循环：
- 准备工具列表 ：将MCP Server提供的工具描述，转换成OpenAI API认识的
tools参数格式。 - 调用大模型 ：将用户问题
user_query和工具列表available_tools发给OpenAI。tool_choice="auto"让模型自己判断是否需要调用工具。 - 解析模型决策 ：检查
assistant_message.tool_calls。如果非空，说明模型决定调用工具。 - 执行工具调用 ：遍历每个工具调用指令，提取工具名和参数，然后通过
await self.session.call_tool(tool_name, tool_args)真正地调用远端的MCP Server。这是Client和Server通信的关键一步。 - 更新对话历史 ：这是实现多轮交互的关键。我们必须严格按照OpenAI的格式，将
assistant的tool_calls消息和tool角色（携带结果）的消息追加到messages列表末尾。这样，在下一次循环中，模型就能看到它之前决定调用工具，以及工具返回的结果，从而基于这些新信息生成最终回答或决定调用下一个工具。 - 生成最终回答 ：当模型不再调用工具时（
assistant_message.tool_calls为空），其content就是最终答案。 - 循环控制 ：添加了
max_iterations限制，防止模型陷入无限调用工具的循环。
启动Client并连接我们的SSE Server：
# 假设SSE Server运行在本地8020端口
uv run client.py http://localhost:8020/sse
如果一切正常，你会看到连接成功和可用工具的提示，然后进入交互式聊天界面。你可以尝试输入：“用LangChain怎么创建一个能使用工具的Agent？” Client会协调OpenAI和我们的MCP Server，最终给出一个结合了实时文档搜索结果的回答。
7.3 错误处理与健壮性增强
上面的基础版本忽略了大量错误处理。一个健壮的Client应该考虑：
- 网络连接与重试 ：SSE连接可能不稳定。需要添加重试逻辑，并在连接断开时尝试重新连接。
- 工具调用超时 ：
session.call_tool应该设置超时，防止某个工具执行时间过长卡死整个对话。 - 模型输出解析 ：对模型返回的
tool_calls参数进行更严格的校验，确保name在可用工具列表中，arguments是合法的JSON。 - 上下文长度管理 ：随着多轮对话进行，
messages历史会越来越长。需要设计策略来修剪或总结历史，避免超出模型的上下文窗口。 - 流式输出 ：当前是等待所有工具调用和最终回答完成后再一次性输出。更好的体验是使用OpenAI的流式响应，让用户看到模型“思考”和工具调用的过程。
8. 常见问题、调试技巧与扩展方向
8.1 开发与调试中的常见问题
1. 连接失败： ConnectionRefusedError 或 Cannot connect to host
- 原因 ：SSE Server没有运行，或者主机/端口不对。
- 排查 ：
- 在服务器端，确认
uvicorn已成功启动，并监听在正确的端口（如0.0.0.0:8020）。 - 在客户端，检查URL是否正确（如
http://服务器IP:8020/sse）。 - 检查防火墙设置，确保端口（如8020）对客户端开放。
- 在服务器端，确认
2. MCP握手失败：初始化后没有列出工具
- 原因 ：Server端的工具没有正确注册，或者传输层协议有误。
- 排查 ：
- 确保Server代码中使用了
@mcp.tool()装饰器，并且工具函数是async的。 - 检查Server启动日志，看是否有导入错误或运行时异常。
- 对于Stdio Server，可以尝试用一个简单的测试Client（或使用
mcpCLI工具）来验证。例如，安装mcp[cli]后，可以运行：uvx mcp dev main_stdio.py，这会启动一个开发服务器并提供一个简单的交互界面来测试工具。
- 确保Server代码中使用了
3. 工具调用返回错误或超时
- 原因 ：工具函数内部出错（如API密钥无效、网络超时、网页结构变化导致解析失败）。
- 排查 ：
- 在Server工具函数内部添加更详细的日志，打印关键步骤和错误信息。
- 检查环境变量（
SERPER_API_KEY）是否已正确设置并被Server进程读取。 - 模拟工具调用：直接在Python REPL中导入并调用你的
search_web和fetch_url函数，看是否能正常工作。
4. Cursor/Cline中MCP Server显示未连接
- 原因 ：配置文件路径错误、命令执行失败、或环境变量缺失。
- 排查 ：
- 绝对路径 ：确认
mcp.json中的args里的--directory参数是 绝对路径 。 - 手动执行 ：打开终端，切换到项目目录，手动执行配置文件中定义的命令（如
uv --directory /path run main.py），看是否能成功启动Server并观察有无报错。 - 查看日志 ：Cursor和Cline通常有输出面板或日志文件，可以查看MCP连接的具体错误信息。
- 绝对路径 ：确认
8.2 性能优化与扩展思路
当前实现是一个功能完整但基础的版本。在实际生产应用中，可以从以下几个方向进行优化和扩展：
1. 增加结果缓存 频繁搜索相同的查询是一种浪费。可以引入一个缓存层（如 redis 或 diskcache ），将 (query, library) 作为键，将抓取到的文本作为值缓存起来，并设置一个合理的过期时间（例如1小时）。这能显著降低对Serper API的调用次数，提升响应速度。
2. 实现更智能的内容提取与摘要 当前的 fetch_url 函数只是提取了全部文本。对于技术文档，我们可能只关心与查询最相关的几个段落。可以： - 使用HTML解析器更精准地定位 <main> 、 <article> 或包含特定CSS类的内容区域。 - 在抓取到全文后，使用嵌入模型（如OpenAI的 text-embedding-3-small ）计算查询与文档各段落的相似度，只返回最相关的几个片段。 - 或者，直接使用大模型对抓取到的长文本进行摘要，再返回摘要结果。
3. 支持更多数据源和工具类型 MCP的魅力在于可扩展性。除了搜索网页文档，你可以轻松添加更多工具： - 数据库查询工具 ：连接你的业务数据库，让AI能查询用户数据、订单状态等。 - 内部API工具 ：封装公司内部的CRM、ERP等系统API。 - 代码仓库工具 ：连接Git，让AI能搜索代码库、查看提交历史、甚至创建Pull Request。 - 文件系统工具 ：在安全可控的前提下，让AI能读取指定目录下的文档、配置文件。
4. 构建更强大的Client端 当前的Client只是一个简单的演示。一个成熟的AI应用Client可能包含： - 多模型支持 ：除了OpenAI，可以集成Anthropic Claude、Google Gemini、本地部署的Ollama模型等。 - 复杂的对话管理 ：支持多轮对话、对话记忆、上下文窗口的滑动管理。 - 工具调用策略 ：不仅仅是依赖模型决定，可以加入规则引擎，根据用户意图主动推荐或强制使用某些工具。 - 前端界面 ：构建一个Web或桌面聊天界面，而不仅仅是命令行。
5. 安全与权限控制 当工具能力越来越强时，安全至关重要。 - Server端 ：对传入的 query 和 library 参数进行严格的验证和清洗，防止注入攻击。对可访问的 site: 域名进行白名单限制。 - Client-Server通信 ：为SSE Server增加API密钥认证。使用HTTPS加密传输数据。 - 工具级权限 ：可以为不同的工具或不同的用户设置访问权限。例如，查询内部数据库的工具只能由特定角色的用户触发。
这个项目从零开始，搭建了一个连接AI模型与外部知识（技术文档）的桥梁。通过实现MCP协议，我们不仅解决了一个具体问题，更掌握了一种标准化扩展AI能力的方法。无论是集成到日常使用的IDE中，还是作为微服务部署到云端，MCP都提供了一条清晰、规范的路径。希望这篇详细的实践记录，能帮助你快速上手MCP，并激发出更多将AI与具体业务场景结合的想法。
小龙虾开发者社区是 CSDN 旗下专注 OpenClaw 生态的官方阵地，聚焦技能开发、插件实践与部署教程，为开发者提供可直接落地的方案、工具与交流平台，助力高效构建与落地 AI 应用
更多推荐
6
0- 0
已为社区贡献17条内容
所有评论(0)