MCP协议深度解析

MCP协议深度解析：大模型工具调用新标准Model Context Protocol全面解读，从原理到实战前言2024年底，Anthropic发布了**MCP（Model Context Protocol）**协议，这是一个开放标准，旨在统一大模型与外部工具、数据源的交互方式。如果你用过OpenAI的Function Calling、LangChain的Tools，你可能会问：为什么还需要MCP？答案是：互操作性。MCP让任何模型都能调用任何工具，不再被特定平台锁定。你将学到：MCP协议的核心设计理念协议架构与通信机制如何开发MCP Server如何开发MCP Client实际应用案例与Function Calling的对比一、为什么需要MCP？1.1 当前工具调用的碎片化# OpenAI的方式response=openai.chat.completions.create(model="gpt-4",messages=[...],tools=[{"type":"function","function":{"name":"get_weather","description":"获取天气","parameters":{...}}}])# Anthropic的方式response=anthropic.messages.create(model="claude-3",messages=[...],tools=[{"name":"get_weather","description":"获取天气","input_schema":{...}}])# LangChain的方式fromlangchain.toolsimporttool@tooldefget_weather(city:str)-str:"""获取天气"""...问题：每个平台的工具定义格式不同切换模型需要重写工具代码工具无法跨平台复用没有统一的服务发现机制1.2 MCP的解决方案# MCP统一方式# 工具定义一次，任何模型都能调用frommcpimportTool,ToolResult@Tool(name="get_weather",description="获取指定城市的天气信息")defget_weather(city:str)-ToolResult:weather=fetch_weather(city)returnToolResult(content=weather)# 任何支持MCP的模型都能调用这个工具MCP的核心价值：标准化：统一的协议规范互操作性：工具跨平台复用服务发现：自动发现可用工具安全性：统一的权限控制二、MCP协议架构2.1 核心组件┌─────────────────┐ ┌─────────────────┐ │ MCP Client │────▶│ MCP Server │ │ (模型/应用) │◀────│ (工具提供方) │ └─────────────────┘ └─────────────────┘ │ │ ▼ ▼ ┌─────────────────┐ ┌─────────────────┐ │ 模型推理引擎 │ │ 工具/数据源 │ │ (LLM Runtime) │ │ (External) │ └─────────────────┘ └─────────────────┘MCP Client：集成在应用或模型中负责发现和调用工具处理工具返回结果MCP Server：提供工具实现暴露工具接口处理工具调用请求2.2 通信协议MCP支持两种传输方式：# 1. stdio（标准输入输出）- 本地工具# 适合：本地文件操作、系统命令# 2. HTTP + SSE（Server-Sent Events）- 远程工具# 适合：网络API、云服务# stdio方式示例importsysimportjsondefread_message():"""从stdin读取消息"""line=sys.stdin.readline()returnjson.loads(line)defwrite_message(message):"""向stdout写入消息"""sys.stdout.write(json.dumps(message)+"\n")sys.stdout.flush()# HTTP+SSE方式示例fromfastapiimportFastAPIfromsse_starlette.sseimportEventSourceResponse app=FastAPI()@app.post("/mcp/v1/call")asyncdefcall_tool(request:dict):"""处理工具调用"""result=awaitexecute_tool(request)returnresult2.3 消息格式// 请求格式{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"get_weather","arguments":{"city":"北京"}}}// 响应格式{"jsonrpc":"2.0","id":1,"result":{"content":[{"type":"text","text":"北京今天晴，25°C"}]}}// 错误格式{"jsonrpc":"2.0","id":1,"error":{"code":-32600,"message":"Invalid Request"}}三、开发MCP Server3.1 基础Server实现# server.pyfrommcp.serverimportServerfrommcp.typesimportTool,TextContent,ToolResultimportjson# 创建Server实例server=Server("weather-server")# 定义工具@server.tool()asyncdefget_weather(city:str)-ToolResult:"""获取指定城市的天气信息 Args: city: 城市名称，如"北京"、"上海" """# 模拟天气API调用weather_data={"北京":{"temp":25,"condition":"晴"},"上海":{"temp":28,"condition":"多云"},"广州":{"temp":32,"condition":"雷阵雨"}}data=weather_data.get(city,{"temp":20,"condition":"未知"})returnToolResult(content=[TextContent(type="text",text=f"{city}今天{data['condition']}，温度{data['temp']}°C")])@server.tool()asyncdefcalculate(expression:str)-ToolResult:"""计算数学表达式 Args: expression: 数学表达式，如 "2+3*4" """try:# 安全的数学计算result=eval(expression,{"__builtins__":{}},{})returnToolResult(content=[TextContent(type="text",text=str(result))])exceptExceptionase:returnToolResult(content=[TextContent(type="text",text=f"计算错误:{str(e)}")],isError=True)# 运行Serverif__name__=="__main__":importasyncio asyncio.run(server.run_stdio())3.2 高级Server功能frommcp.serverimportServerfrommcp.typesimport(Tool,TextContent,ImageContent,EmbeddedResource,ToolResult)fromtypingimportList,Optionalimportbase64 server=Server("advanced-server")# 带复杂参数的工具@server.tool()asyncdefsearch_documents(query:str,limit:int=10,filters:Optional[dict]=None)-ToolResult:"""搜索文档 Args: query: 搜索关键词 limit: 返回结果数量限制 filters: 过滤条件，如 {"type": "pdf", "date_after": "2024-01-01"} """results=awaitdo_search(query,limit,filters)content=[]fordocinresults:content.append(TextContent(type="text",text=f"标题:{doc['title']}\n摘要:{doc['summary']}\n链接:{doc['url']}"))returnToolResult(content=content)# 返回图片的工具@server.tool()asyncdefgenerate_chart(data:dict,chart_type:str)-ToolResult:"""生成图表 Args: data: 图表数据 chart_type: 图表类型，如 "bar", "line", "pie" """# 生成图表image_bytes=awaitcreate_chart(data,chart_type)returnToolResult(content=[ImageContent(type="image",data=base64.b64encode(image_bytes).decode(),mimeType=