说实话,第一次听说 MCP(Model Context Protocol,模型上下文协议)的时候,我脑子里第一反应是“又一个新协议?别折腾了”。但等我真正拿 Python 动手写了一个 MCP Server 之后,我改主意了——这东西确实是 AI 应用从“聊天玩具”走向“生产力工具”的关键一环。它解决的痛点非常实在:你辛辛苦搭好的数据、工具、内部服务,怎么让 AI 模型安全、规范地调起来?不是靠 prompt 里塞一段文字让 AI 去猜,而是给模型一套“可发现、可调用、可校验”的工具接口。这篇博文我就拿一个实战项目——任务管理器,从零开始徒手撸一个 MCP Server,全程用 Python,不用那些一键生成的脚手架,把协议细节、工具注册、参数校验、测试联调这些东西都过一遍。看完你不仅能跑通这套代码,还能理解 MCP 的设计思路,接下来接到自己的业务场景里也知道从哪里下手。
1. MCP 到底是什么,为什么值得自己动手写一次
在动键盘之前,先把基础概念捋清楚。MCP 是一个开放协议,通俗点说,它给“AI 模型”和“外部工具/数据源”之间画了一条标准化的“接口线”。你不必纠结某个模型用的是哪家的 API,也不用为每个场景单独写一套工具适配逻辑,只要实现 MCP 协议,任何支持 MCP 的客户端(比如 Claude Desktop、各种 Agent 框架)都能直接发现并调用你的工具。
1.1 没有 MCP 的时候,工具接入有多痛苦
假设你开发了一个内部的任务管理系统,想让 AI 助手帮你创建任务、查询待办、标记完成。没有 MCP 之前通常的做法有两种:一种是把所有工具调用逻辑硬编码到业务流程里,写一堆 if/else 或者长长的函数列表,让模型通过 function calling 去猜参数,猜错了还要反复重试;另一种是把数据导出成文本喂给模型,让它“看着办”,结果模型经常一本正经地胡编乱造。
这两种方式的问题其实都是同一个:工具的“语义”没有被标准化描述出来。模型看到的只是函数名和参数名,但它不知道这个工具到底是干嘛的、参数有什么约束、该在什么时候调用。MCP 做的事情就是把“工具发现”“参数描述”“结果返回”这几个环节全部规范化,让模型的工具调用过程变得像浏览器访问网页一样自然——先拿到接口清单,再按清单调用,最后把结果解析掉。
1.2 MCP 的核心设计原则
MCP 的架构里有两个核心角色:MCP Server 和 MCP Client。Server 负责把工具能力暴露出来,Client 负责发现并调用这些能力。协议本身基于 JSON-RPC 2.0,所以消息格式非常轻量,传输层可以选择标准输入输出(stdio)或 HTTP + SSE 方式。最关键的三个原语是:
- Tools(工具):给模型提供的可调用函数,通过 JSON Schema 描述参数。
- Resources(资源):给模型暴露的只读数据,相当于可访问的文件或数据片段。
- Prompts(提示词模板):预定义好的提示词,方便复用常见问答模式。
任务管理器这个项目里,最主要用到的是 Tools,因为它的核心需求就是“让模型能操作待办数据”。
1.3 为什么我建议你“徒手”写一遍
你可能想说,网上不是有现成的 MCP SDK 吗?用官方 SDK 几行代码就搭起来了,何必手写?这话对,但也不对。官方 SDK 确实能帮你把底层细节封装好,但如果你不了解协议本身,碰到问题会特别被动——尤其是当你需要在特定网络环境里跑、要自定义认证方式、要调试消息格式的时候,不懂协议细节就无从下手。徒手写一遍最大的价值不是“不用 SDK”,而是让你把 initialize、tools/list、tools/call 这些流程彻底搞清楚,以后再用 SDK 封装也好,直接裸写也罢,心里都有底。
2. 动手前的设计拆解:任务管理器的功能边界
开始写代码之前,先明确我们要做一个什么样的任务管理器。这里我不想做一个“为了演示而演示”的空壳,而是忠实复刻一个简化但完整的待办事项服务:能创建任务、能查询任务、能修改状态、能删除任务。数据存储直接用 JSON 文件,不引数据库,这样方便你理解整个流程,也方便后续扩展。
2.1 功能清单和工具定义
任务管理器的核心数据模型很简单,一个任务包含以下字段:
- id:任务唯一标识,用 UUID 字符串
- title:任务标题,必填,最长为 200 字
- description:任务详细描述,可选
- status:状态,取值为 pending / done
- priority:优先级,取值为 low / mid / high,默认 mid
- created_at:创建时间,时间戳字符串
- updated_at:最后更新时间
基于这个模型,我设计了 5 个工具:
| 工具名 | 功能说明 | 关键参数 |
|---|---|---|
| add_task | 创建新任务 | title(必填)、description(可选)、priority(可选) |
| list_tasks | 查询任务列表 | status(可选)、keyword(可选) |
| complete_task | 将任务标记为完成 | task_id(必填) |
| delete_task | 按 id 删除任务 | task_id(必填) |
| get_task_detail | 查询单个任务详情 | task_id(必填) |
为什么选这 5 个?因为它们覆盖了一个待办服务最常见的操作类型:增、查、改、删、单查。通过这 5 个工具,你可以完整跑通 MCP 的“工具发现 → 参数校验 → 逻辑执行 → 结果返回”链路,不会因为功能太多而冲淡对协议本身的理解。
2.2 存储设计:为什么用 JSON 文件而不是 SQLite
很多人在这一步会纠结:任务管理器按理说用数据库更正规,SQLite 也很轻量,为什么要选 JSON 文件?
我的考虑有几点。第一,这个项目的核心目标是演示 MCP 服务器本身的实现,不是演示持久化方案,存储层越简单越利于聚焦。第二,JSON 文件的好处是“所见即所得”,你可以随时打开 data.json 检查数据写入是否正确,调试起来非常直观。第三,后续如果想换成 SQLite 或 PostgreSQL,你只需要替换存储层那几个读写函数,MCP 工具层完全不用动。所以这里用 JSON 是一个刻意简化,但绝不是偷懒,而是合理取舍。
2.3 传输方式选型:stdio vs HTTP
MCP 支持两种主流传输方式。stdio 模式适用于本地场景:客户端启动服务器子进程,两者通过标准输入输出通信。HTTP + SSE 模式适用于远程部署场景:客户端通过网络请求发起连接。任务管理器我默认用 stdio 模式,因为本地调试最省事,Claude Desktop 这类客户端也支持直接配置 stdio 服务。当然,后面的代码里我也会提一句怎么扩展成 HTTP 模式,让你知道区别在哪。
3. 核心协议流程拆解:initialize、tools/list、tools/call
这一节是整个项目的灵魂。很多人看 MCP 文档觉得云里雾里,就是因为没搞清楚消息的流转过程。实际上 MCP 的交互非常像一个面试流程:客户端先问“你支持什么功能”,服务器列出来,客户端再按需调用。
3.1 initialize:握手阶段
客户端和服务器建立连接后,第一件事是发送 initialize 请求。这个请求里会包含协议版本号、客户端能力描述等信息。服务器收到后需要返回自己的协议版本、服务器能力列表、服务器的名称和版本。握手成功后,双方才会进入正常的资源与工具交互。如果你在日志里看到 initialize failed,多半是版本号不匹配,或者返回的报文格式不符合 JSON-RPC 规范。
3.2 tools/list:工具发现阶段
握手完成后,客户端会主动发送 tools/list 请求,询问服务器“你提供了哪些工具”。这个请求不携带任何参数,服务器需要返回一个工具列表,每个工具都包含 name、description、inputSchema 三部分。这里的 inputSchema 是 JSON Schema,它对每个参数做了详细的类型约束和语义说明。不要小看这块描述,模型的调用质量很大程度上取决于你把参数说明写得够不够清楚。
3.3 tools/call:工具调用阶段
客户端根据工具列表选择合适的工具,发送 tools/call 请求。请求中会携带 tool_name 和 arguments 两个字段。服务器执行对应逻辑后,需要把结果包装成 MCP 规定的格式返回——尤其要注意 content 数组的结构。常见的坑是很多人直接把字符串丢进 content,没包装成带 type 的对象,结果客户端解析时报错。返回结果建议包含一个 isError 字段,让客户端可以区分正常结果和业务错误。
3.4 消息格式细节示例
一个典型的 tools/call 请求长这样:
{ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "add_task", "arguments": { "title": "写一篇 MCP 实战教程", "priority": "high" } } }对应的返回报文长这样:
{ "jsonrpc": "2.0", "id": 1, "result": { "content": [ { "type": "text", "text": "{\"task_id\": \"123e4567-e89b-12d3-a456-426614174000\"}" } ], "isError": false } }这些格式如果你手写很容易出错,所以这也是我建议用 SDK 辅助的原因之一。不过理解每个字段的含义非常有必要,不管你是手写还是用 SDK,出了问题都能快速定位。
4. 逐步手写 MCP Server:代码实战
好了,到了整个项目最核心的部分。我先把实现拆成几个层次:协议层、业务逻辑层、入口层。这样代码结构清晰,调试也方便。下面的代码我会用尽量少的第三方依赖,让你真正理解核心机制。工具库只用标准库和官方 mcp 包(用于消息语法校验),核心 IO 逻辑我用类来封装。
4.1 项目目录结构
建议你按下面这个结构组织代码:
mcp-task-manager/ ├── server.py # MCP 服务器主入口 ├── storage.py # JSON 文件存储层 ├── tools.py # 工具定义与业务逻辑 ├── data/ │ └── tasks.json # 任务数据文件 ├── requirements.txt # 依赖列表 └── client_test.py # 本地调试客户端4.2 存储层:JSON 文件读写
storage.py 的核心是四个操作:读所有任务、写所有任务、按 id 获取任务、持久化。我写了一个 TaskStore 类来管理这些操作。
import json import os import threading import uuid from datetime import datetime DATA_DIR = os.path.join(os.path.dirname(__file__), "data") DATA_FILE = os.path.join(DATA_DIR, "tasks.json") class TaskStore: def __init__(self, data_file=DATA_FILE): self.data_file = data_file self._lock = threading.Lock() self._ensure_file() def _ensure_file(self): if not os.path.exists(DATA_DIR): os.makedirs(DATA_DIR) if not os.path.exists(self.data_file): self._write([]) def _read(self): with open(self.data_file, encoding="utf-8") as f: return json.load(f) def _write(self, tasks): with open(self.data_file, "w", encoding="utf-8") as f: json.dump(tasks, f, ensure_ascii=False, indent=2) def add_task(self, title, description="", priority="mid"): task = { "id": str(uuid.uuid4()), "title": title, "description": description, "priority": priority, "status": "pending", "created_at": datetime.now().isoformat(), "updated_at": datetime.now().isoformat() } with self._lock: tasks = self._read() tasks.append(task) self._write(tasks) return task def list_tasks(self, status=None, keyword=None): with self._lock: tasks = self._read() if status: tasks = [t for t in tasks if t["status"] == status] if keyword: tasks = [t for t in tasks if keyword in t["title"]] return tasks def get_task(self, task_id): with self._lock: tasks = self._read() for t in tasks: if t["id"] == task_id: return t return None def update_status(self, task_id, status): with self._lock: tasks = self._read() for t in tasks: if t["id"] == task_id: t["status"] = status t["updated_at"] = datetime.now().isoformat() self._write(tasks) return t return None def delete_task(self, task_id): with self._lock: tasks = self._read() new_tasks = [t for t in tasks if t["id"] != task_id] if len(new_tasks) == len(tasks): return False self._write(new_tasks) return True这段代码几个地方值得说明一下。加锁是因为 JSON 文件读写不是原子操作,如果未来你把这个服务部署成多线程模式,多个请求同时写文件就容易出现数据丢失。UUID 作为主键而不是自增数字,好处是分布式的环境下不会出现主键冲突。时间戳用 isoformat 输出字符串,方便后续前端直接展示。
4.3 工具定义层:把业务逻辑变成 MCP 可识别的工具
tools.py 里我会定义两个核心的东西:一个工具注册表,一个分发的函数。工具注册表描述了每个工具的元信息,分发函数根据客户端传过来的工具名调用对应逻辑。
先定义工具的 Schema 和注册表:
TOOLS = [ { "name": "add_task", "description": "创建一个新的待办任务。返回新任务的完整信息,包括任务ID。", "inputSchema": { "type": "object", "properties": { "title": { "type": "string", "description": "任务标题,例如「完成季度汇报PPT」" }, "description": { "type": "string", "description": "任务详细说明,可选" }, "priority": { "type": "string", "enum": ["low", "mid", "high"], "description": "优先级,默认 mid" } }, "required": ["title"] } }, { "name": "list_tasks", "description": "查询任务列表。可按状态和关键字过滤。", "inputSchema": { "type": "object", "properties": { "status": { "type": "string", "enum": ["pending", "done"], "description": "按状态过滤,可选" }, "keyword": { "type": "string", "description": "按标题关键字搜索,可选" } } } }, { "name": "complete_task", "description": "将指定任务标记为已完成。需要提供任务ID。", "inputSchema": { "type": "object", "properties": { "task_id": { "type": "string", "description": "要标记完成的任务ID" } }, "required": ["task_id"] } }, { "name": "delete_task", "description": "按任务ID删除一个任务。", "inputSchema": { "type": "object", "properties": { "task_id": { "type": "string", "description": "要删除的任务ID" } }, "required": ["task_id"] } }, { "name": "get_task_detail", "description": "获取指定任务的完整详情。", "inputSchema": { "type": "object", "properties": { "task_id": { "type": "string", "description": "任务ID" } }, "required": ["task_id"] } }, ]注意 inputSchema 的写法。每个参数类型要明确,有枚举的记得写 enum,这样模型在调用时就能自动规避非法参数。比如 priority 如果不加 enum,模型可能自由发挥,传一个 “urgent” 进来,你的服务器还得做容错;加了 enum 后,大部分符合规范的客户端会直接限制可选项。
接着是分发函数:
from storage import TaskStore store = TaskStore() def handle_tool_call(name, arguments): if name == "add_task": title = arguments.get("title") description = arguments.get("description", "") priority = arguments.get("priority", "mid") if not title or not title.strip(): return {"isError": True, "content": [{"type": "text", "text": "参数错误:title 不能为空"}]} task = store.add_task(title.strip(), description, priority) return {"isError": False, "content": [{"type": "text", "text": __json_dumps(task)}]} elif name == "list_tasks": tasks = store.list_tasks(arguments.get("status"), arguments.get("keyword")) return {"isError": False, "content": [{"type": "text", "text": __json_dumps(tasks)}]} elif name == "complete_task": task_id = arguments.get("task_id", "") task = store.update_status(task_id, "done") if task: return {"isError": False, "content": [{"type": "text", "text": __json_dumps(task)}]} return {"isError": True, "content": [{"type": "text", "text": f"任务不存在:{task_id}"}]} elif name == "delete_task": ok = store.delete_task(arguments.get("task_id", "")) if ok: return {"isError": False, "content": [{"type": "text", "text": "删除成功"}]} return {"isError": True, "content": [{"type": "text", "text": f"任务不存在:{arguments.get('task_id', '')}"}]} elif name == "get_task_detail": task = store.get_task(arguments.get("task_id", "")) if task: return {"isError": False, "content": [{"type": "text", "text": __json_dumps(task)}]} return {"isError": True, "content": [{"type": "text", "text": f"任务不存在:{arguments.get('task_id', '')}"}]} else: return {"isError": True, "content": [{"type": "text", "text": f"未知工具:{name}"}]} def __json_dumps(obj): return json.dumps(obj, ensure_ascii=False, indent=2)4.4 协议层:用标准库实现 MCP 消息循环
现在到了最关键的协议层。我要实现一个简单的 MCP 协议循环,从标准输入读取 JSON 消息,解析请求,分发处理,返回响应。这里我选择用官方 mcp 的底层方法来做消息解析与校验,但 IO 流程自行控制。
一个标准的 stdio MCP Server 的启动流程是这样的:读取本机的 Python 环境,作为子进程被客户端拉起,然后循环读 stdin。每一行是一个 JSON-RPC 消息。我们处理完消息后,把响应写到 stdout。注意,日志信息不要往 stdout 打,否则会污染协议通道,调试日志请写到 stderr。
import sys import json import logging logging.basicConfig(stream=sys.stderr, level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") def handle_message(message): method = message.get("method") msg_id = message.get("id") if method == "initialize": return { "jsonrpc": "2.0", "id": msg_id, "result": { "protocolVersion": "2024-11-05", "capabilities": { "tools": {} }, "serverInfo": { "name": "task-manager-mcp-server", "version": "0.1.0" } } } elif method == "notifications/initialized": return None elif method == "tools/list": return { "jsonrpc": "2.0", "id": msg_id, "result": { "tools": TOOLS } } elif method == "tools/call": params = message.get("params", {}) name = params.get("name") arguments = params.get("arguments", {}) result = handle_tool_call(name, arguments) return { "jsonrpc": "2.0", "id": msg_id, "result": result } else: logging.warning(f"未知方法: {method}") return None def main(): logging.info("Task Manager MCP Server started") for line in sys.stdin: line = line.strip() if not line: continue try: message = json.loads(line) response = handle_message(message) if response is not None: sys.stdout.write(json.dumps(response, ensure_ascii=False) + "\n") sys.stdout.flush() except json.JSONDecodeError as e: logging.error(f"JSON 解析失败: {e}") if __name__ == "__main__": main()如果你手边有官方 mcp 库,并且不想自己处理协议细节,也可以用 SDK 极简实现,两三行就搞定。但作为“徒手撸”项目,我建议至少跑通上面这个版本,再去看 SDK 的封装。
用官方 SDK 的版本会长这样:
from mcp.server.fastmcp import FastMCP mcp = FastMCP("task-manager-mcp-server") @mcp.tool() def add_task(title: str, description: str = "", priority: str = "mid") -> str: """创建一个新的待办任务""" task = store.add_task(title, description, priority) return json.dumps(task, ensure_ascii=False) if __name__ == "__main__": mcp.run()两个版本的区别很明显:SDK 版本工具函数就是普通 Python 函数,插件帮你完成参数校验和消息循环;手写版本则是自己解析所有消息。我建议你两个都写一遍,手写版本让你懂原理,SDK 版本让你省时间。生产环境建议用 SDK,因为你不需要重新发明轮子。
5. 本地调试与真实客户端联调
服务器写完了,接下来是最容易卡住新手的环节:怎么知道我的服务器确实能被 MCP 客户端正常调用?这一步我分开讲,先讲最快速的命令行验证方式,再讲怎么接 Claude Desktop 这类真实客户端。
5.1 用命令行模拟客户端
最直接的测试是写一个最简单的客户端,通过 subprocess 启动服务器,然后向 stdin 发 JSON-RPC 消息,再从 stdout 读响应。这里我写了一个 client_test.py,方便你验证流程。
import subprocess import json import time proc = subprocess.Popen( ["python", "server.py"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, encoding="utf-8", ) def send(message): proc.stdin.write(json.dumps(message, ensure_ascii=False) + "\n") proc.stdin.flush() return json.loads(proc.stdout.readline()) # 1. initialize 握手 resp = send({ "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "test-client", "version": "1.0.0"} } }) print("初始化响应:", resp) # 2. tools/list 工具发现 resp = send({ "jsonrpc": "2.0", "id": 2, "method": "tools/list" }) print("工具列表:", json.dumps(resp, ensure_ascii=False, indent=2)) # 3. tools/call 调用 add_task resp = send({ "jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": { "name": "add_task", "arguments": { "title": "测试任务", "priority": "high" } } }) print("添加任务结果:", resp) # 4. 结束进程 proc.terminate()运行这个脚本后,你会在命令行看到完整的握手、工具发现、工具调用过程。如果一切正常,data/tasks.json 里会出现一条新任务记录。
5.2 用 MCP Inspector 可视化调试
如果你不想自己写客户端,MCP 官方提供了一个调试面板叫 MCP Inspector。你可以通过 npx 启动它:
npx @modelcontextprotocol/inspector python server.py启动后浏览器会自动打开一个调试页面,你可以在里面看到工具列表、手动触发工具调用、查看请求响应日志。用这个工具的好处是你能直观看到工具调用过程中的 JSON 报文,对排查消息格式问题特别有用。我第一次写手写协议版本时,就是靠 Inspector 发现自己在 tools/list 返回里漏了 toolSchema 这个字段,客户端一直报解析错误。
5.3 接入 Claude Desktop 或其他 MCP 客户端
如果是 Claude Desktop,配置文件一般位于 claude_desktop_config.json,你只需要添加一个 mcpServers 配置项:
{ "mcpServers": { "task-manager": { "command": "python", "args": ["/path/to/your/mcp-task-manager/server.py"] } } }配置完后重启客户端,Chat 界面里就能看到任务管理器提供的工具了。你可以直接对 AI 说“帮我创建一个优先级为高的任务:完成季度汇报”,模型就会自己调用 add_task 工具,并把返回的任务 ID 呈现在回答里。
6. 常见问题与排查技巧实录
整个项目从零写下来,我踩了不少坑,也看到群里不少人卡在同一类问题上。这些是真实项目中容易遇到的高频问题,我整理成了一份速查表,并且把排查思路也附上。
| 常见问题 | 可能原因 | 排查与解决办法 |
|---|---|---|
| 客户端提示 initialize 失败 | 协议版本号不匹配,或返回的 capabilities 格式不对 | 检查服务器返回的 protocolVersion 是否在客户端支持范围内,推荐返回 2024-11-05 或 2024-10-07 |
| 工具列表能加载,但调用时无响应 | 消息循环没有正确 flush stdout | 每写一条响应后务必调用 sys.stdout.flush(),否则数据滞留在缓冲区 |
| 服务器 stderr 有报错,但客户端无感知 | 工具函数内部抛了未捕获异常 | 在 handle_tool_call 外层加 try/except,把异常包装成 isError 返回,不要直接让进程崩溃 |
| 添加任务后,JSON 文件是空的 | 工作目录不对,data 目录被创建到了错误路径 | 不要用相对路径,把 DATA_FILE 改为基于__file__的绝对路径 |
| 模型调用时,枚举参数传了非法值 | Schema 里没写 enum | 在 inputSchema 中为有固定取值的字段添加 enum 约束 |
| 并发请求导致 JSON 文件损坏 | 多线程同时读写文件 | 给存储层加 threading.Lock,保证写操作串行执行 |
| 日志刷屏污染 stdout | 把 logging.StreamHandler 指向了 stdout | 日志输出到 stderr,协议消息走 stdout |
6.1 各种奇怪报错的通用排查路径
当你遇到一个说不清来源的报错时,不要瞎猜,按下面的顺序来做:
先看 stderr 日志。MCP 服务器的调试日志都走 stderr,所以第一步是把客户端的错误输出完整捞出来,很多问题在这一步就能定位。再看协议报文。用 MCP Inspector 复现问题,查看客户端实际发出的 JSON-RPC 消息,以及服务器返回的响应。重点检查 jsonrpc 字段、id 字段是否配对、method 拼写是否正确。再看数据文件。如果工具逻辑没问题,检查 tasks.json 里的数据是否符合预期,判断是读的问题还是写的问题。最后再考虑升级或降级 SDK 版本。如果手写版本和官方 SDK 版本同时遇到诡异问题,检查 mcp 库版本与客户端版本兼容性。
6.2 几个容易忽略的经验细节
第一个经验是,在工具定义里,description 一定要用完整句子写清楚用途,不要只写一句话。比如 add_task 的描述,我写的是“创建一个新的待办任务。返回新任务的完整信息,包括任务ID。”而不是简单写“新增任务”。你以为 AI 模型能靠猜理解你的意思,实际上一个好的 description 能显著降低模型误调用的概率。
第二个经验是,参数校验不要完全依赖模型遵守 Schema,服务器端也要做一层防御性校验。原因很简单,你无法保证所有调用方都严格实现了 MCP 协议的参数校验规范,而且未来可能有未知客户端直接以命令行方式调用你的工具逻辑。所以在 handle_tool_call 里对 title 空值、task_id 不存在等情况做了显式判断,返回语义化错误信息,而不是抛异常。
第三个经验是,优先用 Python 3.10+ 配合类型注解写工具函数。MCP 官方 SDK 会利用类型注解自动生成 JSON Schema,包含类型说明;你手写 Schema 的时候,也建议在业务函数上用类型注解,方便未来自动生成文档和测试用例。
7. 如何扩展成一个真正能上线的 MCP 服务
现在你已经有了一个完整可运行的 MCP Server,但离生产级应用还有几步路要走。我在实现任务管理器过程中,思考过这些扩展方向,分享给你。
7.1 从 stdio 扩展到 HTTP + SSE
如果你要部署到远程服务器,供浏览器端或远程客户端调用,就不能用 stdio 了。MCP 官方支持 HTTP + SSE 传输模式。原理很简单:客户端先通过 HTTP POST 建立 SSE 流,之后消息通过这个流双向传递。Python 里可以用 FastAPI 包装一层,把上文的消息处理逻辑暴露成一个 POST 接口,同时用 SSE 推送服务器返回的结果。
伪代码如下:
from fastapi import FastAPI from fastapi.responses import StreamingResponse import json app = FastAPI() @app.post("/mcp") async def mcp_endpoint(request: Request): body = await request.json() response = handle_message(body) return StreamingResponse( iter([json.dumps(response, ensure_ascii=False)]), media_type="text/event-stream" )实际部署时,你还需要考虑鉴权、限流、HTTPS 等事情。但最核心的消息处理逻辑是可以完全复用的。
7.2 接入数据库
JSON 文件存储只适合轻量级单机场景。如果要支持多人协作、任务量大、需要事务,建议换成 SQLite 或 PostgreSQL。关键点是保留 Storage 层的接口签名不变,只修改实现即可。MCP 工具层完全感知不到底层存储变化,这也是模块化设计带来的好处。
7.3 增加 Resources 和 Prompts
任务管理器目前只演示了 Tools,但 MCP 还支持 Resources 和 Prompts。如果你想让 AI 直接读取某个任务的完整内容来生成周报,可以把它暴露为 Resource。如果你想给用户提供一些预设的提示词,比如“帮我写一份任务周报模板”,可以做成 Prompt。这些能力可以有效扩展 AI 与数据的交互方式,值得深入研究。
7.4 多工具服务器的组织方式
真实项目里,你大概率不会只有一个任务管理器,可能还有用户系统、订单系统、报表系统。这时候有两种组织方式:一种是把所有工具挂到同一个 MCP Server 上,适合内部工具数量少、耦合度高的情况;另一种是每个业务域一个 Server,通过 MCP 客户端同时挂载多个服务器,适合大型团队各业务独立迭代的场景。任务管理器你完全可以按第二种方式独立部署,因为它与业务域的解耦做得很好。
写在最后的几个建议
这个项目的完整跑通,让我对 MCP 的理解从“又一个 json-rpc 协议”变成了“一套标准的 AI 工具接入范式”。从实际操作的体会来看,我觉得有三点值得再强调一下。第一,别急着上框架,先徒手实现一遍协议流程,你才能真正理解 MCP 的边界在哪里,不会被各种封装搞得一头雾水。第二,工具参数的描述质量直接影响 AI 调用成功率,这是个值得反复打磨的细节,你投入的每一分钟都会有回报。第三,任何工具服务上线之前,都要想清楚鉴权方案,MCP 本身只是调用规范,安全能力需要你自己构建。
如果你照着这篇博文跑通了代码,你可以再试着扩展一下:做一个人机交互页面来管理这些任务,或者把任务数据接入你的日历系统,又或者给这个服务器写一个定时触发工具。MCP 的生态还在快速演进,现在动手,可能正好踩在趋势前面。