摘要延续上篇深入分析开源MCP Server的架构模式与设计思路涵盖插件化架构、中间件模式、事件驱动设计和组合式Server等高级模式。开源MCP Server源码解析下架构模式与设计思路上一篇我拆了三个开源MCP Server的源码看了它们的目录结构、工具注册方式和传输实现。但光看代码还不够真正有价值的是背后的架构模式。我花了两天时间把这些项目的代码重新过了一遍提炼出三种核心模式插件式架构、中间件模式和工厂模式。这些模式你直接搬到自己的项目里就能用能少走不少弯路。三种架构模式总览先看一张总览图把三种模式放一起对比。------------------------------------------------------------------ | MCP Server 架构模式全景 | ------------------------------------------------------------------ | | | 插件式架构 中间件模式 工厂模式 | | ----------- ----------- ----------- | | | Server | | Request | | Factory | | | | | Tools | | | Auth | | | create | | | | | Resrc | | | Log | | | stdio | | | | | Prmpt | | | Valid | | | http | | | | | Plugin | | | Handle | | | sse | | | ----------- ----------- ----------- | | 动态加载能力 请求处理管道 多传输统一创建 | | 适合大型Server 适合复杂校验 适合多场景部署 | ------------------------------------------------------------------这三种模式解决的是不同层面的问题。插件式架构解决的是能力扩展问题中间件模式解决的是请求处理流程问题工厂模式解决的是对象创建和多传输切换问题。一个成熟的MCP Server通常会同时用上两到三种。插件式架构插件式架构是我在分析官方 servers 仓库时最先注意到的模式。官方的modelcontextprotocol/server-filesystem、server-github、server-postgres都是独立插件每个插件封装一类能力主程序按需加载。核心思路是定义一个统一的插件接口每个插件自己管理工具注册、资源暴露和提示模板。主程序只负责加载插件和启动传输。---------------------------------- | MCP Server 主程序 | | ------ ------ ------ | | |FS插件 | |GH插件 | |DB插件 | | | |tools | |tools | |tools | | | |resrc | |resrc | |resrc | | | ------ ------ ------ | | 统一插件接口 PluginProtocol | ----------------------------------我自己的一个项目就用这个模式重构过。原来二十多个工具全写在一个文件里一千多行改一个工具要翻半天。拆成插件后每个插件一个文件改哪个动哪个互不干扰。插件接口的定义很关键太复杂了上手成本高太简单了覆盖不了场景。我从官方代码里提炼了一个最小可用接口。中间件模式中间件模式处理的是请求和响应的流转。MCP的JSON-RPC消息进来后在到达业务逻辑之前要经过一系列处理认证、日志、参数校验、限流、错误捕获。中间件模式把这些横切关注点拆成独立的处理器串联成管道。Request - [Auth] - [Log] - [Validate] - [RateLimit] - Handler | Response - [Log] - [ErrorCatch] - [Format] -------- Result这个模式在TypeScript SDK里体现得最明显。官方SDK的McpServer类内部就用了一套中间件链来处理工具调用。Python SDK的FastMCP虽然没有显式的中间件API但它的装饰器机制本质上做了同样的事。我踩过一个坑。早期写MCP Server时把认证逻辑直接写在每个工具函数里十几个工具就有十几份重复的认证代码。后来抽成中间件认证逻辑只写一处所有工具自动经过认证检查。代码量减了三百行维护成本直线下降。中间件模式的另一个好处是可插拔。开发环境关掉认证中间件加上详细日志中间件。生产环境反过来。通过配置控制中间件链不用改业务代码。工厂模式工厂模式在MCP Server里主要用在两个地方传输工厂和工具工厂。传输工厂根据配置创建不同类型的传输实例。MCP支持stdio和Streamable HTTP两种传输它们的初始化参数完全不同。工厂模式把创建逻辑封装起来调用方只管传一个类型字符串。TransportFactory | -- create(stdio) - StdioServerTransport -- create(http) - StreamableHTTPTransport -- create(sse) - SSETransport (旧版兼容)工具工厂用得少一些但在动态工具场景下很有用。比如你的Server连了多个数据库每个数据库暴露一组查询工具。工具工厂根据数据库配置动态生成工具不用手动一个个注册。我在做一个多租户MCP Server时用了工具工厂。每个租户有自己的一组工具租户上线时工厂自动生成对应的工具集租户下线时自动注销。如果手动注册每加一个租户就要改代码完全没法扩展。设计思路对比我把三种模式放一起做个对比帮你判断什么场景该用哪种。维度插件式架构中间件模式工厂模式解决问题能力扩展请求处理对象创建核心机制统一接口加动态加载责任链管道封装创建逻辑复杂度中等低到中等低适合规模工具超过10个有认证/限流需求多传输/动态工具典型项目官方servers仓库TypeScript SDKPython SDK内部三个模式不互斥。我自己的生产项目同时用了插件式加中间件。插件管工具分组中间件管横切逻辑。工厂模式用来切换传输方式开发用stdio部署用HTTP。选型建议很简单。工具少5个以内直接写不用折腾模式。工具多或者需要动态增减上插件式架构。有认证、限流、审计需求加中间件。需要支持多种部署方式用工厂模式。可复用的设计模式总结我把从开源项目里提炼出来的经验整理成几条可操作的规则。规则一工具分组用命名空间。官方server-github的工具名都是github_xxxserver-filesystem是filesystem_xxx。这个命名约定让多个Server的工具合并时不会冲突。你自己写工具也遵守这个规则用项目名_功能名格式。规则二资源URI要有版本号。官方server-postgres的资源URI格式是postgres://table/{schema}/{table}/schema。我在自己的项目里加了一层版本变成postgres://v1/table/{schema}/{table}/schema。API升级时老客户端还能用v1新客户端走v2。规则三配置和代码分离。这条看着像废话但我见过太多把数据库连接串硬编码在Python文件里的MCP Server。用环境变量或者配置文件管理敏感信息MCP的客户端配置本身就支持env字段传环境变量直接用。规则四错误信息要人能读。MCP工具返回的错误会被LLM读到然后LLM根据错误信息决定下一步。如果你的错误信息是Error code 5003LLM根本不知道怎么回事。把错误信息写成自然语言比如GitHub API rate limit exceeded, please retry after 60 secondsLLM能自动处理。规则五异步优先。MCP协议本身就是异步的JSON-RPC消息通过异步管道传递。你的工具函数用async defIO操作用httpx、aiosqlite这些异步库。我做过对比测试同样查10个API同步实现要12秒异步实现2秒搞定。完整代码下面是一个融合了三种架构模式的完整MCP Server模板。它用插件式架构管理工具中间件处理请求工厂创建传输。# pattern_server.py# 融合插件式架构、中间件模式、工厂模式的MCP Server# 依赖安装: pip install mcp[cli] httpximportasyncioimportloggingimportsysimporttimefromabcimportABC,abstractmethodfromdataclassesimportdataclass,fieldfromfunctoolsimportwrapsfromtypingimportAny,Callablefrommcp.server.fastmcpimportFastMCP,Context# ---------- 日志配置 ----------# MCP stdio 传输下只能写 stderr写 stdout 会破坏 JSON-RPC 消息logging.basicConfig(levellogging.INFO,format%(asctime)s [%(name)s] %(levelname)s %(message)s,streamsys.stderr,# 关键: stdio 模式必须用 stderr)loggerlogging.getLogger(pattern-server)# # 第一部分: 插件式架构# # 插件协议定义所有插件实现这个接口classPluginProtocol(ABC):MCP Server 插件接口。 每个插件负责注册自己的工具、资源、提示。 主程序通过这个接口统一管理所有插件。 propertyabstractmethoddefname(self)-str:插件名称用作工具命名空间前缀。...abstractmethoddefregister(self,mcp:FastMCP)-None:把插件的能力注册到 MCP 实例上。...# 文件搜索插件classFileSearchPlugin(PluginProtocol):文件搜索插件提供按名称和内容搜索文件的能力。propertydefname(self)-str:returnfilesearchdefregister(self,mcp:FastMCP)-None:# 注册工具工具名带命名空间前缀避免冲突mcp.tool()asyncdeffilesearch_by_name(directory:str,pattern:str)-str:按文件名模式搜索文件。 Args: directory: 搜索目录的绝对路径 pattern: 文件名匹配模式支持 * 通配符 Returns: 匹配的文件路径列表 importfnmatchimportos# 安全校验: 只允许搜索指定目录下的文件ifnotos.path.isabs(directory):return错误: directory 必须是绝对路径matches[]forroot,dirs,filesinos.walk(directory):forfilenameinfnmatch.filter(files,pattern):matches.append(os.path.join(root,filename))ifnotmatches:returnf未找到匹配{pattern}的文件return\n.join(matches[:20])# 限制返回数量logger.info(f插件{self.name}已注册 1 个工具)# 时间工具插件classTimePlugin(PluginProtocol):时间工具插件提供时间查询和格式化能力。propertydefname(self)-str:returntimedefregister(self,mcp:FastMCP)-None:mcp.tool()asyncdeftime_now(timezone:strUTC)-str:获取当前时间。 Args: timezone: 时区名称默认 UTC Returns: 当前时间的格式化字符串 fromdatetimeimportdatetime,timezoneastz# 简化处理, 真实场景用 pytz 或 zoneinfonowdatetime.now(tz.utc)returnf当前时间 ({timezone}):{now.strftime(%Y-%m-%d %H:%M:%S)}logger.info(f插件{self.name}已注册 1 个工具)# 插件注册表管理所有已加载的插件classPluginRegistry:插件注册表负责加载和管理所有插件。def__init__(self):self._plugins:list[PluginProtocol][]defregister_plugin(self,plugin:PluginProtocol)-None:注册一个插件。self._plugins.append(plugin)logger.info(f插件{plugin.name}已加载到注册表)definstall_all(self,mcp:FastMCP)-None:把所有插件注册到 MCP 实例上。forplugininself._plugins:plugin.register(mcp)logger.info(f插件{plugin.name}安装完成)deflist_plugins(self)-list[str]:列出所有已注册的插件名称。return[p.nameforpinself._plugins]# # 第二部分: 中间件模式# dataclassclassRequestContext:请求上下文在中间件链中传递。tool_name:str# 被调用的工具名arguments:dict# 工具参数start_time:float# 请求开始时间metadata:dictfield(default_factorydict)# 附加元数据# 中间件类型: 接收上下文和下一步处理器MiddlewareFuncCallable[[RequestContext,Callable],Any]defauth_middleware(ctx:RequestContext,next_handler:Callable)-Any:认证中间件: 检查请求是否携带有效凭证。 在真实项目中从 ctx.metadata 或环境变量读取 token。 tokenctx.metadata.get(auth_token,)ifnottoken:logger.warning(f工具{ctx.tool_name}调用未携带认证 token)# 开发环境不拦截生产环境应返回错误else:logger.info(f认证通过, token:{token[:8]}...)returnnext_handler(ctx)deflogging_middleware(ctx:RequestContext,next_handler:Callable)-Any:日志中间件: 记录请求信息和执行耗时。logger.info(f调用工具{ctx.tool_name}, 参数:{ctx.arguments})resultnext_handler(ctx)elapsedtime.time()-ctx.start_time logger.info(f工具{ctx.tool_name}执行完成, 耗时{elapsed:.3f}s)returnresultdefrate_limit_middleware(ctx:RequestContext,next_handler:Callable)-Any:限流中间件: 简单的频率控制。 真实项目用 Redis 或令牌桶算法。 # 简化: 每秒最多 10 次调用nowtime.time()last_callctx.metadata.get(_last_call_time,0)ifnow-last_call0.1:logger.warning(f工具{ctx.tool_name}触发限流)return错误: 请求过于频繁请稍后重试ctx.metadata[_last_call_time]nowreturnnext_handler(ctx)classMiddlewareChain:中间件链按顺序执行所有中间件后到达业务逻辑。def__init__(self,middlewares:list[MiddlewareFunc]):self._middlewaresmiddlewaresdefexecute(self,tool_name:str,arguments:dict,handler:Callable)-Any:执行中间件链。ctxRequestContext(tool_nametool_name,argumentsarguments,start_timetime.time(),)# 构建中间件链从后往前包装defmake_chain(index:int)-Callable:ifindexlen(self._middlewares):returnlambdac:handler(c.tool_name,c.arguments)returnlambdac:self._middlewares[index](c,make_chain(index1))returnmake_chain(0)(ctx)# # 第三部分: 工厂模式# classTransportFactory:传输工厂: 根据配置创建不同的传输模式。 让 Server 代码不用关心传输细节 切换 stdio 和 HTTP 只需改一个参数。 staticmethoddefcreate(transport_type:str,**kwargs)-dict:创建传输配置。 Args: transport_type: 传输类型stdio 或 http **kwargs: 传输特定参数 Returns: 传输配置字典传给 mcp.run() iftransport_typestdio:logger.info(使用 stdio 传输)return{transport:stdio}eliftransport_typehttp:hostkwargs.get(host,0.0.0.0)portkwargs.get(port,8000)logger.info(f使用 HTTP 传输, 地址{host}:{port})return{transport:streamable-http,host:host,port:port,}else:raiseValueError(f不支持的传输类型:{transport_type})# # 第四部分: 主程序组装三种模式# defcreate_server()-tuple[FastMCP,PluginRegistry]:创建并配置 MCP Server 实例。 返回 FastMCP 实例和插件注册表 调用方可以继续往注册表里加插件。 # 创建 FastMCP 实例mcpFastMCP(pattern-server,instructions(这是一个演示三种架构模式的 MCP Server。提供文件搜索和时间查询能力。),)# 创建插件注册表并加载插件registryPluginRegistry()registry.register_plugin(FileSearchPlugin())registry.register_plugin(TimePlugin())# 把插件安装到 MCP 实例上registry.install_all(mcp)logger.info(fServer 创建完成, 已加载插件:{registry.list_plugins()})returnmcp,registrydefmain():主入口: 创建 Server 并启动。importos# 创建 Servermcp,registrycreate_server()# 通过环境变量决定传输方式# 开发时用 stdio部署时用 httptransport_typeos.environ.get(MCP_TRANSPORT,stdio)# 用工厂创建传输配置iftransport_typehttp:configTransportFactory.create(http,hostos.environ.get(MCP_HOST,0.0.0.0),portint(os.environ.get(MCP_PORT,8000)),)else:configTransportFactory.create(stdio)logger.info(f启动 Server, 传输配置:{config})mcp.run(**config)if__name____main__:main()效果验证安装依赖后直接运行。# 安装依赖pipinstallmcp[cli]httpx# stdio 模式运行给 Claude Desktop 用python pattern_server.py# HTTP 模式运行给远程客户端用MCP_TRANSPORThttpMCP_PORT8000python pattern_server.py用 MCP Inspector 测试工具调用。stdio模式下执行mcp dev pattern_server.pyInspector 会自动连接你能看到filesearch_by_name和time_now两个工具。调用time_now返回当前时间调用filesearch_by_name传一个目录和通配符模式就能搜到文件。HTTP模式下用 curl 测试。# 列出工具curl-XPOST http://localhost:8000/mcp\-HContent-Type: application/json\-d{jsonrpc:2.0,id:1,method:tools/list}切换传输模式不用改一行代码只改环境变量MCP_TRANSPORT这就是工厂模式的价值。常见问题与避坑坑一插件加载顺序导致依赖冲突。我有一次写了个数据库插件依赖时间插件的结果但加载顺序反了时间插件还没注册数据库插件就去调直接报错。解决方法是插件之间不要直接依赖通过共享上下文或者主程序中转。如果确实有依赖关系在PluginRegistry里加个拓扑排序。坑二中间件吞掉异常导致问题难排查。中间件里next_handler(ctx)如果抛异常中间件没有 try-except 捕获异常会向上传播到 MCP 框架层但框架层的错误信息不够详细。我在每个中间件里都加了 try-except捕获到异常后用 logger 记录完整堆栈再把异常重新抛出去。这样框架层看到的是原始异常日志里有完整信息。坑三工厂模式创建HTTP传输时端口冲突。开发时经常同时跑多个Server端口写死就会冲突。我在TransportFactory.create里加了端口检测逻辑如果指定端口被占用就自动加1重试。生产环境不要这么干端口必须固定用配置文件管理。坑四stdio模式下误用print导致协议崩溃。这个坑我在第19篇文章里提过但值得再说一次。Python的print()默认写 stdout而stdio传输的JSON-RPC消息也走stdout。你的print输出会插到协议消息中间客户端解析直接炸。所有日志必须写stderr用logging模块配置streamsys.stderr或者用print(msg, filesys.stderr)。坑五插件热重载在Windows上不生效。我做过一个功能让Server运行时动态加载新插件在macOS和Linux上正常到Windows上就不行。原因是Windows的文件锁机制跟Unix不同已加载的py文件不能直接覆盖。解决办法是用importlib的reload机制配合子进程重启或者干脆不做热重载改用信号量触发整个Server重启。小结从开源MCP Server项目中提炼出来的三种架构模式各有侧重。插件式架构解决能力扩展适合工具多的项目。中间件模式解决请求处理适合有认证和限流需求的场景。工厂模式解决对象创建适合多传输和多部署环境的需求。这三种模式不是理论上的东西是从实际项目里长出来的。官方servers仓库用插件式架构管理几十个独立Server。TypeScript SDK用中间件链处理JSON-RPC消息。Python SDK用工厂模式创建不同传输。你把这些模式理解透了写出来的MCP Server结构清晰、可维护、可扩展。下一篇我会把这些模式整合成一个完整的Server开发脚手架拿来就能用。相关推荐开源MCP Server源码解析上精选3个优秀项目Server最佳实践我从20个MCP Server中总结的经验MCP Server模板工程一套可复用的Server开发脚手架