开发Agent的时间越长我越发现大部分项目的瓶颈根本不在模型能力而在“工具接入”这件事上。每个Agent都自带一套调用方式有走函数调用的有自己定义JSON协议的还有直接拼系统提示词的项目一多就开始失控。直到我把所有能力统一收敛到MCP-Server上整个结构才真正稳定下来。这篇文章是Agent系列8.4篇只讲一件事如何从零开发一个生产可用的MCP-Server包括工具设计、协议细节、Agent接入链路以及我在实操中踩过的坑。适合那些已经把Agent框架玩明白、正在做工具层标准化的同学也适合刚接触MCP、想看一个完整落地过程的人。1. MCP到底是什么一个被脚手架掩盖的架构问题很多人第一次接触MCP会把它理解成“又一个函数调用库”换了一套装饰器而已。真这么想后面肯定会踩坑。MCPModel Context Protocol解决的根本不是“少写几行代码”而是工具能力的属主边界问题。1.1 没有MCP之前Agent接工具是这样的在没有MCP之前我自己项目里Agent接工具无非三种形态直接把函数注册进Agent框架。快但是工具代码和Agent逻辑揉在一起换一个框架就得改一遍。自己定义HTTP接口Agent用function calling去请求。可行但每个接口都要自己设计协议、处理鉴权、写错误码、做限流工作量大且零散。全部塞进系统提示词。Agent确实能“读”到信息但无法可靠地执行操作也不是真正的接入。这三种方式最大的问题在于工具的属主是模糊的。工具到底算Agent的一部分还是算平台的一部分API地址变化了、工具逻辑升级了、换了一个Agent框架改动都会蔓延到各个角落。MCP把这件事拆成了清晰的两端。Agent侧是MCP Host/Client只管通过协议发请求、收结果工具侧是MCP-Server负责把真实能力包装成标准化的资源、工具和提示模板。两边不直接依赖对方。我后来在团队里推行MCP时经常说一句话工具由平台统一治理Agent只消费协议谁也不要碰谁的内部结构。1.2 MCP的三类原语和一次完整调用MCP抽象了三类原语理解这三样东西整个协议就懂了一半Tools可执行的操作。Agent调用后会产生副作用比如创建任务、发消息、修改数据。这是开发中最常用的。Resources只读数据。类似文件、数据库查询结果、API响应。Agent需要读取知识或上下文时用。Prompts可复用的提示模板。可以把一些复杂的任务流程封装起来让Agent按固定结构去执行。一次完整调用走的链路是这样的客户端先发送initialize握手协商协议版本然后通过tools/list拉取服务端能力清单Agent决定调用哪个工具后发送tools/call请求并带上参数服务端执行真实逻辑返回执行结果。消息格式用的是JSON-RPC 2.0传输层既支持本进程内的stdio也支持基于HTTP的streamable HTTP。这套设计看起来很学术实际价值却很通俗它把Agent和工具之间的一切交互收敛成了一套有版本、可发现、可校验的标准协议。你不需要关心对方的工具是用Python写的还是Node写的也不需要关心它是本地进程还是远程服务。1.3 本实战的技术选型为什么用官方Python SDK这个实战项目我选的是官方Python SDK具体来说是FastMCP封装层。原因不复杂我手头所有Agent框架都是Python生态统一语言能少维护一层技术栈。FastMCP用装饰器暴露工具和Agent框架里的function calling结构天然对齐学习成本低。官方SDK维护及时新协议能力出来后会尽快跟进。传输层方面本机联调时用stdio模式服务端作为子进程和Agent通信简单可靠部署到远程环境时切到streamable-http模式通过HTTP暴露能力。实际开发中我建议先跑通stdio再上HTTP排查问题会容易很多。2. 开发环境与工程骨架搭建MCP-Server本身不复杂但工程骨架要提前搭好否则业务一多很快就会乱。2.1 依赖安装与最小工程结构先用官方SDK搭建基础环境pip install mcp[cli][cli]和纯mcp包的区别在于多带了一个mcp dev调试命令后面会用到。工程结构上我不喜欢把所有工具堆在一个文件里推荐的最小结构是这样的mcp-server/ ├── pyproject.toml ├── src/ │ └── project_assistant/ │ ├── __init__.py │ ├── server.py # FastMCP实例创建与启动 │ ├── tools/ │ │ ├── __init__.py # 工具注册入口 │ │ ├── tasks.py # 业务工具 │ │ └── memory.py # 记忆与状态相关工具 │ └── storage/ │ └── file_store.py # 持久化存储server.py只做一件事创建FastMCP实例并启动。业务逻辑按领域拆到单独的模块里每个模块负责一组相关工具。我见过很多项目把所有工具塞在一个文件里到后面单个文件上千行改一个函数都要翻半天这就是架构问题提前生了。2.2 第一个工具从hello开始先写一个最简单的Server感受一下协议是怎么跑通的from mcp.server.fastmcp import FastMCP mcp FastMCP(project-assistant) mcp.tool() def get_server_info() - dict: 获取服务器基础信息包括当前时间与状态。 from datetime import datetime, timezone return { server: project-assistant, status: ready, time: datetime.now(timezone.utc).isoformat(), } if __name__ __main__: mcp.run(transportstdio)启动调试是这个流程mcp dev server.pymcp dev会启动一个MCP Inspector的本地调试面板默认走http://localhost:6274你在面板里可以直接看到tools/list返回了哪些工具也能手动发起tools/call并查看结果。这个调试面板被很多人忽略了实际上它是排查问题的第一步。2.3 动态注册工具从静态清单到目录扫描真实项目里工具清单不可能是写死的。我曾经遇到一个需求不同项目组接入同一套MCP-Server但每个组只希望暴露自己相关的工具。于是做成动态注册就很有必要。FastMCP支持通过list_tools回调函数动态返回工具快照。我实现过一个简单的配置驱动方案import json from pathlib import Path def load_tool_configs() - list[dict]: config_dir Path(config/tools) configs [] for file in config_dir.glob(*.json): with open(file, encodingutf-8) as f: configs.append(json.load(f)) return configs mcp.list_tools() def list_tools() - list: configs load_tool_configs() snapshot [] for cfg in configs: snapshot.append({ name: cfg[name], description: cfg[description], inputSchema: cfg[inputSchema], }) return snapshot这样每个工具的开关、描述、参数Schema都从配置中心下发不改代码就能调整Agent可见的能力。但要提醒一点动态注册不代表没有约束工具名必须全局唯一每个工具的description必须认真写因为LLM是靠它做调用决策的。3. 实现一个真实可用的MCP Server以“项目进度助手”为例为了让过程有体感这一节用一个完整的业务场景带大家走一遍实现一个项目进度助手。Agent可以通过它创建任务、查询任务、更新状态并在最后生成进度摘要。3.1 需求拆解与工具设计先想清楚Agent到底需要什么能力不要一上来就写。我给这个项目拆了四类工具工具名作用关键参数create_task创建新任务title, priority, assignee, due_datelist_tasks按状态筛选查询任务列表status, page_size, pageupdate_task更新任务状态或字段task_id, status, title, assigneeget_progress_summary生成项目进度摘要project_id工具不是越多越好而是每个都要回答“Agent在什么场景下会用到它”。比如get_progress_summary就是专门给“生成周报”这种高层级Agent用的它把多个任务状态汇总成一段结构化文本省得Agent自己拉一批任务再慢慢算。3.2 核心工具实现存储我用JSON文件而不是真正的数据库目的是让示例自包含拿到代码就能跑。但持久化有几个细节要注意直接上代码import json import uuid import threading from datetime import datetime, timezone from pathlib import Path from mcp.server.fastmcp import FastMCP mcp FastMCP(project-assistant) DATA_FILE Path(data/tasks.json) LOCK threading.Lock() def _read_tasks() - list[dict]: if not DATA_FILE.exists(): return [] with open(DATA_FILE, encodingutf-8) as f: return json.load(f) def _write_tasks(tasks: list[dict]) - None: DATA_FILE.parent.mkdir(parentsTrue, exist_okTrue) tmp_file DATA_FILE.with_suffix(.json.tmp) with open(tmp_file, w, encodingutf-8) as f: json.dump(tasks, f, ensure_asciiFalse, indent2) tmp_file.replace(DATA_FILE)两个细节跨进程锁。MCP-Server可能在并发请求下运行同一时刻多个Agent都在调create_task所以对文件操作加锁原子写。先写临时文件再rename替代原文件避免程序中途崩溃导致数据文件损坏。别小看这两步我见过直接用open(..., w)覆盖写的工具一次并发请求就把整个任务数据清空了。接下来是具体的工具实现mcp.tool() def create_task(title: str, priority: str medium, assignee: str | None None, due_date: str | None None) - dict: 创建一条新任务。 Args: title: 任务标题必须简洁明确。 priority: 优先级可选 low/medium/high。 assignee: 负责人可空。 due_date: 截止日期ISO格式如 2025-06-30。 if priority not in (low, medium, high): raise ValueError(priority 只能是 low/medium/high) task { id: task_ uuid.uuid4().hex[:12], title: title.strip(), priority: priority, status: todo, assignee: assignee, due_date: due_date, created_at: datetime.now(timezone.utc).isoformat(), updated_at: datetime.now(timezone.utc).isoformat(), } with LOCK: tasks _read_tasks() tasks.append(task) _write_tasks(tasks) return {ok: True, task_id: task[id]}这里参数校验没省。LLM并不是每次都传合法参数发布于一个不存在的优先级值的情况我遇到过不少直接报错会打断整个Agent推理链路所以在工具入口就把参数卡死返回明确的错误信息。再写查询和更新的逻辑mcp.tool() def list_tasks(status: str | None None, page_size: int 20, page: int 1) - dict: 按条件查询任务列表。 Args: status: 状态过滤可选 todo/in_progress/done不传则返回全部。 page_size: 分页大小默认20。 page: 页码从1开始。 tasks _read_tasks() if status: tasks [t for t in tasks if t[status] status] tasks.sort(keylambda x: x[updated_at], reverseTrue) total len(tasks) start (page - 1) * page_size items tasks[start:start page_size] return { total: total, page: page, page_size: page_size, items: items, } mcp.tool() def update_task(task_id: str, status: str | None None, title: str | None None, assignee: str | None None) - dict: 更新任务部分字段未提供的字段保持不变。 allowed_status {todo, in_progress, done} if status and status not in allowed_status: raise ValueError(fstatus 必须是 {allowed_status} 之一) with LOCK: tasks _read_tasks() target next((t for t in tasks if t[id] task_id), None) if target is None: return {ok: False, error: f任务不存在: {task_id}} if status: target[status] status if title: target[title] title.strip() ...update_task故意设计成“只更新传入字段其他保持不变”。这个细节很多人不会注意但对Agent很重要——它通常只知道要改什么不知道要保留什么。如果是全量覆盖语义Agent必须把任务所有字段拿回来再提交多一个调用环节出错的概率就大了一圈。最后是进度摘要工具。这个工具真正的价值在于把多个任务的低层信息汇总成高层的项目视图mcp.tool() def get_progress_summary(project_id: str) - dict: 生成项目进度摘要统计各状态任务数量并计算完成率。 tasks _read_tasks() stats {todo: 0, in_progress: 0, done: 0} for t in tasks: if t[status] in stats: stats[t[status]] 1 total len(tasks) done stats[done] rate round(done / total * 100, 1) if total else 0.0 return { project_id: project_id, total: total, stats: stats, completion_rate: rate, summary: f项目共{total}个任务已完成{done}个完成率{rate}%。 }这类“摘要工具”是Agent开发里很容易被遗漏的一块。很多开发者只提供原子操作增删改查然后指望Agent自己汇总结果就是Agent要循环调用很多次才能得出结论既慢又费token。实际上在服务端把常用的聚合逻辑做成工具反而让整个链路更合理。3.3 数据持久化与记忆雏形MCP协议本身是无状态的每次tools/call都是独立的请求这是很多人一开始会困惑的地方。但真实Agent需要状态所以记忆由Server端自己管理。这里我把记忆分成两层会话内短期状态保存在Server进程内存里比如“最近一次查询的任务列表”。Agent在后续对话里提到“刚才那些任务”就会用到这层缓存。跨会话持久状态写进JSON文件或数据库比如任务数据本身。重启Server也能恢复。如果你要做更完整的Agent记忆体系可以在Server内部接一个向量库做长期语义检索再把检索方法封装成一个工具。但记住一条原则MCP-Server是工具不是记忆库。工具做到“能力语义完整”记忆由上层Agent或专用模块去编排千万别把状态逻辑堆进工具里。3.4 工具描述是Agent的“使用说明书”这里是所有实战里最容易被轻视的一点工具描述写得好不好直接决定Agent调用得对不对。同样的一个list_tasks这样写查询任务列表。和这样写按条件查询任务列表支持按状态过滤todo/in_progress/done返回结果按更新时间倒序排列。适合在用户询问‘我的待办’‘项目进度’‘任务列表’时使用。效果是完全不同的。前者模型经常不知道该不该调用、该传什么参数后者能大幅减少误调用和重复调用。我自己有个习惯写完一个工具隔一天再读一遍描述如果自己都搞不清这工具是干嘛的那Agent肯定也搞不清。4. 把MCP Server接入Agent全链路联调服务端写好了接下来是Agent侧的接入。这一步最容易出问题很多人以为有工具列表就能跑通实际联调时才发现各种环境问题。4.1 MCP Client建立连接与工具发现Agent端借助官方客户端连接stdio模式的服务端import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): server_params StdioServerParameters( commandpython, args[src/project_assistant/server.py], ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools_result await session.list_tools() for t in tools_result.tools: print(t.name, t.description) resp await session.call_tool(list_tasks, {status: todo}) print(resp.content)跑通这个流程后还要做一步关键转换把MCP Server暴露的tools转换成Agent框架的function calling schema。大部分Agent框架不是原生理解MCP协议格式而是有自己的工具格式。转换通常在Agent的接入层完成def to_agent_schema(mcp_tool): return { type: function, function: { name: mcp_tool.name, description: mcp_tool.description, parameters: mcp_tool.inputSchema, }, }这样Agent框架把转换后的schema发给LLMLLM决定调用哪个工具再由回调函数去执行session.call_tool把结果送回给LLM继续推理。整个链路就闭环了。4.2 工具返回格式对Agent推理的影响session.call_tool返回的结果是CallToolResult其中content是一个ContentBlock列表。每个ContentBlock可以包含文本、图片等不同格式。实操中一个很值得注意的点LLM能处理的是文本结构化数据应该转换成JSON字符串返回而不是丢一个Python对象过去。我见过有人直接从工具函数里返回一个Pydantic对象MCP把它序列化成复杂嵌套结构Agent这边解析半天拿不到关键信息。正确的做法是工具函数内部返回规范的JSON字符串并让description告诉模型“返回的是JSON字符串包含哪些关键字段”。另外工具返回文本的长度要控制。我遇到过list_tasks把几千条任务一次性返回直接撑爆了上下文窗口。所以分页参数和结果截断必须在Server端就控制好别把压力全交给Agent。4.3 超时、错误与重试机制Agent调用工具最怕的就是“等”字。一个MCP Server如果因为业务逻辑或网络问题卡住不响应Agent侧可能一直干等。我的做法是客户端对每次call_tool设置超时用asyncio.wait_for包一层超过指定时间就取消。服务端对耗时操作主动上报进度用ctx.report_progress。调用失败时区分“工具业务失败”如参数错误、数据不存在和“服务端异常”如进程崩溃前者直接返回结构化错误给模型后者才做重试。重试策略上我习惯用指数退避加一点抖动第一次失败等1秒第二次等2秒第三次等4秒最多重试3次。盲目重试往往只是浪费资源得再补一个熔断逻辑——连续失败超过一定次数就主动拉黑该工具一段时间。5. 实战中容易踩的坑我的排查记录MCP-Server看起来代码量不大但上一线跑起来坑不少。这些坑都不是什么高深问题就是位置非常隐蔽我按自己的排查顺序写出来。5.1 stdio模式下print调试会毁掉整个协议第一次联调时我很自然地在一个工具函数里加了句print(收到任务创建请求)然后整个Client会话立刻挂掉报错信息还看不懂。原因在于stdio模式是用标准输出和父进程通信的任何额外的print都会污染JSON-RPC消息流。记住在stdio模式下不要往stdout写任何数据调试一律走logging到stderr或者服务端提供调试专用的debug参数把调试信息写到独立的日志文件。5.2 tools/list返回的Schema不合法导致Agent拒绝解析有一次Agent提示“tool schema不合法”我反复检查工具定义都觉得没问题。后来一个字段一个字段地对照JSON Schema规范发现是某个可选字段的默认值类型和声明的类型不一致SDK生成了带矛盾约束的Schema模型侧解析直接报错。这个问题的规范做法是工具函数的类型注解和默认值必须严格一致别写def f(a: int 1)这种代码。同时建议在Server代码里加一个Schema自检的测试用例每次启动时先把tools/list的结果跑一遍严格JSON Schema校验有问题宁可启动失败也别上线后让Agent莫名其妙拿不到工具。5.3 长耗时操作导致Agent超时进度摘要工具在数据量大了以后可能要扫描很多记录第一次试跑时直接把Agent侧的10秒超时打爆了。我当时的处理是把同步阻塞逻辑改成异步加上ctx.report_progress分阶段上报。FastMCP里可以通过ctx参数拿到上下文对象mcp.tool() async def long_task(ctx) - str: total 100 for i in range(total): await asyncio.sleep(0.1) ctx.report_progress(i 1, total) return done这样Agent端能持续收到进度信号而不是一直干等能接受的等待时长会大大提高。如果任务实在无法在超时时间内完成正确设计是“任务提交接口轮询接口”拆成两步不要硬扛。5.4 同步与异步混用导致的事件循环问题MCP-Server里同时存在同步工具和异步工具时要特别小心。FastMCP内部会调度事件循环但如果你在同步工具里直接调用asyncio.run去执行某个协程很可能碰上“event loop is already running”的错误。我的原则是工具内部统一走异步实现同步包装只是在特殊场景下用。大部分业务IO用httpx.AsyncClient或aiofiles保持整条链路是异步的。5.5 并发访问下的数据竞争当多个Agent会话同时连接同一个MCP-Server时文件型存储的竞争问题会瞬间暴露。即便我加了线程锁也只是解决了单机单进程内的并发跨进程场景还是要靠文件锁或升级数据库。实际上后来我直接把这个项目的存储换成了SQLite用BEGIN IMMEDIATE事务处理写入比手动加锁稳妥得多。这里给两个方案数据量小、单机演示用threading.Lock加JSON文件就够了生产级并发就果断上SQLite或PostgreSQL别把时间花在手工锁上。6. 进阶记忆选型、多Agent协作与安全边界到这里一个能跑的MCP-Server已经成型了。但要做成生产级还有三块需要补充。6.1 Agent记忆的分级与简单实现我实践下来的记忆分级是比较朴素的短期记忆会话内的最近几轮上下文直接存在Agent推理的context里。中期记忆跨会话但时效性强的信息比如“上次查了哪些任务”写进服务端一个带过期时间的KV就行。长期记忆需要考虑语义关联的信息需要向量化存储并做检索。如果你只想用最小成本落地“记忆”不要一开始就上向量库。先用MCP Server里的文件或SQLite存结构化信息Agent按照固定的调用规则去写读。当你会遇到“用户提到了很久之前的一句话”这类场景时再加向量检索也来得及。6.2 多Agent共享MCP-Server时的设计要点多Agent协作的项目里MCP-Server的位置天然适合做“公共能力层”。不同Agent都可以通过同一个Server访问项目数据但要让它们各查各的、互不干扰。我的项目里有两个必做的隔离措施一是工具参数里带上project_id或namespace用业务字段隔离数据范围二是如果Server要为不同的平台或客户服务就必须在Server入口做认证鉴权不能只靠业务参数隔离。否则Agent A误操作把Agent B的数据改了排查起来相当头大。6.3 工具权限与密钥管理的最小可行方案Agent的权限比人的权限更需要管紧。我的最小可行方案是三件事密钥注入MCP-Server运行时的API密钥、数据库密码一律从环境变量注入绝对不进代码仓库。最小权限每个Agent账号只授予它能调用的那部分工具做不到就给工具加白名单。输入校验所有参数都做类型与取值范围校验防止LLM拼接出非预期输入导致安全问题。有条件的一定要加操作日志记录哪个Agent在什么时候调用了哪个工具传了什么参数返回了什么结果。Agent行为出了偏差日志是回溯的唯一手段。这套MCP-Server的开发流程跑通之后我自己最大的感受是真正省下来的不是代码量而是心智负担。以前Agent侧每加一个能力工具都要重新理顺一套契约现在只要往Server里塞一个规范的工具函数Agent框架那边几乎不用动。排错也简单了Agent行为不对先看是不是工具返回了脏数据再看是不是模型没理解工具描述边界非常清晰。最后分享一个调优小技巧联调阶段不要直接接大模型先用MCP Inspector把所有工具手工调用一遍确认每个工具都返回符合预期的结构再让Agent接入。这样能把服务端问题和模型调用问题干净利落地区分开省下的时间足够你多写十个工具。