1. 从“调API”到“造轮子”为什么我们需要亲手实现一个MCP Server如果你最近在AI应用开发领域尤其是围绕Claude、Cursor这类智能编码工具那么“MCP”这个词一定高频地出现在你的视野里。Model Context Protocol这个由Anthropic提出的协议正在迅速成为连接大模型与外部工具、数据源的事实标准。市面上已经涌现了成百上千个MCP Server从搜索、文件读写到数据库操作几乎无所不包。那么问题来了既然有这么多现成的轮子为什么我还要费劲“从零手搓”一个MCP Server呢这恰恰是“工程化”思维与“调包侠”思维的分水岭。直接调用现成的MCP Server就像使用一个封装好的黑盒函数你只知道输入和输出对其内部的工作机制、性能瓶颈、安全边界一无所知。当工具的行为不符合预期或者你需要一个高度定制化的功能时这种“拿来主义”就会让你束手无策。而亲手实现一遍哪怕是一个最简单的Server其价值也远超你的想象。这个过程会让你彻底理解MCP协议的核心——它如何定义工具Tools、资源ResourcesServer与Client比如Claude Desktop之间是如何通过JSON-RPC进行通信的消息的序列化、反序列化、错误处理、生命周期管理又是如何运作的这些底层细节是你在调用mcp-client库时永远无法触及的。更重要的是MCP的定位是“工程化”的粘合剂。一个成熟的AI应用绝不是简单地问答。它需要能稳定、安全、高效地调用企业内部API、查询专有知识库、执行自动化流程。理解MCP就是掌握了为AI大模型“装配手脚”和“扩展感官”的核心方法。通过这次实战我们的目标不是造一个能上生产环境的复杂Server而是像拆解一台精密的钟表一样把MCP的每一个齿轮、每一根发条都看清楚然后自己动手把它们组装起来让它重新滴答作响。当你完成时你将获得的不仅是一个可运行的代码更是一套能够自主设计、调试和优化任何MCP集成方案的“元能力”。2. 庖丁解牛拆解MCP协议的核心三要素在动手写代码之前我们必须像建筑师看蓝图一样彻底理解MCP协议的设计哲学。它本质上是一个基于JSON-RPC 2.0的通信协议其核心思想是将外部能力抽象为两类实体工具Tools和资源Resources并通过一个标准化的Server暴露给Client通常是AI助手。2.1 工具ToolsAI的“可调用函数”工具是MCP中最核心的概念。你可以把它理解为一个函数签名告诉AI“我能做什么”。每个工具必须明确定义name: 工具的唯一标识符例如get_weather。description: 工具功能的自然语言描述。这部分至关重要因为AI主要靠它来决定是否以及如何调用这个工具。描述应清晰、具体包含输入参数的预期。inputSchema: 定义调用此工具所需的参数遵循JSON Schema格式。这严格规定了AI必须提供什么样的数据才能成功调用。例如一个获取天气的工具定义可能如下所示在Server初始化时声明{ name: get_weather, description: 获取指定城市的当前天气信息。, inputSchema: { type: object, properties: { city: { type: string, description: 城市名称例如Beijing, Shanghai }, unit: { type: string, enum: [celsius, fahrenheit], description: 温度单位默认为摄氏度celsius } }, required: [city] } }当AIClient决定使用这个工具时它会向Server发送一个tools/call请求其中包含name和匹配inputSchema的arguments。Server收到后执行实际逻辑如调用第三方天气API然后将结果通过tools/call响应返回给AI。注意工具描述的质量直接决定了AI使用的准确性。模糊的描述会导致AI误用或不用。好的描述应像一份精简的API文档。2.2 资源ResourcesAI的“可读取文档”资源代表了Client可以读取的静态或动态内容比如文件、数据库查询结果、API返回的文本等。与工具不同资源是“只读”的AI不能通过资源执行操作但可以获取其内容来丰富上下文。资源的关键属性包括uri: 资源的唯一标识符格式类似file:///path/to/doc.md或dynamic://stock/TSLA。mimeType: 内容的媒体类型如text/plain,text/markdown,application/json。name和description: 便于AI理解资源内容的元数据。Server通过resources/list和resources/read请求来向Client宣告和提供资源内容。例如一个Server可以声明一个动态资源dynamic://system/time当AI需要知道当前时间时Client会请求读取这个资源Server则实时生成当前时间的文本返回。2.3 Server与Client基于JSON-RPC的对话机制MCP Server是一个独立的进程通过标准输入输出stdio或HTTP等传输层与Client通信。所有的交互都封装在JSON-RPC 2.0的消息中。一个典型的启动和交互流程如下初始化Initialization: Client启动Server后首先发送initialize请求。Server回复其支持的协议版本、能力如支持哪些工具和资源列表以及一个唯一的serverId。能力宣告Capabilities Announcement: 在初始化完成后Server会主动或响应Client请求发送notifications或响应requests来告知Client当前可用的工具列表tools/list和资源列表resources/list。工具调用Tool Invocation: AI在对话中判断需要调用工具时Client会向Server发送tools/call请求。Server执行实际工作并返回tools/call响应其中包含结果或错误信息。资源读取Resource Reading: 类似地Client发送resources/read请求来获取特定URI的资源内容。心跳与关闭Keep-alive Shutdown: 为了管理连接协议通常还包含ping/pong或自定义的心跳机制以及优雅关闭的shutdown请求。理解了这个通信模型我们就知道编写一个MCP Server主要工作就是监听特定的JSON-RPC请求根据请求类型method执行对应的业务逻辑然后返回格式正确的JSON-RPC响应。接下来我们就用代码将这个蓝图变为现实。3. 实战第一步搭建最小可行MCP Server骨架我们选择使用Python来实现因为它生态丰富、上手简单并且有不错的MCP基础库支持。这里我们不直接使用最高抽象的框架如mcp而是从更底层的sse-starlette和pydantic入手这样能更清晰地看到每一部分是如何工作的。3.1 项目初始化与依赖安装首先创建一个新的项目目录并初始化虚拟环境这是保证依赖隔离的好习惯。mkdir my-first-mcp-server cd my-first-mcp-server python -m venv venv # 在Windows上使用 venv\Scripts\activate source venv/bin/activate接着安装核心依赖。我们将使用fastapi和sse-starlette来方便地处理HTTP和服务器发送事件Server-Sent Events这是MCP over HTTP的一种常见传输方式。pydantic用于严谨的数据验证和序列化。pip install fastapi sse-starlette pydantic3.2 定义协议数据模型用Pydantic塑造“坚固的管道”MCP协议中流动的所有数据——请求、响应、通知——都有严格的结构。使用Pydantic模型来定义它们相当于为我们的数据管道加上了编译时类型检查能极大减少低级错误。我们在models.py文件中定义几个最核心的模型from typing import Any, Dict, List, Optional, Union from pydantic import BaseModel, Field # JSON-RPC 2.0 基础请求结构 class JSONRPCRequest(BaseModel): jsonrpc: str Field(2.0, constTrue) id: Optional[Union[int, str]] None method: str params: Optional[Dict[str, Any]] None # JSON-RPC 2.0 基础响应/错误结构 class JSONRPCResponse(BaseModel): jsonrpc: str Field(2.0, constTrue) id: Optional[Union[int, str]] None result: Optional[Any] None error: Optional[Dict[str, Any]] None # MCP 工具定义 class Tool(BaseModel): name: str description: str inputSchema: Dict[str, Any] # JSON Schema # MCP 资源定义 class Resource(BaseModel): uri: str name: str description: Optional[str] None mimeType: str text/plain # MCP 初始化参数 class InitializeParams(BaseModel): protocolVersion: str 2024-11-05 clientInfo: Optional[Dict[str, str]] None # MCP 工具调用参数 class CallToolParams(BaseModel): name: str arguments: Optional[Dict[str, Any]] None # MCP 读取资源参数 class ReadResourceParams(BaseModel): uri: str这些模型就像乐高积木的模具确保了我们将要组装和传递的每一个数据块都是形状正确的。3.3 实现Server核心循环处理请求的路由器现在我们来创建Server的主文件server.py。它的核心是一个能根据JSON-RPC请求的method字段将请求路由到对应处理函数的路由器。from fastapi import FastAPI, Request from sse_starlette.sse import EventSourceResponse import asyncio import json from models import * app FastAPI() # 存储当前可用的工具和资源 available_tools: List[Tool] [] available_resources: List[Resource] [] async def handle_jsonrpc_request(request_data: Dict) - Dict: 核心请求处理器 try: req JSONRPCRequest(**request_data) except Exception as e: # 如果连基础请求结构都不对返回解析错误 return JSONRPCResponse( idNone, error{code: -32700, message: Parse error, data: str(e)} ).dict() handler_map { initialize: handle_initialize, tools/list: handle_tools_list, tools/call: handle_tools_call, resources/list: handle_resources_list, resources/read: handle_resources_read, # 可以继续添加其他method的处理函数 } handler handler_map.get(req.method) if not handler: return JSONRPCResponse( idreq.id, error{code: -32601, message: fMethod not found: {req.method}} ).dict() try: result await handler(req.params, req.id) # 处理函数应返回一个符合JSONRPCResponse结构的字典 return result except Exception as e: # 处理函数内部出错 return JSONRPCResponse( idreq.id, error{code: -32603, message: Internal error, data: str(e)} ).dict() # 定义各个请求的处理函数暂时留空 async def handle_initialize(params: Optional[Dict], request_id: Any) - Dict: pass async def handle_tools_list(params: Optional[Dict], request_id: Any) - Dict: pass async def handle_tools_call(params: Optional[Dict], request_id: Any) - Dict: pass async def handle_resources_list(params: Optional[Dict], request_id: Any) - Dict: pass async def handle_resources_read(params: Optional[Dict], request_id: Any) - Dict: pass # HTTP端点用于接收Client的POST请求标准JSON-RPC over HTTP app.post(/jsonrpc) async def jsonrpc_endpoint(request: Request): body await request.json() response await handle_jsonrpc_request(body) return response # SSE端点用于Client建立长连接Server可以主动推送通知如工具列表更新 app.get(/sse) async def sse_endpoint(request: Request): async def event_generator(): # 这里可以维护一个连接队列当有通知时推送给所有连接的Client # 例如当工具列表更新时发送一个 tools/list 通知 # 为了简单我们先返回一个空的事件流 while True: if await request.is_disconnected(): break # 可以在这里检查是否有需要推送的通知 # 暂时只发送一个保持连接的心跳注释 yield {event: comment, data: heartbeat} await asyncio.sleep(30) # 30秒心跳 return EventSourceResponse(event_generator())这个骨架搭建了起来。它有一个HTTP端点/jsonrpc来接收请求一个SSE端点/sse用于服务器主动推送。handle_jsonrpc_request函数是大脑负责解析请求并分发给对应的处理函数。现在我们需要为这些处理函数注入灵魂。4. 注入灵魂实现核心工具与资源处理器骨架已经搭好现在需要实现具体的业务逻辑。我们来实现两个最经典的功能一个计算器工具和一个动态时间资源。4.1 实现初始化与列表查询首先在Server启动时我们需要注册我们的工具和资源。修改server.py的顶部和初始化函数# 在文件顶部定义我们的工具和资源 available_tools [ Tool( namecalculate, description执行简单的数学计算。支持加()、减(-)、乘(*)、除(/)。, inputSchema{ type: object, properties: { expression: { type: string, description: 数学表达式例如3 5 * 2。注意乘号是*除号是/。 } }, required: [expression] } ) ] available_resources [ Resource( uridynamic://server/current_time, name当前服务器时间, description获取服务器当前的日期和时间。, mimeTypetext/plain ) ] async def handle_initialize(params: Optional[Dict], request_id: Any) - Dict: 处理初始化请求 init_params InitializeParams(**(params or {})) # 这里可以检查protocolVersion是否兼容 response JSONRPCResponse( idrequest_id, result{ protocolVersion: init_params.protocolVersion, capabilities: { tools: {listChanged: True}, # 告知Client工具列表可能会变 resources: {listChanged: True} # 告知Client资源列表可能会变 }, serverInfo: { name: My First MCP Server, version: 0.1.0 } } ) return response.dict() async def handle_tools_list(params: Optional[Dict], request_id: Any) - Dict: 返回当前可用的工具列表 response JSONRPCResponse( idrequest_id, result{tools: [tool.dict() for tool in available_tools]} ) return response.dict() async def handle_resources_list(params: Optional[Dict], request_id: Any) - Dict: 返回当前可用的资源列表 response JSONRPCResponse( idrequest_id, result{resources: [resource.dict() for resource in available_resources]} ) return response.dict()初始化处理函数handle_initialize向Client宣告了Server的基本信息和能力。tools/list和resources/list处理函数则直接返回我们预定义好的列表。注意capabilities中的listChanged字段设为True这意味着我们的Server支持在运行时动态更新列表并会通过SSE通道发送通知本例暂未实现动态更新。4.2 实现计算器工具调用这是核心中的核心。当AI发送tools/call请求要求调用calculate工具时我们需要验证参数是否符合schema。安全地执行计算表达式这是重点和难点。返回结果或错误。import ast import operator # 安全评估表达式的辅助函数 def safe_eval_expression(expr: str): 极其有限且安全地评估一个只包含数字和基础运算符的字符串表达式。 警告绝对不要在生产环境中用eval()直接执行用户输入 这里使用ast.literal_eval也有局限我们实现一个简单的解析器作为示例。 # 一个非常简陋的、仅用于演示的解析器按空格分割假设是逆波兰表达式或简单二元运算 # 例如只处理 a op b 形式如 3 5 tokens expr.split() if len(tokens) ! 3: raise ValueError(表达式格式暂只支持 数字 运算符 数字如 3 5) try: a float(tokens[0]) b float(tokens[2]) except ValueError: raise ValueError(操作数必须是数字) op_map { : operator.add, -: operator.sub, *: operator.mul, /: operator.truediv } op_func op_map.get(tokens[1]) if not op_func: raise ValueError(f不支持的运算符: {tokens[1]}。支持: , -, *, /) if tokens[1] / and b 0: raise ZeroDivisionError(除数不能为零) result op_func(a, b) # 如果是整数返回整数形式 if result.is_integer(): return int(result) return result async def handle_tools_call(params: Optional[Dict], request_id: Any) - Dict: 处理工具调用请求 if not params: return JSONRPCResponse( idrequest_id, error{code: -32602, message: Invalid params} ).dict() call_params CallToolParams(**params) if call_params.name calculate: # 1. 获取参数 arguments call_params.arguments or {} expression arguments.get(expression) if not expression: return JSONRPCResponse( idrequest_id, error{code: -32602, message: Missing required argument: expression} ).dict() # 2. 执行计算在真实场景中这里需要更复杂和安全的方法 try: # 注意这里使用自定义的安全函数而非eval result_value safe_eval_expression(expression) except ZeroDivisionError: return JSONRPCResponse( idrequest_id, error{code: -32000, message: Calculation error, data: Division by zero.} ).dict() except Exception as e: return JSONRPCResponse( idrequest_id, error{code: -32000, message: Calculation error, data: str(e)} ).dict() # 3. 返回成功结果 response JSONRPCResponse( idrequest_id, result{ content: [ { type: text, text: f表达式 {expression} 的计算结果是{result_value} } ] } ) return response.dict() # 如果工具名未找到 return JSONRPCResponse( idrequest_id, error{code: -32601, message: fTool not found: {call_params.name}} ).dict()这里有一个至关重要的安全警告在真实的生产环境中绝对禁止使用Python内置的eval()函数来执行用户AI提供的表达式字符串这会带来严重的代码注入安全风险。上面的safe_eval_expression是一个极度简化的示例仅用于演示原理。在实际项目中你需要使用更安全的数学表达式解析库如asteval它利用AST进行有限制评估或者将计算任务委托给一个严格沙箱化的环境。4.3 实现动态时间资源读取资源读取的逻辑相对简单主要是根据请求的URI生成或获取对应的内容。from datetime import datetime async def handle_resources_read(params: Optional[Dict], request_id: Any) - Dict: 处理资源读取请求 if not params: return JSONRPCResponse( idrequest_id, error{code: -32602, message: Invalid params} ).dict() read_params ReadResourceParams(**params) if read_params.uri dynamic://server/current_time: # 动态生成当前时间 current_time datetime.now().strftime(%Y-%m-%d %H:%M:%S) content_text f当前服务器时间是{current_time} response JSONRPCResponse( idrequest_id, result{ contents: [ { uri: read_params.uri, mimeType: text/plain, text: content_text } ] } ) return response.dict() # 如果资源URI未找到 return JSONRPCResponse( idrequest_id, error{code: -32601, message: fResource not found: {read_params.uri}} ).dict()至此一个具备最小功能集的MCP Server就实现了。它能够响应初始化、列表查询、工具调用和资源读取请求。你可以使用uvicorn来运行它uvicorn server:app --reload --port 8000Server将在http://localhost:8000启动。接下来我们需要一个Client来测试它。5. 验证与调试打造一个简易的MCP Client测试器为了验证我们的Server是否正常工作我们不能只依赖Claude Desktop。自己写一个简单的测试Client是理解和调试MCP协议的最佳方式。这个Client会模拟标准MCP Client如Claude的行为向我们的Server发送请求并打印响应。创建test_client.pyimport asyncio import aiohttp import json async def test_mcp_server(): server_url http://localhost:8000 async with aiohttp.ClientSession() as session: print(1. 发送初始化请求...) init_request { jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 2024-11-05, clientInfo: {name: TestClient} } } async with session.post(f{server_url}/jsonrpc, jsoninit_request) as resp: init_result await resp.json() print(f初始化响应: {json.dumps(init_result, indent2, ensure_asciiFalse)}) print(\n2. 查询工具列表...) tools_list_request { jsonrpc: 2.0, id: 2, method: tools/list, params: {} } async with session.post(f{server_url}/jsonrpc, jsontools_list_request) as resp: tools_result await resp.json() print(f工具列表: {json.dumps(tools_result, indent2, ensure_asciiFalse)}) print(\n3. 调用计算器工具...) call_tool_request { jsonrpc: 2.0, id: 3, method: tools/call, params: { name: calculate, arguments: { expression: 10 * 2 3 } } } async with session.post(f{server_url}/jsonrpc, jsoncall_tool_request) as resp: call_result await resp.json() print(f工具调用结果: {json.dumps(call_result, indent2, ensure_asciiFalse)}) print(\n4. 查询资源列表...) resources_list_request { jsonrpc: 2.0, id: 4, method: resources/list, params: {} } async with session.post(f{server_url}/jsonrpc, jsonresources_list_request) as resp: resources_result await resp.json() print(f资源列表: {json.dumps(resources_result, indent2, ensure_asciiFalse)}) print(\n5. 读取时间资源...) read_resource_request { jsonrpc: 2.0, id: 5, method: resources/read, params: { uri: dynamic://server/current_time } } async with session.post(f{server_url}/jsonrpc, jsonread_resource_request) as resp: read_result await resp.json() print(f资源内容: {json.dumps(read_result, indent2, ensure_asciiFalse)}) if __name__ __main__: asyncio.run(test_mcp_server())运行这个测试脚本确保Server已在运行python test_client.py你应该能看到一系列格式规范的JSON响应。如果一切顺利在调用calculate工具时你会收到一个包含计算结果文本的响应在读取时间资源时会得到当前的服务器时间。这个过程让你清晰地看到了MCP协议下“请求-响应”的完整数据流。6. 进阶与生产化思考从玩具到工具我们的“手搓”Server已经能跑通了但它离一个健壮、可用的生产级MCP Server还有巨大差距。这一步的思考才是“工程化”的真正开始。6.1 传输层与部署不止于HTTP我们使用了HTTPSSE这是MCP的一种传输方式transport: sse。但MCP官方还支持标准的stdio标准输入输出这对于与Claude Desktop等本地应用集成更为常见。这意味着你的Server需要能够从sys.stdin读取JSON-RPC请求并将响应写入sys.stdout。你需要修改Server的启动入口根据环境变量如MCP_TRANSPORT来决定使用哪种传输方式。对于生产部署你可能会将Server打包成Docker容器并通过stdio与宿主机上的AI助手通信。6.2 连接管理与状态维护我们的示例Server是无状态的每次请求都是独立的。但在真实场景中会话SessionClient和Server之间可能维持一个会话包含一些上下文信息。资源订阅Resource SubscriptionClient可以订阅某个资源当其内容变化时Server需要通过SSE主动推送更新通知notifications/resources/updated。这要求Server维护资源的状态和Client的订阅列表。工具列表动态更新同样当Server安装或移除了一个工具需要广播notifications/tools/list_changed通知。6.3 安全性、错误处理与日志输入验证与消毒我们强调了计算器工具的安全问题这只是一个缩影。所有来自AI的输入都必须视为不可信的需要进行严格的验证和消毒防止注入攻击。身份验证与授权如果你的Server连接了内部数据库或敏感API那么必须实现身份验证。MCP协议本身不规定认证方式这需要你在传输层如HTTP头添加API Key或应用层自行实现。全面的错误处理我们的示例只有基础错误。一个健壮的Server需要对各种边界情况网络超时、第三方API失败、无效参数组合等定义清晰的错误码和友好的错误信息并通过JSON-RPC的error字段返回。结构化日志为了方便运维和调试所有重要的操作收到请求、调用工具、发生错误都应该被记录并包含请求ID、工具名、耗时等上下文信息。6.4 性能与可观测性异步与并发使用asyncio如我们所用或其它异步框架来处理并发请求避免阻塞。指标Metrics暴露Prometheus格式的指标端点监控请求量、延迟、错误率。跟踪Tracing集成OpenTelemetry追踪一个用户请求从AI发出经过MCP Server调用下游服务再返回的完整链路。7. 整合到真实环境在Claude Desktop中连接你的Server最后让我们把亲手打造的Server用起来。以Claude Desktop为例你需要编辑其配置文件来添加我们的自定义Server。在macOS上配置文件通常位于~/Library/Application Support/Claude/claude_desktop_config.json。在Windows上位于%APPDATA%\Claude\claude_desktop_config.json。你需要添加一个mcpServers配置项。由于我们的Server使用HTTP配置可能如下所示注意Claude Desktop默认更倾向于stdio对SSE的支持可能需要特定版本或配置{ mcpServers: { my-calculator-server: { command: uvicorn, args: [ server:app, --host, 0.0.0.0, --port, 8000 ], env: { PYTHONPATH: /path/to/your/project } // 或者如果你的Server已经作为常驻进程运行可以配置为使用SSE传输 // url: http://localhost:8000/sse, // transport: sse } } }更常见的做法是让Server支持stdio传输。这意味着你需要修改Server使其能够从命令行启动并通过标准输入输出进行通信。这通常涉及解析sys.argv监听sys.stdin的输入并写入sys.stdout。许多MCP SDK如Python的mcp库已经帮你处理了这部分样板代码。配置完成后重启Claude Desktop。在对话中你应该能看到Claude已经识别到了新的工具。你可以尝试对它说“请用我的计算器工具计算一下(15 - 3) * 4 的结果。” 如果一切配置正确Claude会调用你的Server并返回计算结果。这个过程可能会遇到各种问题环境变量路径不对、端口冲突、协议版本不匹配、Claude Desktop缓存了旧的工具列表等等。调试的关键在于查看Claude Desktop的日志通常可以在其设置中找到日志文件路径以及确保你的Server日志是详细且可读的。这就是为什么前面强调日志和错误处理的重要性——当集成出现问题时清晰的日志是你唯一的救生索。从亲手解析第一个JSON-RPC请求到安全地实现一个工具再到最终与AI助手成功联动这个完整的闭环体验就是理解MCP工程化精髓的最佳路径。你不再只是一个API的调用者而是成为了扩展AI能力边界的构建者。下次当你再看到Tavily Search MCP、Brave Search MCP这些复杂的Server时你看到的将不再是一个黑盒而是一个个由类似我们今天搭建的骨架填充了不同业务逻辑后形成的、有生命力的服务。这就是“彻底搞懂”之后世界在你眼中呈现出的不同模样。 从零手搓MCP Server：深入理解AI工具扩展协议与Python实战
1. 从“调API”到“造轮子”为什么我们需要亲手实现一个MCP Server如果你最近在AI应用开发领域尤其是围绕Claude、Cursor这类智能编码工具那么“MCP”这个词一定高频地出现在你的视野里。Model Context Protocol这个由Anthropic提出的协议正在迅速成为连接大模型与外部工具、数据源的事实标准。市面上已经涌现了成百上千个MCP Server从搜索、文件读写到数据库操作几乎无所不包。那么问题来了既然有这么多现成的轮子为什么我还要费劲“从零手搓”一个MCP Server呢这恰恰是“工程化”思维与“调包侠”思维的分水岭。直接调用现成的MCP Server就像使用一个封装好的黑盒函数你只知道输入和输出对其内部的工作机制、性能瓶颈、安全边界一无所知。当工具的行为不符合预期或者你需要一个高度定制化的功能时这种“拿来主义”就会让你束手无策。而亲手实现一遍哪怕是一个最简单的Server其价值也远超你的想象。这个过程会让你彻底理解MCP协议的核心——它如何定义工具Tools、资源ResourcesServer与Client比如Claude Desktop之间是如何通过JSON-RPC进行通信的消息的序列化、反序列化、错误处理、生命周期管理又是如何运作的这些底层细节是你在调用mcp-client库时永远无法触及的。更重要的是MCP的定位是“工程化”的粘合剂。一个成熟的AI应用绝不是简单地问答。它需要能稳定、安全、高效地调用企业内部API、查询专有知识库、执行自动化流程。理解MCP就是掌握了为AI大模型“装配手脚”和“扩展感官”的核心方法。通过这次实战我们的目标不是造一个能上生产环境的复杂Server而是像拆解一台精密的钟表一样把MCP的每一个齿轮、每一根发条都看清楚然后自己动手把它们组装起来让它重新滴答作响。当你完成时你将获得的不仅是一个可运行的代码更是一套能够自主设计、调试和优化任何MCP集成方案的“元能力”。2. 庖丁解牛拆解MCP协议的核心三要素在动手写代码之前我们必须像建筑师看蓝图一样彻底理解MCP协议的设计哲学。它本质上是一个基于JSON-RPC 2.0的通信协议其核心思想是将外部能力抽象为两类实体工具Tools和资源Resources并通过一个标准化的Server暴露给Client通常是AI助手。2.1 工具ToolsAI的“可调用函数”工具是MCP中最核心的概念。你可以把它理解为一个函数签名告诉AI“我能做什么”。每个工具必须明确定义name: 工具的唯一标识符例如get_weather。description: 工具功能的自然语言描述。这部分至关重要因为AI主要靠它来决定是否以及如何调用这个工具。描述应清晰、具体包含输入参数的预期。inputSchema: 定义调用此工具所需的参数遵循JSON Schema格式。这严格规定了AI必须提供什么样的数据才能成功调用。例如一个获取天气的工具定义可能如下所示在Server初始化时声明{ name: get_weather, description: 获取指定城市的当前天气信息。, inputSchema: { type: object, properties: { city: { type: string, description: 城市名称例如Beijing, Shanghai }, unit: { type: string, enum: [celsius, fahrenheit], description: 温度单位默认为摄氏度celsius } }, required: [city] } }当AIClient决定使用这个工具时它会向Server发送一个tools/call请求其中包含name和匹配inputSchema的arguments。Server收到后执行实际逻辑如调用第三方天气API然后将结果通过tools/call响应返回给AI。注意工具描述的质量直接决定了AI使用的准确性。模糊的描述会导致AI误用或不用。好的描述应像一份精简的API文档。2.2 资源ResourcesAI的“可读取文档”资源代表了Client可以读取的静态或动态内容比如文件、数据库查询结果、API返回的文本等。与工具不同资源是“只读”的AI不能通过资源执行操作但可以获取其内容来丰富上下文。资源的关键属性包括uri: 资源的唯一标识符格式类似file:///path/to/doc.md或dynamic://stock/TSLA。mimeType: 内容的媒体类型如text/plain,text/markdown,application/json。name和description: 便于AI理解资源内容的元数据。Server通过resources/list和resources/read请求来向Client宣告和提供资源内容。例如一个Server可以声明一个动态资源dynamic://system/time当AI需要知道当前时间时Client会请求读取这个资源Server则实时生成当前时间的文本返回。2.3 Server与Client基于JSON-RPC的对话机制MCP Server是一个独立的进程通过标准输入输出stdio或HTTP等传输层与Client通信。所有的交互都封装在JSON-RPC 2.0的消息中。一个典型的启动和交互流程如下初始化Initialization: Client启动Server后首先发送initialize请求。Server回复其支持的协议版本、能力如支持哪些工具和资源列表以及一个唯一的serverId。能力宣告Capabilities Announcement: 在初始化完成后Server会主动或响应Client请求发送notifications或响应requests来告知Client当前可用的工具列表tools/list和资源列表resources/list。工具调用Tool Invocation: AI在对话中判断需要调用工具时Client会向Server发送tools/call请求。Server执行实际工作并返回tools/call响应其中包含结果或错误信息。资源读取Resource Reading: 类似地Client发送resources/read请求来获取特定URI的资源内容。心跳与关闭Keep-alive Shutdown: 为了管理连接协议通常还包含ping/pong或自定义的心跳机制以及优雅关闭的shutdown请求。理解了这个通信模型我们就知道编写一个MCP Server主要工作就是监听特定的JSON-RPC请求根据请求类型method执行对应的业务逻辑然后返回格式正确的JSON-RPC响应。接下来我们就用代码将这个蓝图变为现实。3. 实战第一步搭建最小可行MCP Server骨架我们选择使用Python来实现因为它生态丰富、上手简单并且有不错的MCP基础库支持。这里我们不直接使用最高抽象的框架如mcp而是从更底层的sse-starlette和pydantic入手这样能更清晰地看到每一部分是如何工作的。3.1 项目初始化与依赖安装首先创建一个新的项目目录并初始化虚拟环境这是保证依赖隔离的好习惯。mkdir my-first-mcp-server cd my-first-mcp-server python -m venv venv # 在Windows上使用 venv\Scripts\activate source venv/bin/activate接着安装核心依赖。我们将使用fastapi和sse-starlette来方便地处理HTTP和服务器发送事件Server-Sent Events这是MCP over HTTP的一种常见传输方式。pydantic用于严谨的数据验证和序列化。pip install fastapi sse-starlette pydantic3.2 定义协议数据模型用Pydantic塑造“坚固的管道”MCP协议中流动的所有数据——请求、响应、通知——都有严格的结构。使用Pydantic模型来定义它们相当于为我们的数据管道加上了编译时类型检查能极大减少低级错误。我们在models.py文件中定义几个最核心的模型from typing import Any, Dict, List, Optional, Union from pydantic import BaseModel, Field # JSON-RPC 2.0 基础请求结构 class JSONRPCRequest(BaseModel): jsonrpc: str Field(2.0, constTrue) id: Optional[Union[int, str]] None method: str params: Optional[Dict[str, Any]] None # JSON-RPC 2.0 基础响应/错误结构 class JSONRPCResponse(BaseModel): jsonrpc: str Field(2.0, constTrue) id: Optional[Union[int, str]] None result: Optional[Any] None error: Optional[Dict[str, Any]] None # MCP 工具定义 class Tool(BaseModel): name: str description: str inputSchema: Dict[str, Any] # JSON Schema # MCP 资源定义 class Resource(BaseModel): uri: str name: str description: Optional[str] None mimeType: str text/plain # MCP 初始化参数 class InitializeParams(BaseModel): protocolVersion: str 2024-11-05 clientInfo: Optional[Dict[str, str]] None # MCP 工具调用参数 class CallToolParams(BaseModel): name: str arguments: Optional[Dict[str, Any]] None # MCP 读取资源参数 class ReadResourceParams(BaseModel): uri: str这些模型就像乐高积木的模具确保了我们将要组装和传递的每一个数据块都是形状正确的。3.3 实现Server核心循环处理请求的路由器现在我们来创建Server的主文件server.py。它的核心是一个能根据JSON-RPC请求的method字段将请求路由到对应处理函数的路由器。from fastapi import FastAPI, Request from sse_starlette.sse import EventSourceResponse import asyncio import json from models import * app FastAPI() # 存储当前可用的工具和资源 available_tools: List[Tool] [] available_resources: List[Resource] [] async def handle_jsonrpc_request(request_data: Dict) - Dict: 核心请求处理器 try: req JSONRPCRequest(**request_data) except Exception as e: # 如果连基础请求结构都不对返回解析错误 return JSONRPCResponse( idNone, error{code: -32700, message: Parse error, data: str(e)} ).dict() handler_map { initialize: handle_initialize, tools/list: handle_tools_list, tools/call: handle_tools_call, resources/list: handle_resources_list, resources/read: handle_resources_read, # 可以继续添加其他method的处理函数 } handler handler_map.get(req.method) if not handler: return JSONRPCResponse( idreq.id, error{code: -32601, message: fMethod not found: {req.method}} ).dict() try: result await handler(req.params, req.id) # 处理函数应返回一个符合JSONRPCResponse结构的字典 return result except Exception as e: # 处理函数内部出错 return JSONRPCResponse( idreq.id, error{code: -32603, message: Internal error, data: str(e)} ).dict() # 定义各个请求的处理函数暂时留空 async def handle_initialize(params: Optional[Dict], request_id: Any) - Dict: pass async def handle_tools_list(params: Optional[Dict], request_id: Any) - Dict: pass async def handle_tools_call(params: Optional[Dict], request_id: Any) - Dict: pass async def handle_resources_list(params: Optional[Dict], request_id: Any) - Dict: pass async def handle_resources_read(params: Optional[Dict], request_id: Any) - Dict: pass # HTTP端点用于接收Client的POST请求标准JSON-RPC over HTTP app.post(/jsonrpc) async def jsonrpc_endpoint(request: Request): body await request.json() response await handle_jsonrpc_request(body) return response # SSE端点用于Client建立长连接Server可以主动推送通知如工具列表更新 app.get(/sse) async def sse_endpoint(request: Request): async def event_generator(): # 这里可以维护一个连接队列当有通知时推送给所有连接的Client # 例如当工具列表更新时发送一个 tools/list 通知 # 为了简单我们先返回一个空的事件流 while True: if await request.is_disconnected(): break # 可以在这里检查是否有需要推送的通知 # 暂时只发送一个保持连接的心跳注释 yield {event: comment, data: heartbeat} await asyncio.sleep(30) # 30秒心跳 return EventSourceResponse(event_generator())这个骨架搭建了起来。它有一个HTTP端点/jsonrpc来接收请求一个SSE端点/sse用于服务器主动推送。handle_jsonrpc_request函数是大脑负责解析请求并分发给对应的处理函数。现在我们需要为这些处理函数注入灵魂。4. 注入灵魂实现核心工具与资源处理器骨架已经搭好现在需要实现具体的业务逻辑。我们来实现两个最经典的功能一个计算器工具和一个动态时间资源。4.1 实现初始化与列表查询首先在Server启动时我们需要注册我们的工具和资源。修改server.py的顶部和初始化函数# 在文件顶部定义我们的工具和资源 available_tools [ Tool( namecalculate, description执行简单的数学计算。支持加()、减(-)、乘(*)、除(/)。, inputSchema{ type: object, properties: { expression: { type: string, description: 数学表达式例如3 5 * 2。注意乘号是*除号是/。 } }, required: [expression] } ) ] available_resources [ Resource( uridynamic://server/current_time, name当前服务器时间, description获取服务器当前的日期和时间。, mimeTypetext/plain ) ] async def handle_initialize(params: Optional[Dict], request_id: Any) - Dict: 处理初始化请求 init_params InitializeParams(**(params or {})) # 这里可以检查protocolVersion是否兼容 response JSONRPCResponse( idrequest_id, result{ protocolVersion: init_params.protocolVersion, capabilities: { tools: {listChanged: True}, # 告知Client工具列表可能会变 resources: {listChanged: True} # 告知Client资源列表可能会变 }, serverInfo: { name: My First MCP Server, version: 0.1.0 } } ) return response.dict() async def handle_tools_list(params: Optional[Dict], request_id: Any) - Dict: 返回当前可用的工具列表 response JSONRPCResponse( idrequest_id, result{tools: [tool.dict() for tool in available_tools]} ) return response.dict() async def handle_resources_list(params: Optional[Dict], request_id: Any) - Dict: 返回当前可用的资源列表 response JSONRPCResponse( idrequest_id, result{resources: [resource.dict() for resource in available_resources]} ) return response.dict()初始化处理函数handle_initialize向Client宣告了Server的基本信息和能力。tools/list和resources/list处理函数则直接返回我们预定义好的列表。注意capabilities中的listChanged字段设为True这意味着我们的Server支持在运行时动态更新列表并会通过SSE通道发送通知本例暂未实现动态更新。4.2 实现计算器工具调用这是核心中的核心。当AI发送tools/call请求要求调用calculate工具时我们需要验证参数是否符合schema。安全地执行计算表达式这是重点和难点。返回结果或错误。import ast import operator # 安全评估表达式的辅助函数 def safe_eval_expression(expr: str): 极其有限且安全地评估一个只包含数字和基础运算符的字符串表达式。 警告绝对不要在生产环境中用eval()直接执行用户输入 这里使用ast.literal_eval也有局限我们实现一个简单的解析器作为示例。 # 一个非常简陋的、仅用于演示的解析器按空格分割假设是逆波兰表达式或简单二元运算 # 例如只处理 a op b 形式如 3 5 tokens expr.split() if len(tokens) ! 3: raise ValueError(表达式格式暂只支持 数字 运算符 数字如 3 5) try: a float(tokens[0]) b float(tokens[2]) except ValueError: raise ValueError(操作数必须是数字) op_map { : operator.add, -: operator.sub, *: operator.mul, /: operator.truediv } op_func op_map.get(tokens[1]) if not op_func: raise ValueError(f不支持的运算符: {tokens[1]}。支持: , -, *, /) if tokens[1] / and b 0: raise ZeroDivisionError(除数不能为零) result op_func(a, b) # 如果是整数返回整数形式 if result.is_integer(): return int(result) return result async def handle_tools_call(params: Optional[Dict], request_id: Any) - Dict: 处理工具调用请求 if not params: return JSONRPCResponse( idrequest_id, error{code: -32602, message: Invalid params} ).dict() call_params CallToolParams(**params) if call_params.name calculate: # 1. 获取参数 arguments call_params.arguments or {} expression arguments.get(expression) if not expression: return JSONRPCResponse( idrequest_id, error{code: -32602, message: Missing required argument: expression} ).dict() # 2. 执行计算在真实场景中这里需要更复杂和安全的方法 try: # 注意这里使用自定义的安全函数而非eval result_value safe_eval_expression(expression) except ZeroDivisionError: return JSONRPCResponse( idrequest_id, error{code: -32000, message: Calculation error, data: Division by zero.} ).dict() except Exception as e: return JSONRPCResponse( idrequest_id, error{code: -32000, message: Calculation error, data: str(e)} ).dict() # 3. 返回成功结果 response JSONRPCResponse( idrequest_id, result{ content: [ { type: text, text: f表达式 {expression} 的计算结果是{result_value} } ] } ) return response.dict() # 如果工具名未找到 return JSONRPCResponse( idrequest_id, error{code: -32601, message: fTool not found: {call_params.name}} ).dict()这里有一个至关重要的安全警告在真实的生产环境中绝对禁止使用Python内置的eval()函数来执行用户AI提供的表达式字符串这会带来严重的代码注入安全风险。上面的safe_eval_expression是一个极度简化的示例仅用于演示原理。在实际项目中你需要使用更安全的数学表达式解析库如asteval它利用AST进行有限制评估或者将计算任务委托给一个严格沙箱化的环境。4.3 实现动态时间资源读取资源读取的逻辑相对简单主要是根据请求的URI生成或获取对应的内容。from datetime import datetime async def handle_resources_read(params: Optional[Dict], request_id: Any) - Dict: 处理资源读取请求 if not params: return JSONRPCResponse( idrequest_id, error{code: -32602, message: Invalid params} ).dict() read_params ReadResourceParams(**params) if read_params.uri dynamic://server/current_time: # 动态生成当前时间 current_time datetime.now().strftime(%Y-%m-%d %H:%M:%S) content_text f当前服务器时间是{current_time} response JSONRPCResponse( idrequest_id, result{ contents: [ { uri: read_params.uri, mimeType: text/plain, text: content_text } ] } ) return response.dict() # 如果资源URI未找到 return JSONRPCResponse( idrequest_id, error{code: -32601, message: fResource not found: {read_params.uri}} ).dict()至此一个具备最小功能集的MCP Server就实现了。它能够响应初始化、列表查询、工具调用和资源读取请求。你可以使用uvicorn来运行它uvicorn server:app --reload --port 8000Server将在http://localhost:8000启动。接下来我们需要一个Client来测试它。5. 验证与调试打造一个简易的MCP Client测试器为了验证我们的Server是否正常工作我们不能只依赖Claude Desktop。自己写一个简单的测试Client是理解和调试MCP协议的最佳方式。这个Client会模拟标准MCP Client如Claude的行为向我们的Server发送请求并打印响应。创建test_client.pyimport asyncio import aiohttp import json async def test_mcp_server(): server_url http://localhost:8000 async with aiohttp.ClientSession() as session: print(1. 发送初始化请求...) init_request { jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 2024-11-05, clientInfo: {name: TestClient} } } async with session.post(f{server_url}/jsonrpc, jsoninit_request) as resp: init_result await resp.json() print(f初始化响应: {json.dumps(init_result, indent2, ensure_asciiFalse)}) print(\n2. 查询工具列表...) tools_list_request { jsonrpc: 2.0, id: 2, method: tools/list, params: {} } async with session.post(f{server_url}/jsonrpc, jsontools_list_request) as resp: tools_result await resp.json() print(f工具列表: {json.dumps(tools_result, indent2, ensure_asciiFalse)}) print(\n3. 调用计算器工具...) call_tool_request { jsonrpc: 2.0, id: 3, method: tools/call, params: { name: calculate, arguments: { expression: 10 * 2 3 } } } async with session.post(f{server_url}/jsonrpc, jsoncall_tool_request) as resp: call_result await resp.json() print(f工具调用结果: {json.dumps(call_result, indent2, ensure_asciiFalse)}) print(\n4. 查询资源列表...) resources_list_request { jsonrpc: 2.0, id: 4, method: resources/list, params: {} } async with session.post(f{server_url}/jsonrpc, jsonresources_list_request) as resp: resources_result await resp.json() print(f资源列表: {json.dumps(resources_result, indent2, ensure_asciiFalse)}) print(\n5. 读取时间资源...) read_resource_request { jsonrpc: 2.0, id: 5, method: resources/read, params: { uri: dynamic://server/current_time } } async with session.post(f{server_url}/jsonrpc, jsonread_resource_request) as resp: read_result await resp.json() print(f资源内容: {json.dumps(read_result, indent2, ensure_asciiFalse)}) if __name__ __main__: asyncio.run(test_mcp_server())运行这个测试脚本确保Server已在运行python test_client.py你应该能看到一系列格式规范的JSON响应。如果一切顺利在调用calculate工具时你会收到一个包含计算结果文本的响应在读取时间资源时会得到当前的服务器时间。这个过程让你清晰地看到了MCP协议下“请求-响应”的完整数据流。6. 进阶与生产化思考从玩具到工具我们的“手搓”Server已经能跑通了但它离一个健壮、可用的生产级MCP Server还有巨大差距。这一步的思考才是“工程化”的真正开始。6.1 传输层与部署不止于HTTP我们使用了HTTPSSE这是MCP的一种传输方式transport: sse。但MCP官方还支持标准的stdio标准输入输出这对于与Claude Desktop等本地应用集成更为常见。这意味着你的Server需要能够从sys.stdin读取JSON-RPC请求并将响应写入sys.stdout。你需要修改Server的启动入口根据环境变量如MCP_TRANSPORT来决定使用哪种传输方式。对于生产部署你可能会将Server打包成Docker容器并通过stdio与宿主机上的AI助手通信。6.2 连接管理与状态维护我们的示例Server是无状态的每次请求都是独立的。但在真实场景中会话SessionClient和Server之间可能维持一个会话包含一些上下文信息。资源订阅Resource SubscriptionClient可以订阅某个资源当其内容变化时Server需要通过SSE主动推送更新通知notifications/resources/updated。这要求Server维护资源的状态和Client的订阅列表。工具列表动态更新同样当Server安装或移除了一个工具需要广播notifications/tools/list_changed通知。6.3 安全性、错误处理与日志输入验证与消毒我们强调了计算器工具的安全问题这只是一个缩影。所有来自AI的输入都必须视为不可信的需要进行严格的验证和消毒防止注入攻击。身份验证与授权如果你的Server连接了内部数据库或敏感API那么必须实现身份验证。MCP协议本身不规定认证方式这需要你在传输层如HTTP头添加API Key或应用层自行实现。全面的错误处理我们的示例只有基础错误。一个健壮的Server需要对各种边界情况网络超时、第三方API失败、无效参数组合等定义清晰的错误码和友好的错误信息并通过JSON-RPC的error字段返回。结构化日志为了方便运维和调试所有重要的操作收到请求、调用工具、发生错误都应该被记录并包含请求ID、工具名、耗时等上下文信息。6.4 性能与可观测性异步与并发使用asyncio如我们所用或其它异步框架来处理并发请求避免阻塞。指标Metrics暴露Prometheus格式的指标端点监控请求量、延迟、错误率。跟踪Tracing集成OpenTelemetry追踪一个用户请求从AI发出经过MCP Server调用下游服务再返回的完整链路。7. 整合到真实环境在Claude Desktop中连接你的Server最后让我们把亲手打造的Server用起来。以Claude Desktop为例你需要编辑其配置文件来添加我们的自定义Server。在macOS上配置文件通常位于~/Library/Application Support/Claude/claude_desktop_config.json。在Windows上位于%APPDATA%\Claude\claude_desktop_config.json。你需要添加一个mcpServers配置项。由于我们的Server使用HTTP配置可能如下所示注意Claude Desktop默认更倾向于stdio对SSE的支持可能需要特定版本或配置{ mcpServers: { my-calculator-server: { command: uvicorn, args: [ server:app, --host, 0.0.0.0, --port, 8000 ], env: { PYTHONPATH: /path/to/your/project } // 或者如果你的Server已经作为常驻进程运行可以配置为使用SSE传输 // url: http://localhost:8000/sse, // transport: sse } } }更常见的做法是让Server支持stdio传输。这意味着你需要修改Server使其能够从命令行启动并通过标准输入输出进行通信。这通常涉及解析sys.argv监听sys.stdin的输入并写入sys.stdout。许多MCP SDK如Python的mcp库已经帮你处理了这部分样板代码。配置完成后重启Claude Desktop。在对话中你应该能看到Claude已经识别到了新的工具。你可以尝试对它说“请用我的计算器工具计算一下(15 - 3) * 4 的结果。” 如果一切配置正确Claude会调用你的Server并返回计算结果。这个过程可能会遇到各种问题环境变量路径不对、端口冲突、协议版本不匹配、Claude Desktop缓存了旧的工具列表等等。调试的关键在于查看Claude Desktop的日志通常可以在其设置中找到日志文件路径以及确保你的Server日志是详细且可读的。这就是为什么前面强调日志和错误处理的重要性——当集成出现问题时清晰的日志是你唯一的救生索。从亲手解析第一个JSON-RPC请求到安全地实现一个工具再到最终与AI助手成功联动这个完整的闭环体验就是理解MCP工程化精髓的最佳路径。你不再只是一个API的调用者而是成为了扩展AI能力边界的构建者。下次当你再看到Tavily Search MCP、Brave Search MCP这些复杂的Server时你看到的将不再是一个黑盒而是一个个由类似我们今天搭建的骨架填充了不同业务逻辑后形成的、有生命力的服务。这就是“彻底搞懂”之后世界在你眼中呈现出的不同模样。