大模型学习之MCP
MCP
RAG 技术的局限性
RAG（检索增强生成，Retrieval-Augmented Generation）是当前大模型领域的热门方向。它结合了信息检索与生成式模型，旨在提升知识准确性、上下文理解和对最新信息的利用能力。
RAG 的主要缺点：
- 检索精度有限，可能无法获取最相关的信息
- 生成内容可能不完整或片面
- 缺乏全局视角
- 检索能力受限时效果下降
理论基础
Function Call
Function Call 是 OpenAI 于 2023 年提出的重要概念，本质上为大模型提供了与外部系统交互的能力，相当于为模型配备“外挂工具箱”。当模型无法直接回答问题时，可主动调用预设函数（如查询天气、计算数据、访问数据库等），获取实时且精准的信息后再生成回答。
Model Context Protocol（MCP）
MCP（模型上下文协议）由 Anthropic 公司提出，是一个开放标准协议，旨在解决 AI 模型与外部数据源、工具的交互难题。
开发者按照 MCP 协议开发，无需为每个模型与不同资源重复编写适配代码，大幅节省开发工作量。MCP Server 作为通用协议实现，可直接开放给开发者使用，进一步减少重复劳动。
核心特点：
- 协议标准化：统一工具调用格式（请求/响应/错误处理）
- 生态兼容性：一次开发对接所有兼容 MCP 的模型
- 动态扩展：新增工具无需修改模型代码，即插即用
核心价值：
- 数据孤岛 → 打通本地/云端数据源
- 重复开发 → 工具开发者只需适配 MCP 协议
- 生态割裂 → 形成统一工具市场
MCP 与 Function Call 对比
| 对比项 | Function Call (OpenAI) | MCP (Anthropic) |
|---|---|---|
| 标准化 | 仅适用于 OpenAI 生态 | 开放标准，兼容多模型 |
| 工具接入 | 需为每个模型单独适配 | 一次开发，多模型复用 |
| 扩展性 | 新增工具需适配各模型 | 工具即插即用，无需修改模型代码 |
| 生态兼容性 | 主要服务于 OpenAI 平台 | 支持多家模型厂商 |
| 典型应用场景 | 聊天机器人、自动化助手 | 跨平台工具市场、企业级数据集成 |
简而言之，Function Call 更偏向于单一平台的工具调用，而 MCP 致力于打造跨平台、跨模型的统一工具生态。
MCP 的基本使用
MCP 客户端（Host）
MCP 客户端（Host）负责与 MCP Server 通信，发起请求并处理响应。开发者可通过 Host 集成 MCP 协议，实现模型与外部工具的数据交互。
MCP Client
MCP Client 是具体实现 MCP 协议的客户端库，支持多种编程语言。开发者可根据需求选择合适的 Client，快速集成 MCP 能力。
MCP Server
MCP Server 官方描述：一个轻量级程序，通过标准化模型上下文协议公开特定功能。
目前的局限性
- 避免让 AI 检索过大数据量。与 RAG 不同，MCP 每次会检索所需全部数据，若一次查询数据量过大，会消耗大量 Token，甚至导致客户端卡死。
- 许多 MCP 客户端依赖大量系统提示词与 MCP 工具通讯，因此使用 MCP 时 Token 消耗会显著增加。
- MCP 生态尚处于早期阶段，工具和资源数量有限，部分场景下可用性有待提升。
- 不同模型对 MCP 协议的支持程度不一，实际集成时需关注兼容性问题。
MCP开发步骤
推荐使用uv进行python管理
然后是fastmcp
uv add “mcp[cli]” httpx openai
以下基本 服务端格式
1 |
|
关于客户端调试可以用Inspector
也可以直接使用postman
然后我们在使用的时候一般是需要一个客户端的
自建客户端代码
import asyncio
import json
import logging
from typing import Dict, List, Any, Optional
from mcp import ClientSession
from mcp.client.sse import sse_client
from openai import OpenAI
# 屏蔽干扰日志
logging.getLogger("httpx").setLevel(logging.WARNING)
class MCPGatewayClient:
def __init__(self, config: Dict):
self.server_configs = config.get("mcpServers", {})
self.sessions: Dict[str, ClientSession] = {}
self.contexts = []
self.tool_to_server = {}
self.resource_to_server = {}
self.prompt_to_server = {}
self.prompts_metadata: List[Dict] = []
async def start(self):
print(f"\n{'='*70}")
print(f"🚀 MCP 智能网关启动中... (正在连接 {len(self.server_configs)} 个节点)")
print(f"{'='*70}")
for name, cfg in self.server_configs.items():
try:
ctx = sse_client(cfg["url"])
read, write = await ctx.__aenter__()
self.contexts.append(ctx)
session = ClientSession(read, write)
await session.__aenter__()
await session.initialize()
self.sessions[name] = session
print(f"\n📂 服务器: 【{name}】")
tools_res = await session.list_tools()
for t in tools_res.tools: self.tool_to_server[str(t.name)] = name
print(f" ├─ 🛠️ Tools: {', '.join([str(t.name) for t in tools_res.tools]) or 'None'}")
res_res = await session.list_resources()
for r in res_res.resources: self.resource_to_server[str(r.uri)] = name
print(f" ├─ 📖 Resources: {', '.join([str(r.uri) for r in res_res.resources]) or 'None'}")
prompts_res = await session.list_prompts()
for p in prompts_res.prompts:
p_name = str(p.name)
self.prompt_to_server[p_name] = name
self.prompts_metadata.append({
"name": p_name,
"description": getattr(p, 'description', '专业领域指令'),
"server": name
})
print(f" └─ 💡 Prompts: {', '.join([str(p.name) for p in prompts_res.prompts]) or 'None'}")
except Exception as e:
print(f"❌ 服务器 [{name}] 失败: {str(e)}")
print(f"\n{'='*70}\n✅ 网关就绪。\n")
async def get_combined_tools(self) -> List[Dict]:
all_tools = []
for name, session in self.sessions.items():
mcp_tools = await session.list_tools()
for t in mcp_tools.tools:
all_tools.append({
"type": "function",
"function": {
"name": str(t.name),
"description": t.description or "",
"parameters": t.inputSchema
}
})
if self.resource_to_server:
all_tools.append({
"type": "function",
"function": {
"name": "read_mcp_resource",
"description": f"读取内部资料。路径: {list(self.resource_to_server.keys())}",
"parameters": {
"type": "object",
"properties": {"uri": {"type": "string"}},
"required": ["uri"]
}
}
})
return all_tools
def robust_parse_args(self, args: Any) -> Dict:
"""解决模型可能产生的双重 JSON 编码问题"""
if isinstance(args, dict):
return args
if isinstance(args, str):
try:
parsed = json.loads(args)
# 如果解析后还是字符串，说明是双重编码，再解一次
if isinstance(parsed, str):
return json.loads(parsed)
return parsed
except Exception:
return {"raw_input": args}
return {}
async def call_action(self, name: str, args: Any):
# 统一在入口处进行鲁棒性解析
parsed_args = self.robust_parse_args(args)
try:
if name == "read_mcp_resource":
uri = parsed_args.get("uri", "").strip()
srv = self.resource_to_server.get(uri)
if not srv: return f"Error: 找不到资源 {uri}"
res = await self.sessions[srv].read_resource(uri)
return res.contents[0].text
else:
srv = self.tool_to_server.get(name)
if not srv: return f"Error: 找不到工具 {name}"
# 确保传入的是 dict，否则 Pydantic 会抛出你遇到的错误
res = await self.sessions[srv].call_tool(name, parsed_args)
return res.content[0].text if hasattr(res, 'content') else str(res)
except Exception as e:
return f"执行失败: {str(e)}"
async def get_prompt_template(self, prompt_name: str, arguments: Dict[str, str] = None) -> Optional[str]:
srv_name = self.prompt_to_server.get(prompt_name)
if not srv_name: return None
try:
session = self.sessions[srv_name]
# 将参数传递给 get_prompt 接口
result = await session.get_prompt(prompt_name, arguments=arguments or {})
if result.messages:
# 提取渲染后的文本内容
return result.messages[0].content.text
return None
except Exception as e:
print(f"⚠️ 获取 Prompt 【{prompt_name}】 失败: {str(e)}")
return None
async def stop(self):
print("\n正在释放 MCP 资源...")
# 显式关闭 Session 再关闭 Context，防止 SSE 异常抛出
for s in self.sessions.values():
try: await s.__aexit__(None, None, None)
except: pass
for c in reversed(self.contexts):
try: await c.__aexit__(None, None, None)
except: pass
print("✅ 连接已安全关闭。")
async def run_task(user_input: str):
config = {"mcpServers": {
"math_server": {"url": "http://127.0.0.1:8029/sse"},
"ticket_server": {"url": "http://127.0.0.1:8030/sse"},
"data_server": {"url": "http://127.0.0.1:8031/sse"},
}}
client = MCPGatewayClient(config)
await client.start()
try:
from config import OPENAI_BASE_URL, OPENAI_API_KEY, LLM_MODEL
oai = OpenAI(api_key=OPENAI_API_KEY, base_url=OPENAI_BASE_URL)
# --- 🧠 语义路由 (强化对复合任务的识别) ---
system_instruction = "你是一个全能助手。"
if client.prompts_metadata:
router_prompt = f"我有以下专家角色库：{json.dumps(client.prompts_metadata, ensure_ascii=False)}。用户的问题涉及文档审阅、逻辑判断或数据分析吗？如果是，请返回对应的模板名称(name)，否则返回 NONE。用户问题：{user_input}"
resp = oai.chat.completions.create(model=LLM_MODEL, messages=[{"role": "user", "content": router_prompt}])
decision = resp.choices[0].message.content.strip()
if "NONE" not in decision:
# 提取 prompt 名称（可能包含额外文本）
prompt_name = decision.split()[0] if decision.split() else decision
# 传递必需的参数：data_analyst_expert 需要 topic 参数
p_content = await client.get_prompt_template(prompt_name, arguments={"topic": user_input})
if p_content:
system_instruction = p_content
print(f"🎯 路由命中: 加载【{prompt_name}】模板")
# --- 🏗️ ReAct 循环 ---
messages = [{"role": "system", "content": system_instruction}, {"role": "user", "content": user_input}]
print(f"🚀 任务开始 (system_instruction: {system_instruction})")
for i in range(5):
tools = await client.get_combined_tools()
response = oai.chat.completions.create(model=LLM_MODEL, messages=messages, tools=tools)
msg = response.choices[0].message
messages.append(msg)
if not msg.tool_calls:
print(f"\n🏁 最终回答:\n{msg.content}")
break
print(f"\n🔄 第 {i+1} 轮：模型请求 {len(msg.tool_calls)} 个并发操作")
# 记录并发任务
call_tasks = []
tool_call_info = []
for tc in msg.tool_calls:
call_tasks.append(client.call_action(tc.function.name, tc.function.arguments))
tool_call_info.append(tc)
results = await asyncio.gather(*call_tasks)
# 执行完所有任务后统一打印结果，分割线加长
for tc, res_text in zip(tool_call_info, results):
print(f"{'-'*40}\n▶️ {tc.function.name}\n参数: {tc.function.arguments[:100]}...\n◀️ 结果: {str(res_text)[:120].replace(chr(10),' ')}\n{'-'*40}")
messages.append({"role": "tool", "tool_call_id": tc.id, "content": res_text})
except Exception as e:
print(f"\n❌ 运行中发生错误: {str(e)}")
finally:
await client.stop()
if __name__ == "__main__":
query = "帮我审阅一下 IT 手册里的安全准则，并帮我查一下成都的天气，如果成都的天气晴朗，就帮小王订一张从今天晚上12点北京飞往成都的机票，如果北京的天气不晴朗，就算了。"
# query = "帮我审阅一下 IT 手册里的安全准则，并帮我查一下北京的天气，如果北京的天气晴朗，就帮小王订一张从今天晚上12点成都飞往北京的机票，如果北京的天气不晴朗，就算了。"
# query = "我们来讨论下大数定律"
asyncio.run(run_task(query))
本博客所有文章除特别声明外，均采用 CC BY-NC-SA 4.0 许可协议。转载请注明来源 廾匸！
