首先进行 Capability Exchange（译者注：Capability Exchange（能力交换）是一种动态服务发现与适配机制，是MCP连接建立的必经步骤，类似于“握手协议”。）
Capability Exchange 流程如下：
(1) 客户端发送初始请求，获取服务器能力信息(2) 服务器返回其能力信息详情(3) 例如当天气 API 服务器被调用时，它可以返回可用的“tools”、“prompts templates”及其他资源供客户端使用
交换完成后，客户端确认连接成功，然后继续交换消息。
Capability Exchange 流程 具体如何实现呢？ MCP协议官方提供了两种主要通信方式：stdio（标准输入输出）和 SSE （Server-Sent Events，服务器发送事件）。
客户端与服务器的通信流程
MCP协议官方提供了两种主要通信方式：stdio（标准输入输出）和 SSE （Server-Sent Events，服务器发送事件）。
这两种方式均采用全双工通信模式，通过独立的读写通道实现服务器消息的实时接收和发送。
什么是SSE Transport？
MCP（Model Context Protocol） 是一个开放协议，旨在标准化应用程序与大型语言模型（LLM）之间的上下文交互。它定义了客户端与服务器如何通过传输层交换消息。
MCP 支持两种标准传输机制：
SSE Transport 是 MCP 中基于 HTTP 的传输方式，利用 SSE 技术实现服务器到客户端的流式消息推送，同时通过 HTTP POST 请求处理客户端到服务器的双向通信。
这种机制特别适合需要实时更新或远程通信的场景。
SSE Transport 的工作原理
SSE（Server-Sent Events）是一种基于 HTTP 协议的服务器推送技术，允许服务器向客户端发送实时更新。
MCP 的 SSE Transport 结合了 SSE 和 HTTP POST，形成了以下工作流程：
SSE Transport 交互流程
1.建立连接：
客户端通过 HTTP GET 请求访问服务器的 SSE 端点（例如 /sse）。
服务器响应一个 text/event-stream 类型的内容，保持连接打开。
服务器发送一个初始的 endpoint 事件，包含一个唯一的 URI（例如 /messages?session_id=xxx），客户端后续通过这个 URI 发送消息。
2.服务器到客户端的消息推送：
3.客户端到服务器的消息发送：
4.连接管理：
数据格式
MCP 使用 JSON-RPC 2.0 协议封装消息，确保请求和响应的结构化处理
SSE 消息遵循 event:\ndata:\n\n 的格式。
示例
客户端 POST 请求：
POST /messages HTTP/1.1
Content-Type: application/json
{"jsonrpc": "2.0", "method": "example", "params": {"text": "Hi"}, "id": 1}
服务器推送：
五、SSE 通信流程详解
1 、建立 SSE 连接
客户端首先需要与服务端建立一个SSE长连接，该连接用于接收服务端推送的所有消息：
http://localhost:8000/sse
提示：
可以直接在浏览器中打开此链接实时查看接收到的消息流
同步使用curl、Postman等工具进行消息发送测试，便于双向操作验证
建立连接后，服务端会首先返回一个专用的消息发送端点（Endpoint），这是后续通信的关键：
event: endpoint
data: /messages/?session_id=2b3c8777119444c1a1b26bc0d0f05a0a
说明：
客户端之后发送的所有消息请求，都必须指向这个专用Endpoint (其实就是专用的 url)。
当请求格式正确时，该Endpoint会返回一个”Accepted”状态码，但是，真正的服务器响应内容都是通过SSE端点异步推送的。
2、 通信初始化流程
客户端需要完成两个关键的初始化步骤，确保通信渠道正常建立：
a) 发送初始化请求
客户端需要先向服务端发送初始化请求，表明自身身份和期望的通信参数：
curl -X POST 'http://localhost:8000/messages/?session_id=2b3c8777119444c1a1b26bc0d0f05a0a' -H 'Content-Type: application/json' -d '{
"jsonrpc": "2.0",
"id": 0,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "mcp",
"version": "0.1.0"
}
}
}'
服务端会返回其支持的功能列表，便于客户端了解可用的交互能力：
event: message
data: {
"jsonrpc": "2.0",
"id": 0,
"result": {
"protocolVersion": "2024-11-05",
"capabilities": {
"experimental": {},
"prompts": {
"listChanged": false
},
"resources": {
"subscribe": false,
"listChanged": false
},
"tools": {
"listChanged": false
}
},
"serverInfo": {
"name": "weather",
"version": "1.3.0"
}
}
}
b) 发送初始化完成通知
初始化协商完成后，客户端需要明确通知服务端,初始化已完成：
curl -X POST 'http://localhost:8000/messages/?session_id=2b3c8777119444c1a1b26bc0d0f05a0a' -H 'Content-Type: application/json' -d '{
"jsonrpc": "2.0",
"method": "notifications/initialized",
"params": {}
}'
3、 工具调用示例
MCP协议的核心价值在于其工具调用能力，以下是一个完整的调用流程：
a) 获取可用工具列表
客户端首先应该获取服务端提供的工具列表，如果已经知道其工具参数也可跳过，直接去调用工具：
curl -X POST 'http://localhost:8000/messages/?session_id=2b3c8777119444c1a1b26bc0d0f05a0a' -H 'Content-Type: application/json' -d '{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/list",
"params": {}
}'
服务端会返回所有可用的工具及其参数规范：
event: message
data: {
"jsonrpc": "2.0",
"id": 1,
"result": {
"tools": [{
"name": "get_alerts",
"description": "Get weather alerts for a US state.\n\n Args:\n state: Two-letter US state code (e.g. CA, NY)\n ",
"inputSchema": {
"properties": {
"state": {
"title": "State",
"type": "string"
}
},
"required": ["state"],
"title": "get_alertsArguments",
"type": "object"
}
}, {
"name": "get_forecast",
"description": "Get weather forecast for a location.\n\n Args:\n latitude: Latitude of the location\n longitude: Longitude of the location\n ",
"inputSchema": {
"properties": {
"latitude": {
"title": "Latitude",
"type": "number"
},
"longitude": {
"title": "Longitude",
"type": "number"
}
},
"required": ["latitude", "longitude"],
"title": "get_forecastArguments",
"type": "object"
}
}]
}
}
b) 调用特定工具
客户端根据需求，选择并调用特定工具：
curl -X POST 'http://localhost:8000/messages/?session_id=2b3c8777119444c1a1b26bc0d0f05a0a' -H 'Content-Type: application/json' -d '{
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {"name": "get_alerts", "arguments": {"state": "CA"}}
}'
服务端执行工具调用并返回结果：
event: message
data: {
"jsonrpc": "2.0",
"id": 2,
"result": {
"content": [{
"type": "text",
"text": "\nEvent: Wind Advisory\nArea: San Gorgonio Pass Near Banning\nSeverity: Moderate\nDescription: * WHAT... \n"
}],
"isError": false
}
}
4、通信特点
1 双向异步通信 ： 接收通道通过 SSE 长连接接收服务端消息，发送通道通过 HTTP POST 向专用端点发送客户端消息
2 会话状态管理 ：使用 session_id 维护客户端与服务端的会话状态，支持多客户端并发连接
3 标准通信协议 ：基于 JSON-RPC 2.0 规范，确保通信格式统一，便于跨平台
4 可扩展性 ：支持动态发现和调用服务端工具
MCP协议通过其灵活而高效的通信机制，为客户端与服务端之间的交互提供了可靠保障。基于SSE的实现方案不仅便于开发和调试，还具备良好的性能表现。
对于开发者而言，深入理解MCP通信模式不仅有助于实现高质量的客户端应用，也为设计其他分布式通信系统提供了宝贵经验。希望本文的详细解析能为您的技术实践提供有益参考。
5、 注意事项
(1) 所有客户端请求必须使用服务端分配的Endpoint
(2) 初始化流程必须严格按照顺序完成
六、MCP官方给出的代码示例
我们可以分析一段MCP官方给出的代码示例
https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/clients/simple-chatbot/mcp_simple_chatbot/main.py#L196
all_tools = []
for server in self.servers: tools = await server.list_tools() all_tools.extend(tools)
tools_description = "\n".join([tool.format_for_llm() for tool in all_tools])
system_message = ("You are a helpful assistant with access to these tools:\n\n"
f "{tools_description}\n"
"Choose the appropriate tool based on the user's question. "
"If no tool is needed, reply directly.\n\n"
"IMPORTANT: When you need to use a tool, you must ONLY respond with "
"the exact JSON object format below, nothing else:\n"
"{\n"
' "tool": "tool-name",\n'
' "arguments": {\n'
' "argument-name": "value"\n'
" }\n"
"}\n\n"
"After receiving a tool's response:\n"
"1. Transform the raw data into a natural, conversational response\n"
"2. Keep responses concise but informative\n"
"3. Focus on the most relevant information\n"
"4. Use appropriate context from the user's question\n"
"5. Avoid simply repeating the raw data\n\n"
"Please use only the tools that are explicitly defined above.")
方便起见，我们翻译一些系统提示词：
您是一个有用的助手， 可以访问这些工具：｛ tools_description｝ 根据用户的问题选择合适的工具。 如果不需要工具， 请直接回复。
重要提示： 当您需要使用工具时， 您必须只回复下面确切的JSON对象格式， 不回复别的内容： {
"tool": "tool-name",
"arguments": {
"argument-name": "value"
}
}
收到执行结果后：
(1) 将原始数据转换为自然的对话式响应
2.保持回答简洁但信息丰富
3.关注最相关的信息
4.使用用户问题中的适当上下文
5.避免简单地重复原始数据
你只能使用上面明确定义的工具。
看到这里，你是不是恍然大悟。
MCP，实际上就是把函数的介绍写在了系统提示词里。
对于Function Calling来说，我需要把函数的参数填到请求参数tools里面，这些函数最终以什么样的方式输入到模型里，这个过程是不透明的，解析tools的工作是在云端进行的，返回的格式也是在云端定义的。我们只能通过官方文档来确定返回参数的格式。
而MCP很巧妙的绕过了云端解析tools的过程。
MCP把函数的介绍、参数的返回格式都写在了系统提示里面，这样一来，只要LLM能够理解系统提示词，就能够按照提示词定义的格式输出。
这就实现了格式的统一。
在理解了MCP的原理后，我们就能很清晰地解释MCP的服务端和 客户端 了：
?MCP服务端 ， 是执行函数工具的场所。
?MCP客户端， 是与LLM交互的场所，包括介绍函数和获取函数的参数
MCP服务端中常用的函数工具、数据查询工具等已经由社区的开发者们开发好了，可以直接进行调用。
对于大多数的开发者来说，需要开发的是MCP客户端。
而在MCP客户端中，开发者根据业务流程，让LLM选择合适的函数工具进行调用，最终实现业务逻辑。
七、LLM大 模型是如何确定工具的选用的？
在学习的过程中，我一直好奇一个问题：Claude（模型）是在什么时候确定使用哪些工具的呢？
Anthropic 在官网为我们提供了一个 简单的解释 ：
当用户提出一个问题时：
（1）客户端（Claude Desktop / Cursor）将你的问题发送给 Claude。
（2）Claude 分析可用的工具，并决定使用哪一个（或多个）。
（3）MCP Client 客户端 通过 MCP Server 执行所选的工具。
（4）工具的执行结果被送回给 Claude。
（5）Claude 结合执行结果构造最终的 prompt 并生成自然语言的回应。
（6）回应最终展示给用户！
MCP Server 是由 Claude 主动选择并调用的。
有意思的是 Claude 具体是如何确定该使用哪些工具呢？以
及是否会使用一些不存在的工具呢（幻觉）？
为了探索这个问题让我们深入源码。
显然这个调用过程可以分为两个步骤：
（1）由 LLM（Claude）确定使用哪些 MCP Server。
（2）执行对应的 MCP Server 并对执行结果进行重新处理。
先给出一个简单可视化帮助理解：
第一步模型如何确定该使用哪些工具
先理解第一步模型如何确定该使用哪些工具？
这里以 MCP 官方提供的 client example[8] 为讲解示例，并简化了对应的代码（删除了一些不影响阅读逻辑的异常控制代码）。
通过阅读代码，可以发现模型是通过 prompt 来确定当前有哪些工具。我们通过将工具的具体使用描述以文本的形式传递给模型，供模型了解有哪些工具以及结合实时情况进行选择。
参考代码中的注释：
... // 省略了无关的代码
async def start(self):
// 初始化所有的 mcp server
for server in self.servers:
await server.initialize()
// 获取所有的 tools 命名为 all_tools
all_tools = []
for server in self.servers:
tools = await server.list_tools()
all_tools.extend(tools)
// 将所有的 tools 的功能描述格式化成字符串供 LLM 使用
// tool.format_for_llm() 我放到了这段代码最后，方便阅读。
tools_description = "\n".join(
[tool.format_for_llm() for tool in all_tools]
)
// 这里就不简化了，以供参考，实际上就是基于 prompt 和当前所有工具的信息
// 询问 LLM（Claude） 应该使用哪些工具。
system_message = (
"You are a helpful assistant with access to these tools:\n\n"
f"{tools_description}\n"
"Choose the appropriate tool based on the user's question. "
"If no tool is needed, reply directly.\n\n"
"IMPORTANT: When you need to use a tool, you must ONLY respond with "
"the exact JSON object format below, nothing else:\n"
"{\n"
' "tool": "tool-name",\n'
' "arguments": {\n'
' "argument-name": "value"\n'
" }\n"
"}\n\n"
"After receiving a tool's response:\n"
"1. Transform the raw data into a natural, conversational response\n"
"2. Keep responses concise but informative\n"
"3. Focus on the most relevant information\n"
"4. Use appropriate context from the user's question\n"
"5. Avoid simply repeating the raw data\n\n"
"Please use only the tools that are explicitly defined above."
)
messages = [{"role": "system", "content": system_message}]
while True:
// Final... 假设这里已经处理了用户消息输入.
messages.append({"role": "user", "content": user_input})
// 将 system_message 和用户消息输入一起发送给 LLM
llm_response = self.llm_client.get_response(messages)
... // 后面和确定使用哪些工具无关
class Tool:
"""Represents a tool with its properties and formatting."""
def init(
self, name: str, description: str, input_schema: dict[str, Any]
) -> None:
self.name: str = name
self.description: str = description
self.input_schema: dict[str, Any] = input_schema
// 把工具的名字 / 工具的用途（description）和工具所需要的参数（args_desc）转化为文本
def format_for_llm(self) -> str:
"""Format tool information for LLM.
Returns:
A formatted string describing the tool.
"""
args_desc = []
if "properties" in self.input_schema:
for param_name, param_info in self.input_schema["properties"].items():
arg_desc = (
f"- {param_name}: {param_info.get('description', 'No description')}"
)
if param_name in self.input_schema.get("required", []):
arg_desc += " (required)"
args_desc.append(arg_desc)
return f"""
Tool: {self.name}
Description: {self.description}
Arguments:
{chr(10).join(args_desc)}
那 tool 的描述和代码中的 input_schema 是从哪里来的呢？
通过进一步分析 MCP 的 Python SDK 源代码可以发现：大部分情况下，当使用装饰器 @mcp.tool() 来装饰函数时，对应的 name 和 description 等其实直接源自用户定义函数的函数名以及函数的 docstring 等。
这里仅截取一小部分片段，想了解更多请参考原始代码 。
@classmethod
def from_function(
cls,
fn: Callable,
name: str | None = None,
description: str | None = None,
context_kwarg: str | None = None,
) -> "Tool":
"""Create a Tool from a function."""
func_name = name or fn.name # 获取函数名
if func_name == "<lambda>":
raise ValueError("You must provide a name for lambda functions")
func_doc = description or fn.doc or "" # 获取函数 docstring
is_async = inspect.iscoroutinefunction(fn)
... 更多请参考原始代码...
总结：模型是通过 prompt engineering，即提供所有工具的结构化描述和 few-shot 的 example 来确定该使用哪些工具。
另一方面，Anthropic 肯定对 Claude 做了专门的训练（毕竟是自家协议，Claude 更能理解工具的 prompt 以及输出结构化的 tool call json 代码）
第二步:工具执行与结果反馈机制
其实工具的执行就比较简单和直接了。承接上一步，我们把 system prompt（指令与工具调用描述）和用户消息一起发送给模型，然后接收模型的回复。当模型分析用户请求后，它会决定是否需要调用工具：
（1）无需工具时：模型直接生成自然语言回复。
（2）需要工具时：模型输出结构化 JSON 格式的工具调用请求。
如果回复中包含结构化 JSON 格式的工具调用请求，则客户端会根据这个 json 代码执行对应的工具。
具体的实现逻辑都在 process_llm_response 中，[代码10]，逻辑非常简单。
如果模型执行了 tool call，则工具执行的结果 result 会和 system prompt 和用户消息一起重新发送给模型，请求模型生成最终回复。
如果 tool call 的 json 代码存在问题或者模型产生了幻觉怎么办呢？通过阅读[代码11] 发现，我们会 skip 掉无效的调用请求。
执行相关的代码与注释如下：
... // 省略无关的代码
async def start(self):
... // 上面已经介绍过了，模型如何选择工具
while True:
// 假设这里已经处理了用户消息输入.
messages.append({"role": "user", "content": user_input})
// 获取 LLM 的输出
llm_response = self.llm_client.get_response(messages)
// 处理 LLM 的输出（如果有 tool call 则执行对应的工具）
result = await self.process_llm_response(llm_response)
// 如果 result 与 llm_response 不同，说明执行了 tool call （有额外信息了）
// 则将 tool call 的结果重新发送给 LLM 进行处理。
if result != llm_response:
messages.append({"role": "assistant", "content": llm_response})
messages.append({"role": "system", "content": result})
final_response = self.llm_client.get_response(messages)
logging.info("\nFinal response: %s", final_response)
messages.append(
{"role": "assistant", "content": final_response}
)
// 否则代表没有执行 tool call，则直接将 LLM 的输出返回给用户。
else:
messages.append({"role": "assistant", "content": llm_response})
结合这部分原理分析：
（1）工具文档至关重要 - 模型通过工具描述文本来理解和选择工具，因此精心编写工具的名称、docstring 和参数说明至关重要。
（2）由于 MCP 的选择是基于 prompt 的，所以任何模型其实都适配 MCP，只要你能提供对应的工具描述。但是当你使用非 Claude 模型时，MCP 使用的效果和体验难以保证（没有做专门的训练）。
MCP (Model Context Protocol) 总结
MCP (Model Context Protocol) 代表了 AI 与外部工具和数据交互的标准建立。通过本文，我们可以了解到：
1 MCP 的本质：它是一个统一的协议标准，使 AI 模型能够以一致的方式连接各种数据源和工具，类似于 AI 世界的"USB-C"接口。
2 MCP 的价值：它解决了传统 function call 的平台依赖问题，提供了更统一、开放、安全、灵活的工具调用机制，让用户和开发者都能从中受益。
3 使用与开发：对于普通用户，MCP 提供了丰富的现成工具，用户可以在不了解任何技术细节的情况下使用；对于开发者，MCP 提供了清晰的架构和 SDK，使工具开发变得相对简单。
MCP 还处于发展初期，但其潜力巨大。
更重要的是生态吧，基于统一标准下构筑的生态也会正向的促进整个领域的发展。
八、MCP (Model Context Protocol) 实战
我们按照官方 github modelcontextprotocol/python-sdk 实际操作一下，这里我们仅开发 MCP Server， MCP Client 直接使用 Claude Desktop App.
我们会为 Claude Desktop App 增加个 tools: calculate_bmi 和 fetch_weather， 默认 Claude Desktop App 是不具备计算 BMI 和查询天气的功能。
实现 MCP Server （Python SDK）
代码实现
// server.py
import httpx
from mcp.server.fastmcp import FastMCP
// Create an MCP server
mcp = FastMCP("Demo")
@mcp.tool()
def calculate_bmi(weight_kg: float, height_m: float) -> float:
"""Calculate BMI given weight in kg and height in meters"""
return weight_kg / (height_m ** 2)
@mcp.tool()
async def fetch_weather(city: str) -> str:
"""Fetch current weather for a city"""
async with httpx.AsyncClient() as client:
response = await client.get(f"https://api.weather.com/{city}")
return response.text
if name == "main":
mcp.run()
实现 MCP Client
MCP 客户端充当 LLM 和 MCP 服务器之间的桥梁，MCP 客户端的工作流程如下：
MCP 客户端首先从 MCP 服务器获取可用的工具列表。
将用户的查询连同工具描述通过 function calling 一起发送给 LLM。
LLM 决定是否需要使用工具以及使用哪些工具。
如果需要使用工具，MCP 客户端会通过 MCP 服务器执行相应的工具调用。
工具调用的结果会被发送回 LLM。
LLM 基于所有信息生成自然语言响应。
最后将响应展示给用户。
MCP Client调用工具
在 MCP 客户端中，我们使用了以下两个函数来与 MCP 服务器进行交互。
list_tools()：获取 MCP 服务器提供的所有可用工具。
call_tool(name, args)：调用指定的工具并获取结果，这里调用 list_indices 来获取 Elasticsearch 集群中的索引信息。
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
async def run():
// 建立连接
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
// 初始化连接
await session.initialize()
// 列出可用工具
tools = await session.list_tools()
print("Tools:", tools)
// 调用工具
indices = await session.call_tool("list_indices")
print("Indices:", indices)
if name == "main":
import asyncio
asyncio.run(run())
集成 LLM调用 MCP Client
在实际应用中，我们通常希望让 LLM（如 OpenAI、Claude、 通义千问 等）自主决定调用哪些工具。
下面的代码将以通义千问为示例进行演示，并使用 OpenAI SDK 与其交互。
为了简化这一过程，我们将借助 OpenRouter （一个统一的 LLM 网关，它提供了 OpenAI 兼容的接口），使我们能够通过相同的 OpenAI API 访问包括通义千问在内的多种 LLM。
OpenRouter 的使用方式非常简单。
我们只需在创建 OpenAI 客户端时指定 OpenRouter 的 base_url 和 api_key，并在调用模型时以 <provider>/<model> 的格式（例如 qwen/qwen-plus）指定目标模型，OpenRouter 就会根据模型名称自动将请求路由到对应的 LLM 上。
除此之外，其他代码与标准的 OpenAI SDK 保持一致。
接下来介绍一下 MCP 客户端的主要代码。
初始化 MCP Client 客户端类
MCPClient 类的初始化包含以下 3 个组件：
1 self.session：用于存储与 MCP 服务器的会话对象，初始设为 None，将在连接服务器时被赋值。
2 self.exit_stack：使用 AsyncExitStack 来管理异步资源，确保所有资源（如服务器连接、会话等）在程序结束时能够正确关闭。
3 self.client：创建 OpenAI 异步客户端，通过 OpenRouter 来访问 LLM。
这里我们：
设置 base_url 为 OpenRouter 的 API 端点。
从环境变量获取 API Key（请确保设置了 OPENROUTER_API_KEY 环境变量）。
class MCPClient:
def init(self):
self.session: Optional[ClientSession] = None
self.exit_stack = AsyncExitStack()
self.client = AsyncOpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.getenv("OPENROUTER_API_KEY"),
)
MCP Client 和 MCP 服务器连接
connect_to_server 方法负责建立与 MCP 服务器的连接。它首先配置服务器进程的启动参数，然后通过 stdio_client 建立双向通信通道，最后创建并初始化会话。
所有的资源管理都通过 AsyncExitStack 来处理，确保资源能够正确释放。
连接成功后，它会打印出 MCP 服务器提供的所有可用工具。
async def connect_to_server(self, server_script_path: str):
server_params = StdioServerParameters(
command="python",
args=[server_script_path],
env=None
)
stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
self.stdio, self.write = stdio_transport
self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
await self.session.initialize()
# 列出可用工具
response = await self.session.list_tools()
tools = response.tools
print("\nConnected to server with tools:", [tool.name for tool in tools])
处理请求
定义一个 process_query 方法，完成 处理请求的流程：
首先，将用户的查询作为初始消息发送给 LLM，同时提供 MCP 服务器上所有可用工具的描述信息。
LLM 分析用户查询，决定是直接回答还是需要调用工具。如果需要工具，它会指定要调用的工具名称和参数。
对于每个工具调用，MCP 客户端执行调用并收集结果。
将工具调用的结果返回给 LLM，让它基于这些新信息生成或更新回答。
如果 LLM 认为还需要更多信息，它会继续请求调用其他工具。
这个过程会一直重复，直到 LLM 收集了足够的信息来完整回答用户的查询。
async def process_query(self, query: str) -> str:
"""使用 LLM 和 MCP 服务器提供的工具处理查询"""
messages = [
{
"role": "user",
"content": query
}
]
response = await self.session.list_tools()
available_tools = [{
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.inputSchema
}
} for tool in response.tools]
// 初始化 LLM API 调用
response = await self.client.chat.completions.create(
model="qwen/qwen-plus",
messages=messages,
tools=available_tools
)
final_text = []
message = response.choices[0].message
final_text.append(message.content or "")
// 处理响应并处理工具调用
while message.tool_calls:
// 处理每个工具调用
for tool_call in message.tool_calls:
tool_name = tool_call.function.name
tool_args = json.loads(tool_call.function.arguments)
// 执行工具调用
result = await self.session.call_tool(tool_name, tool_args)
final_text.append(f"[Calling tool {tool_name} with args {tool_args}]")
// 将工具调用和结果添加到消息历史
messages.append({
"role": "assistant",
"tool_calls": [
{
"id": tool_call.id,
"type": "function",
"function": {
"name": tool_name,
"arguments": json.dumps(tool_args)
}
}
]
})
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": str(result.content)
})
// 将工具调用的结果交给 LLM
response = await self.client.chat.completions.create(
model="qwen/qwen-plus",
messages=messages,
tools=available_tools
)
message = response.choices[0].message
if message.content:
final_text.append(message.content)
return "\n".join(final_text)
九、MCP (Model Context Protocol) 生态
MCP 生态不断发展壮大，越来越多的应用支持 MCP，同时开放平台也提供 MCP Server。
同时也有像 Cloudflare 、 Composio 、 Zapier 使用 SSE 方式将 MCP 进行托管（即接入一个 MCP Endpoint 即接入一批 MCP Servers），通过 Stdio 方式最理想场景是 MCP Servers 和 Agent 系统跑在同一 Docker 容器中（类似 Sidecar 模式）。
实践下来，MCP Server SSE 并不是理想的方案，因为需要保持连接和 session 状态，而云服务（如 FaaS）更倾向于无状态架构 ，所以最近提出了更适配云场景的 Streamable HTTP Transport。
如果 MCP 成为未来的规范，那么 Agent 应用能否准确调用各个 MCP，将成为模型 RL 未来需要支持的关键功能。与 Function Call 模型不同，MCP 是一个动态的工具库，模型需要具备对新增 MCP 的泛化理解能力。
虽然目前 LLM 和上下文之间建立了标准化的通信协议，但 Agent 之间的交互协议尚未形成统一标准，Agent 服务发现、恢复、监控等一系列生产级问题等解决。目前 ANP （Agent Network Protocol）在做这方面的探索与尝试。