实验到实践：抓包分析MCP原理及开发应用(JTFPM22C17000362)
实验到实践：抓包分析 MCP 原理及开发应用(JTFPM22C17000362)
1. 摘要
随着 OpenAI、Manus、字节、阿里等国内外 AI 厂商在 Agent 方向持续发力，调用外部工具 成为 Agent 核心能力之一。有别于 Function Calling 仅处理模型发起的单次函数调用，MCP 允许模型与工具在共享上下文中多轮协作，从而更好地完成复杂任务。本文通过抓包分析的方式，深入剖析 MCP 与 Function Calling 的区别与联系，并提供基于 Python SDK 的 MCP 开发实践指南。
2. 引言
搞清楚 MCP 原理才能更灵活运用，刚接触 MCP 的同学可能都遇到过以下疑惑：
- MCP 和 Function Calling 有什么区别和关系？
- 模型是怎么知道用什么工具的？
- MCP Client 和 MCP Host 什么关系？怎么开发落地？
3. MCP 协议概述
3.1 什么是 MCP？
MCP（模型上下文协议）是由 Anthropic 2024 年 11 月提出的一种通信协议，旨在提供一种让 AI 模型轻松访问和控制外部工具的标准化方式。
3.2 MCP 架构组成
MCP 由三大角色组成：
- MCP Host：用户前端入口，负责协调用户请求并调度模型
- MCP Client：与服务端建立通信通道，将 Host 发来的请求打包、转发给 Server
- MCP Server：提供具体工具能力的模块
3.3 支持的通信协议
MCP 支持多种通信协议，用以适应不同模型厂商的部署环境：
- stdio
- streamable-http
- sse
- websocket
4. MCP 的实际应用场景
注：原文此部分内容为空，此处保留标题结构以备后续补充
5. MCP vs Function Calling
5.1 基本概念对比
Function Calling（函数调用）是 AI 应用服务商为实现工具调用而自定义的接口方式，不同 AI 服务商之间在接口定义和开发文档上存在差异。
MCP 则是将工具调用抽象为客户端-服务器架构，像「USB-C 接口」，定义了 LLM 与外部工具和数据源的通信方式。
5.2 实验环境设置
以天气查询为例，我们通过抓包分析，看下两种方式都做了哪些事情。
本次实验使用 Client 端 HTTP 代理 + Fiddler 作为中间人截获请求信息。
MCP 对比实验：https://github.com/era4d/ai-learning/tree/main/mcp_vs_function_call
也可采用 Client 端 HTTP 代理 + logger 作为中间人，记录日志。
复杂的工具函数调用：https://github.com/MarkTechStation/VideoCode/blob/main/MCP 终极指南 - 番外篇/llm_logger.py
首先，启动天气查询服务：
# weather_api_server.py
from fastapi import FastAPI, Query
from fastapi.responses import JSONResponse
import uvicorn
app = FastAPI()
@app.get("/weather")
def get_weather(city: str = Query("纽约")):
# 只支持纽约，返回默认天气
if city.lower() in ["纽约", "new york"]:
return {"city": "New York", "weather": "Sunny", "temperature": "25°C"}
else:
return JSONResponse(content={"error": f"Only New York supported, got {city}"}, status_code=400)
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=5001)
5.3 Function Calling 方式分析
用户询问「纽约天气怎么样？」时，AI 应用在使用 Function Calling 进行工具调用前，需要先把工具信息定义好、写进配置。
5.3.1 客户端实现
# function_call_client.py
import requests
from langchain_openai import ChatOpenAI
from langchain.tools import BaseTool
from langgraph.prebuilt import create_react_agent
import asyncio
import httpx
# 代理开关，True表示启用代理，False表示禁用代理
enable_proxy = False
class WeatherTool(BaseTool):
name: str = "weather"
description: str = "查询指定城市的天气（目前仅支持纽约）"
def _run(self, city: str = "New York") -> str:
try:
resp = requests.get("http://localhost:5001/weather", params={"city": city})
data = resp.json()
if "error" in data:
return data["error"]
return f"{data['city']} 天气: {data['weather']}, 温度: {data['temperature']}"
except Exception as e:
return f"查询天气失败: {e}"
async def _arun(self, city: str = "New York") -> str:
# 实现异步方法，实际上调用同步方法
return self._run(city)
async def main():
# 初始化天气工具
weather_tool = WeatherTool()
proxy_url = "http://localhost:8888"
# 使用上下文管理器创建异步httpx客户端，根据开关决定是否设置代理
client_kwargs = {"verify": False}
if enable_proxy:
client_kwargs["proxy"] = proxy_url
print("已启用代理:", proxy_url)
else:
print("未启用代理")
async with httpx.AsyncClient(**client_kwargs) as http_client:
# 初始化LLM（需配置OPENAI_API_KEY环境变量）
llm = ChatOpenAI(
api_key="sk-xxx",
model="qwen-max",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
http_async_client=http_client,
model_kwargs={
"tools": [
{
"type": "function",
"function": {
"name": "weather",
"description": "查询指定城市的天气",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，例如：New York"
}
},
"required": ["city"]
}
}
}
]
}
)
# 打印模型信息
print("\n使用的模型:", llm.model_name)
# 创建ReAct代理
agent = create_react_agent(
llm,
tools=[weather_tool]
)
weather_response = await agent.ainvoke(
{"messages": [{"role": "user", "content": "纽约天气怎么样？"}]}
)
final_answer = weather_response["messages"][-1].content
print("代理返回结果:", final_answer)
# 启动异步主函数
if __name__ == "__main__":
asyncio.run(main())
5.3.2 通信流程分析
此时，应用服务端将**工具定义（通信格式）**一并传给大模型。
大模型思考后，给出 tool_calls 指令。
应用端接收到大模型指令后执行了查纽约天气的任务，把结果返回给大模型。
随后大模型把整理后的结果输出给用户。
5.4 MCP 方式分析
而 AI 应用通过 MCP 进行工具调用前，Server 端已经把工具信息封装好，应用只需接入 Server 地址即可获取可用的工具信息。
5.4.1 服务器实现
启动 SSE 模式的 MCP 服务器，通过 @mcp.tool() 快速将函数声明为 MCP 可调用的工具接口。
# mcp_weather_server.py
from fastmcp import FastMCP
import uvicorn
import requests
from typing import Optional
# 创建FastMCP应用实例
mcp = FastMCP("Weather MCP Server")
@mcp.tool
def get_weather(city: str = "纽约") -> str:
"""
查询指定城市的天气信息
Args:
city: 城市名称，默认为纽约
Returns:
天气信息字符串
"""
try:
# 调用本地weather API服务
resp = requests.get("http://localhost:5001/weather", params={"city": city})
data = resp.json()
if "error" in data:
return f"错误: {data['error']}"
return f"{data['city']} 天气: {data['weather']}, 温度: {data['temperature']}"
except Exception as e:
return f"查询天气失败: {str(e)}"
def get_supported_cities() -> str:
"""
获取支持的城市列表
Returns:
支持的城市列表
"""
return "目前仅支持查询纽约(New York)的天气信息"
if __name__ == "__main__":
# 启动SSE模式的MCP服务器
mcp.run(
transport="sse", # 或"http"
host="0.0.0.0",
port=8000,
path="/sse"
)
5.4.2 客户端实现
MCP Client 端将 server 接入，通过 get_tools() 获取可用的工具并完成工具调用。
# mcp_client.py
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
import asyncio
import httpx
# 代理开关，True表示启用代理，False表示禁用代理
enable_proxy = True
async def main():
# 禁用SSL警告
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# 为 MCP 客户端设置代理环境变量
import os
proxy_url = "http://localhost:8888"
if enable_proxy:
os.environ["HTTP_PROXY"] = proxy_url
os.environ["HTTPS_PROXY"] = proxy_url
print(f"已启用代理: {proxy_url}")
else:
# 清除代理环境变量
os.environ.pop("HTTP_PROXY", None)
os.environ.pop("HTTPS_PROXY", None)
print("未启用代理")
# 使用上下文管理器创建异步httpx客户端，根据开关决定是否设置代理
client_kwargs = {"verify": False}
if enable_proxy:
client_kwargs["proxy"] = proxy_url
async with httpx.AsyncClient(**client_kwargs) as http_client:
# 创建LLM实例
llm = ChatOpenAI(
api_key="sk-xxx",
model="qwen-max",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
http_async_client=http_client # 传入带代理的异步httpx客户端
)
client = MultiServerMCPClient(
{
"weather": {
# 确保你在 8000 端口启动天气服务器
"url": "http://localhost:8000/sse",
"transport": "sse",
}
}
)
tools = await client.get_tools()
# 创建ReAct代理
agent = create_react_agent(
llm,
tools
)
weather_response = await agent.ainvoke(
{"messages": [{"role": "user", "content": "what is the weather in 纽约?"}]}
)
final_answer = weather_response["messages"][-1].content
print(final_answer)
# 启动异步主函数
if __name__ == "__main__":
asyncio.run(main())
5.4.3 通信流程分析
MCP Client 发送 3 条请求，通知 Server 初始化已完成，并通过「tools/list」向 Server 索要工具列表。
拿到工具列表后的流程和 Function Calling 一样；将用户问题和工具列表发给大模型，大模型分析后指导 Host 调用工具，最后将结果返回给用户。
5.5 区别与联系总结
最后通过列表总结 Function Calling 和 MCP 的区别与联系。
6. 开发实践
目前很多厂商优秀的 Agent 框架（如：Eino、JManus、LangChain 等）都已集成 MCP 能力。
为了进一步理解 MCP，并将其灵活应用到业务中，我们将采用更原生的方式，基于官方 MCP Python SDK 写一个支持多 MCP Server 的 MCP Client，并提供 logger 和 MCP 协议的通信能力。
MCP Client 示例 https://github.com/era4d/mcp-client-python
6.1 项目结构
项目根目录
├── client.py # 程序主入口，负责加载配置并连接各MCP服务器
├── servers.yaml # 服务器配置文件，定义各MCP Server的连接方式和参数
├── .env # 环境变量配置，如API Key等
├── pyproject.toml # Python项目依赖与元数据
├── core/ # 核心功能模块
│ ├── mcp_client.py # MCP客户端核心逻辑，管理会话、工具调用、与服务器通信
│ ├── context_manager.py # 上下文管理器，负责对话历史、工具调用记录等
│ ├── llm_service.py # LLM服务封装，负责与大模型API交互
│ └── logger.py # 日志系统初始化与管理
├── servers/ # 示例MCP服务器实现
│ ├── calc.py # 计算器服务
│ ├── crawler.py # 网络爬虫服务
│ ├── weather.py # 天气查询服务
│ └── wiki.py # 安全咨询查询服务
├── logs/ # 日志与上下文历史存储
│ ├── context_history.json # 对话历史记录
│ └── mcp_client.log # 运行日志
├── test/ # 测试用例
│ └── test_context.py # 上下文管理器测试
└── .gitignore # Git忽略文件配置
6.2 MCP Server 开发
6.2.1 支持的 SDK
MCP 现支持 C#、Go、Java、TypeScript 等多种语言的 SDK。
项目示例采用 Python SDK，并提供 4 种不同通信协议的 MCP Server 样例。
6.2.2 快速生成 MCP Server
同时可以通过以下 Prompt 快速生成 MCP Server：
# 需求
基于 MCP 相关资料，建一个 MCP Server，需求如下：
- 提供一个查询天气的工具
- 采用sse通信协议
- 要求功能简洁、只包含关键功能
- 使用 Python 编写
# 请访问链接获取MCP server 开发参考资料：
https://modelcontextprotocol.io/quickstart/server
6.2.3 MCP 调试
可以通过 MCP Inspector 对 MCP Server 进行调试跟踪：
# Terminal 运行
npx @modelcontextprotocol/inspector
6.3 MCP Client 开发
Client 端主要包括 Server 连接初始化、process_query 处理用户输入并调用工具、cleanup 关闭连接释放资源。
from mcp.client.stdio import stdio_client
from mcp.client.sse import sse_client
from mcp.client.streamable_http import streamablehttp_client
from mcp.client.websocket import websocket_client
from mcp.client.session import ClientSession
from mcp.types import StdioServerParameters
import asyncio
import logging
logger = logging.getLogger(__name__)
class MCPClient:
...
async def _connect_stdio(self, server: dict):
"""连接 stdio 模式 Server"""
try:
params = StdioServerParameters(
command=server.get("command", "python"),
args=[server["path"]],
env=None
)
logger.info(f"正在连接stdio服务器: {server.get('name')}")
# 添加超时机制，避免无限等待
return await asyncio.wait_for(
self.exit_stack.enter_async_context(stdio_client(params)),
timeout=10.0 # 10秒超时
)
except asyncio.TimeoutError:
logger.error(f"连接stdio服务器超时: {server.get('name')}")
raise
except Exception as e:
logger.error(f"连接stdio服务器失败: {server.get('name')}, 错误: {e}")
import traceback
logger.error(traceback.format_exc())
raise
async def _connect_sse(self, server: dict):
"""连接 SSE 模式 Server"""
url = server.get("url")
if not url:
raise ValueError(f"SSE服务器 {server.get('name')} 未提供URL")
logger.info(f"正在连接SSE服务器: {url}")
try:
return await self.exit_stack.enter_async_context(sse_client(url))
except Exception as e:
logger.error(f"连接SSE服务器失败: {url}, 错误: {e}")
import traceback
logger.error(traceback.format_exc())
raise
async def _connect_streamable_http(self, server: dict):
"""连接 StreamableHTTP 模式 Server"""
url = server.get("url")
if not url:
raise ValueError(f"StreamableHTTP服务器 {server.get('name')} 未提供URL")
headers = server.get("headers", {})
logger.info(f"正在连接StreamableHTTP服务器: {url}")
try:
result = await self.exit_stack.enter_async_context(streamablehttp_client(url, headers=headers))
logger.debug(f"StreamableHTTP连接结果: {result}, 类型: {type(result)}, 长度: {len(result) if hasattr(result, '__len__') else 'N/A'}")
# streamablehttp_client返回(read_stream, write_stream, get_session_id_callback)
# 但ClientSession只需要前两个参数
if isinstance(result, tuple) and len(result) >= 2:
return (result[0], result[1]) # 只返回read_stream和write_stream
else:
raise ValueError(f"StreamableHTTP客户端返回了意外的结果格式: {type(result)}")
except Exception as e:
logger.error(f"连接StreamableHTTP服务器失败: {url}, 错误: {e}")
import traceback
logger.error(traceback.format_exc())
raise
async def _connect_websocket(self, server: dict):
"""连接 WebSocket 模式 Server"""
url = server.get("url")
if not url:
raise ValueError(f"WebSocket服务器 {server.get('name')} 未提供URL")
logger.info(f"正在连接WebSocket服务器: {url}")
try:
return await self.exit_stack.enter_async_context(websocket_client(url))
except Exception as e:
logger.error(f"连接WebSocket服务器失败: {url}, 错误: {e}")
import traceback
logger.error(traceback.format_exc())
raise
7. 模型工具选择机制
最后简单说说模型是怎么知道用什么工具的。
为了让大模型具有工具调用能力，模型在预训练或微调时，采用了数千条工具调用语料来训练模型，格式类似于下面的 json 数据：
{
"conversations": [
{
"from": "human",
"value": "今天天气怎么样?"
},
{
"from": "gpt",
"value": "你想获得某个地方的天气情况，请提供地点信息"
},
{
"from": "human",
"value": "北京"
},
{
"from": "gpt",
"value": "{\n \"function\": \"get_weather\",\n \"arguments\": {\n \"location\": \"北京\"\n }\n}"
}
]
}
8. 总结与展望
以上是 MCP 的原理剖析和应用开发的小试牛刀，后面会持续跟进和更新各领域的创新性应用落地和最佳实践，欢迎关注与交流。