资讯动态

MCP Server工程结构实战:从模块化到基础设施的工业级设计

发布时间:2026/9/9 21:00:01 来源:尧图企业网站定制
最近连续做了几个 MCP Server 的项目从最早的“能跑就行”到后来被线上问题逼着重构我最大的感受是MCP Server 这个玩意儿协议本身不复杂真正决定项目成败的是工程结构。说得直白一点MCP Server 就是一个给 AI 模型用的“工具插线板”它本身不产生业务逻辑但它把 AI 和你的业务系统连接起来。插线板如果内部乱成一团AI 再怎么聪明也没办法稳定地用上你提供的工具。这篇文章就围绕 MCP Server 的工程结构展开把我自己踩过的坑、反复调整后沉淀下来的那套“工业级”结构拿出来拆开讲。如果你正准备自己搭一个 MCP Server或者已经在做了但总觉得项目越写越乱、越改越难维护这篇文章应该能给你一些可以直接抄的答案。我会从设计思路、目录划分、核心细节、完整实操到问题排查一步步讲不是那种只讲概念的文章基本照着做就能落地。1. 整体设计与思路拆解1.1 别急着写代码先把 MCP Server 的职责边界想清楚很多人的第一个 MCP Server 是从一个简单想法开始的写一个工具函数暴露给 Claude 或者别的 AI 客户端调用。但一旦工具数量超过三五个问题就来了配置散落各处、鉴权逻辑和业务逻辑缠在一起、工具之间共享的数据库连接不知道放在哪儿、想给某个工具单独发版又拖泥带水。我个人的经验是在设计工程结构之前必须先想清楚 MCP Server 的三层职责边界第一层是协议层也就是 SDK 帮你处理的部分比如接收客户端的 JSON-RPC 请求、调用工具、返回结果。这一层你基本不用碰但要清楚它的存在。第二层是工具层这是开发者主要写的部分每个工具完成一件具体的事比如查天气、发邮件、调内部 API。第三层是基础设施层包括配置管理、日志、鉴权、数据库连接、错误处理、可观测性。这三层里的核心是第三层也是最容易在项目初期被忽略的部分。搞清楚了职责边界工程结构就有了骨架核心要保证工具层足够“薄”所有横切关注点都下沉到基础设施层。这样设计的直接好处是以后每新增一个工具只需要关心业务逻辑本身其他东西全部复用。这一条决定了整个项目后面是越写越轻松还是越写越痛苦。1.2 为什么说模块化是工业级 MCP Server 的底线我见过很多人写的 MCP Server所有代码堆在几个大文件里工具函数、数据库操作、鉴权逻辑、错误处理全部混在一起。前期确实快但一旦要加新工具、改字段或者排查线上问题整个人就会陷入“在一个 3000 行的文件里找一段逻辑”的噩梦。模块化的价值在 MCP Server 这个场景下被放大了。原因很简单MCP Server 是一个被 AI 高频调用的服务它不像普通 Web 接口那样有清晰的前端触发流程AI 会根据用户的一句话动态选择调哪个工具这意味着你的代码必须支持快速扩展和独立维护。如果工具和基础设施混在一起哪怕只是改一个日志格式都可能不小心弄坏某个正在被 AI 调用的工具。所以我的建议是从一个工具开始就按模块化的方式组织。模块化不意味着过度设计而是把“变了会互相影响的东西”隔离开。具体到目录结构我后面会详细讲核心原则是工具按领域分目录基础设施按类型分目录入口只负责组装。坚持这个原则项目规模增长到几十个工具的时候结构优势会非常明显。1.3 从单体到分层我踩过的结构坑第一次做 MCP Server 时我图省事把所有东西都放进了main.py大概三四百行看着还行。等加到第七个工具时这个文件已经快一千行每次改代码都要全局搜索函数名哪怕是加一个日志字段都要担心会不会影响别的地方。更痛苦的是测试——因为所有代码都耦合在一起根本没有办法单独测试某个工具的响应逻辑。后来我尝试按“工具类型”分文件比如weather.py、email.py确实比单文件好一些但很快发现一个新的问题数据库连接、配置加载、错误处理这些公共逻辑在每个文件里都复制了一份。改一处数据库地址要全局搜索替换。某个工具的鉴权逻辑写错了排查的时候要翻遍所有文件。最终我采用了分层的单体架构应用是一个整体部署单元但代码内部严格分层工具、基础设施、协议处理各司其职。这套结构支撑我们后续平滑地加了二十多个工具没有再出现改一处崩一片的情况。这个过程让我深刻认识到MCP Server 的工程结构不是在项目大了以后才需要考虑的事而是从第一天起就要按“会变大”的预期来设计。2. 核心细节解析与实操要点2.1 目录结构一套可以直接抄作业的骨架我当前推荐的 MCP Server 目录结构长这样mcp-server/ ├── pyproject.toml ├── .env.example ├── src/ │ └── mcp_server/ │ ├── __init__.py │ ├── main.py │ ├── config.py │ ├── server.py │ ├── tools/ │ │ ├── __init__.py │ │ ├── base.py │ │ ├── weather/ │ │ │ ├── __init__.py │ │ │ ├── handler.py │ │ │ ├── schemas.py │ │ │ └── client.py │ │ ├── email/ │ │ │ ├── __init__.py │ │ │ ├── handler.py │ │ │ ├── schemas.py │ │ │ └── client.py │ │ └── ... │ ├── infrastructure/ │ │ ├── __init__.py │ │ ├── logging.py │ │ ├── auth.py │ │ ├── database.py │ │ ├── errors.py │ │ └── metrics.py │ └── shared/ │ ├── __init__.py │ └── utils.py ├── tests/ │ ├── test_tools/ │ │ ├── test_weather.py │ │ └── test_email.py │ └── test_infrastructure/ └── Makefile这个结构里有几个关键设计我逐个解释一下。src/mcp_server/main.py是入口文件只负责加载配置、初始化基础设施、注册工具然后启动服务。src/mcp_server/config.py负责所有配置项的加载与校验包括环境变量、配置文件、敏感信息。src/mcp_server/tools/是工具目录每个工具一个子目录子目录内部再拆分 handler业务逻辑、schemas输入输出定义、client外部 API 调用。src/mcp_server/infrastructure/放跨工具的公共能力。tests/目录结构跟src一一对应保证每个模块都有对应的测试。这样做的好处是新加一个工具只需要在tools/下新建一个子目录然后在main.py里注册一行公共能力变更只需要改infrastructure/下的对应文件所有工具自动生效。边界清晰互不干扰。这套结构我沿用到现在基本没有再为“代码放哪儿”纠结过。2.2 工具注册机制为什么用声明式而不是硬编码在 MCP Server 里工具注册是把“函数”变成“AI 可以调用的工具”的关键一步。很多人的第一版是硬编码# 每个人都会写的第一个版本 server.tool()(get_weather) server.tool()(send_email)工具少的时候没什么问题但工具一多这种硬编码方式就暴露了一个核心痛点每个工具的定义信息——名称、描述、参数 Schema——分散在装饰器、函数定义、类型注解等各个地方。AI 对工具的理解完全依赖这些描述写不好 AI 就不会正确调用而这时你还要去一个被各种装饰器堆满的文件里找哪里出了问题。我后来换成了声明式注册机制。每个工具目录里有一个schemas.py专门定义输入输出的 Pydantic 模型和描述信息再有一个handler.py实现业务逻辑。然后在main.py里统一注册from mcp_server.tools.weather import weather_tool from mcp_server.tools.email import email_tool TOOLS [ weather_tool, email_tool, ] def register_all_tools(server: Server): for tool in TOOLS: tool.register(server)每个工具子目录向外暴露一个xxx_tool实例这个实例包含工具的元信息、描述、参数 Schema 和 handler。这样 AI 用什么名字调用、传什么参数、得到什么结果全都在一个地方定义清楚。同时对注册机制统一封装可以在注册阶段做参数校验、权限声明、埋点上报等横切逻辑。这个设计的核心好处是在“开发效率”和“AI 可理解性”之间找到了平衡点。声明式让工具的定义信息高度内聚AI 拿到的工具描述永远是完整的、及时的而不是散落在代码各处的碎片。2.3 输入输出 SchemaAI 能不能用对你的工具七成看这里MCP Server 的调用方式决定了工具的参数校验非常关键。你在普通 API 里传错参数前端会报错、调用方会自查。但在 MCP Server 里调用方是 AI 模型它根据你对工具的描述 参数 Schema 来生成调用参数。如果参数定义得太模糊——比如一个query: str然后描述写“查询条件”——AI 根本不知道应该传什么最终结果不是报错就是返回无用数据。我的经验是参数定义要遵循几个原则。第一字段描述要写“业务语境”而不是直译字段名。比如city: str的字段你写“城市名称中文例如北京、上海”而不是“城市”。AI 对中文语境的理解比想象中好你给的信息越具体它生成的调用参数越准确。第二尽量用 enum 而不是裸字符串。比如天气工具有“今天、明天、未来三天”这类选项直接定义成枚举AI 就不会自由发挥了。第三合理设置required。我见过一些人为了省事把所有参数都设为可选这把校验压力全部推给业务逻辑。合理的做法是必填参数设为 required可选参数给默认值这样 AI 即使漏传参数也不至于直接报错而是能按默认逻辑执行。第四返回值也要结构化。AI 拿到一个结构化 JSON 和拿到一段格式化文本后续处理能力完全不一样。我一般统一返回{success: bool, data: ..., message: str}这种结构AI 可以根据success字段判断是否重试或换一种调用方式。这些细节看起来很小但它们直接决定了 AI 能不能“正确地”使用你的工具。很多 MCP Server 体验很差不是 AI 笨而是工具定义没写清楚。2.4 基础设施层的设计要点配置、日志、错误处理、鉴权基础设施层是工业级 MCP Server 和玩具级 MCP Server 的重要分水岭。配置管理上我用pydantic-settings做配置加载所有配置集中在一个Settings类里从环境变量读取而不是在代码里到处os.getenv。这样做的好处是配置项有类型、有校验、有默认值漏配了启动时直接报错而不是跑到一半才炸。日志方面MCP Server 的日志用途跟普通 Web 服务不太一样——你不仅要记录发生了什么还要能追溯“AI 因为什么上下文触发了这次调用”。所以我统一在工具调用入口打印结构化日志包含请求 ID、工具名、参数摘要、耗时、结果状态。这样排查问题的时候可以直接说“这次请求的 request_id 是 xxx它调用了天气工具参数是北京”而不是在一堆日志里大海捞针。错误处理上我写了一个统一的MCPServerError异常基类业务层只负责抛出业务错误由基础设施层统一捕获、转换为 MCP 协议要求的错误结构返回给客户端。这样 AI 拿到的错误信息是结构化的、可理解的而不是一串 Stack Trace。鉴权是最容易被忽略的。很多人觉得 MCP Server 是内部服务不需要鉴权但这个想法非常危险。MCP Server 暴露给 AI 的能力其实就是你的业务能力如果某些工具涉及敏感操作——发邮件、改数据库、调支付接口——必须有鉴权。我的做法是支持两种模式内部网络部署时不鉴权公网部署时通过请求头校验 API Key。这个开关只要一个配置项但设计结构上要预留好位置。3. 实操过程与核心环节实现3.1 从零搭建工程骨架环境准备与初始化下面我带你完整走一遍搭建流程。我用 Python 生态举例因为 MCP 的官方 SDK 对 Python 支持最成熟。其他语言也完全可以LangChain 官方支持的 TypeScript SDK 同样很棒但核心思路是一样的。第一步创建项目目录并初始化虚拟环境mkdir mcp-server cd mcp-server python -m venv .venv source .venv/bin/activate第二步安装依赖pip install mcp[cli] pydantic-settingsmcp[cli]会安装官方 Python SDK 以及mcp命令行工具它自带开发服务器和调试客户端非常方便。第三步创建项目结构与配置文件我直接按 2.1 节的目录来。创建.env.example把需要用到的环境变量列出来# .env.example LOG_LEVELinfo AUTH_TOKEN DATABASE_URL创建pyproject.toml管理项目元信息和依赖这里用 uv 或 poetry 都行我用的是 uv它现在是我最喜欢的 Python 包管理器[project] name mcp-server version 0.1.0 requires-python 3.11 dependencies [ mcp[cli]1.2.0, pydantic-settings2.0.0, ]3.2 核心代码实现入口、配置、基础设施、一个完整工具配置模块是第一个要写的因为其他所有模块都依赖它。我用pydantic-settings实现# src/mcp_server/config.py from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): model_config SettingsConfigDict(env_file.env, env_file_encodingutf-8) log_level: str info auth_token: str database_url: str | None None settings Settings()日志模块给全局打底我用标准库logging加 JSON 格式# src/mcp_server/infrastructure/logging.py import json import logging import sys class JsonFormatter(logging.Formatter): def format(self, record: logging.LogRecord) - str: log_entry { timestamp: self.formatTime(record), level: record.levelname, logger: record.name, message: record.getMessage(), } if hasattr(record, request_id): log_entry[request_id] record.request_id return json.dumps(log_entry, ensure_asciiFalse) def setup_logging(level: str info) - None: handler logging.StreamHandler(sys.stdout) handler.setFormatter(JsonFormatter()) root logging.getLogger() root.handlers [handler] root.setLevel(level.upper())错误处理模块统一异常结构和转换逻辑# src/mcp_server/infrastructure/errors.py class MCPServerError(Exception): def __init__(self, message: str, code: str MCP_SERVER_ERROR): self.message message self.code code super().__init__(message) class ToolExecutionError(MCPServerError): def __init__(self, message: str, tool_name: str): super().__init__(messagemessage, codeTOOL_EXECUTION_ERROR) self.tool_name tool_name鉴权模块设计成中间件形式的校验函数入口处统一调用# src/mcp_server/infrastructure/auth.py from fastapi import HTTPException, Request async def verify_auth(request: Request, expected_token: str) - None: if not expected_token: return # 未配置 token 时跳过鉴权仅限内网 token request.headers.get(Authorization, ).replace(Bearer , ) if token ! expected_token: raise HTTPException(status_code401, detailUnauthorized)到这里基础设施核心模块就绪了。接下来写一个完整的天气工具作为示例。首先是schemas.py定义输入输出# src/mcp_server/tools/weather/schemas.py from typing import Literal from pydantic import BaseModel, Field class WeatherQuery(BaseModel): city: str Field(description城市名称中文例如北京、上海) days: Literal[today, tomorrow, 3days] Field( defaulttoday, description预报范围今天、明天、未来三天 ) class WeatherResult(BaseModel): city: str forecast: str temperature: float humidity: float然后是client.py负责调外部天气 API# src/mcp_server/tools/weather/client.py import httpx class WeatherClient: def __init__(self, api_key: str): self.api_key api_key self.base_url https://api.example.com/weather async def fetch(self, city: str, days: str) - dict: async with httpx.AsyncClient() as client: response await client.get( self.base_url, params{city: city, days: days, key: self.api_key}, timeout5.0, ) response.raise_for_status() return response.json()然后是handler.py写业务逻辑调用 client解析结果# src/mcp_server/tools/weather/handler.py from mcp_server.infrastructure.errors import ToolExecutionError from mcp_server.tools.weather.client import WeatherClient from mcp_server.tools.weather.schemas import WeatherQuery, WeatherResult async def handle_weather(query: WeatherQuery, client: WeatherClient) - WeatherResult: try: data await client.fetch(query.city, query.days) except Exception as e: raise ToolExecutionError(messagef天气服务调用失败: {e}, tool_nameweather) return WeatherResult( citydata[city], forecastdata[forecast], temperaturedata[temperature], humiditydata[humidity], )最后在tools/weather/__init__.py里把工具组装成一个可注册的对象。这里的关键是定义 MCP 工具调用协议——mcpSDK 支持用server.tool()装饰器定义工具也可以用更底层的Tool对象手动注册。我用装饰器方式最少样板代码# src/mcp_server/tools/weather/__init__.py from mcp.server import Server from mcp_server.config import settings from mcp_server.tools.weather.client import WeatherClient from mcp_server.tools.weather.handler import handle_weather from mcp_server.tools.weather.schemas import WeatherQuery _weather_client WeatherClient(api_keysettings.weather_api_key) def register_weather_tool(server: Server) - None: server.tool( nameget_weather, description查询指定城市的天气情况支持今天、明天和未来三天。, ) async def get_weather(query: WeatherQuery) - dict: result await handle_weather(query, _weather_client) return result.model_dump()最后在server.py里组装所有工具和中间件# src/mcp_server/server.py from mcp.server import Server from mcp_server.infrastructure.logging import setup_logging from mcp_server.tools.weather import register_weather_tool from mcp_server.tools.email import register_email_tool def create_server() - Server: setup_logging() server Server(mcp-server) # 注册工具 register_weather_tool(server) register_email_tool(server) # 给 server 挂载鉴权、日志等中间件逻辑 # 具体的挂载方式根据你选择的 MCP 传输层而定 return servermain.py作为入口启动服务# src/mcp_server/main.py from mcp.server import stdio_server from mcp_server.server import create_server async def main(): server create_server() async with stdio_server() as (read_stream, write_stream): await server.run(read_stream, write_stream, server.create_initialization_options()) if __name__ __main__: import asyncio asyncio.run(main())这个流程走下来一个结构化的 MCP Server 就搭好了。这里我用的是 stdio 传输模式方便本地调试部署到远程时可以用 SSE 或 HTTP Streamable 模式SDK 提供了对应的传输构造器核心代码不需要大改。3.3 模板化新增一个业务工具5 分钟起步结构定下来之后新增一个工具就是重复一套固定的流程。以新增一个“发送待办提醒邮件”的工具为例第一步在tools/下创建email_reminder/目录。第二步写schemas.py定义入参和出参# src/mcp_server/tools/email_reminder/schemas.py from pydantic import BaseModel, Field, EmailStr class ReminderEmailRequest(BaseModel): to_email: EmailStr Field(description收件人邮箱) subject: str Field(description邮件主题) body: str Field(description邮件正文支持纯文本) class ReminderEmailResult(BaseModel): message_id: str Field(description邮件服务返回的邮件 ID) status: str Field(defaultsent, description发送状态)第三步写handler.py实现发送逻辑。此时数据库、日志、错误处理全是现成的直接 import 用# src/mcp_server/tools/email_reminder/handler.py from mcp_server.infrastructure.errors import ToolExecutionError from mcp_server.tools.email_reminder.schemas import ReminderEmailRequest, ReminderEmailResult async def handle_send_reminder(req: ReminderEmailRequest, mailer) - ReminderEmailResult: try: message_id await mailer.send(toreq.to_email, subjectreq.subject, bodyreq.body) except Exception as e: raise ToolExecutionError(messagef邮件发送失败: {e}, tool_nameemail_reminder) return ReminderEmailResult(message_idmessage_id)第四步在tools/email_reminder/__init__.py里写注册函数。第五步到server.py的create_server()里加一行register_email_reminder_tool(server)。加一个工具总共就五步而且每一步都是固定的。这种“新增业务工具零思考”的体验正是工业级工程结构带来的最大红利。4. 常见问题与排查技巧实录4.1 为什么 AI 总是不按我预期的方式调用工具这是 MCP Server 上线后最常遇到的问题。用户问“北京明天天气怎么样”AI 调用了get_weather但传的参数是“Beijing”而不是“北京”或者把days传成了“tomorrow”而不是枚举里的“tomorrow”——这种情况非常典型。我的排查思路是先看日志里 AI 实际发送的工具调用参数然后对照schemas.py里的字段描述。绝大多数情况是字段描述写得不够具体AI 只能靠猜测理解参数含义。把描述改得足够业务化、足够具体问题大概率就解决了。比如把city的描述从“城市名称”改成“城市中文名例如北京、上海、广州、深圳”AI 的准确率会立刻上一个台阶。如果字段描述已经写得很清晰但还是调用不对那就看看是不是有同名字段或同义工具导致 AI 混淆了。工具命名和描述要尽量互斥不要出现两个工具都能“查信息”的局面。4.2 工具调用超时问题不一定在 MCP ServerAI 调用工具的等待耐心是有限的如果某个工具执行超过十几秒还不返回AI 客户端可能直接判定失败或者进行重试。我遇到过的最头疼的超时问题表面上看是“MCP Server 响应慢”实际定位下来是工具内部调用的第三方 API 太慢比如某个供应商的接口平均耗时 15 秒。解决思路有两个层面。第一在工具内部对第三方调用加超时控制和熔断机制不要无限等待。比如httpx请求设置timeout5.0超过就快速失败把错误通过结构化的MCPServerError返回给 AI让 AI 决定是重试还是跟用户说明。第二对于确实慢的操作比如生成报告、批量处理尽量改造成异步任务模式工具立即返回“任务已提交任务 ID 为 xxx”再提供一个查询任务状态的工具。这样 AI 的操作体验会好很多用户也不会干等。4.3 鉴权配置了但无效中间件挂载顺序的坑我在把 MCP Server 从本地搬到公网时遇到过鉴权失效的问题明明在verify_auth里写了校验逻辑但请求进来根本没有执行到那一层。查了很久才发现问题出在中间件的挂载顺序上——在 FastAPI 里中间件是按添加顺序执行的而我一开始把鉴权中间件加在了路由解析之后导致请求已经进入了具体接口才去校验。解决方法是把鉴权逻辑作为依赖项放到路由定义上或者在流式传输的 read_stream 处统一做拦截确保任何请求在进入工具执行前就要过鉴权。这里想提醒的是鉴权不是一个可以事后补的东西它必须在请求链路的最前面。如果你用的是自定义传输层而非现成的 HTTP 适配器一定要在设计阶段就留好鉴权钩子的位置。4.4 测试怎么写得既有用又不累为 MCP Server 写测试不需要追求高覆盖率但要把“协议正确性”和“核心业务逻辑”这两层卡住。协议层测试我做的比较轻主要是验证工具注册后客户端用给定的工具名和参数调用能收到预期的结构化响应。这个可以发一个假的 JSON-RPC 请求然后断言响应里的content字段包含预期关键词。业务层测试就针对handler.py写单测。做法很简单把所有外部依赖比如WeatherClient、mailer通过依赖注入传进去测试时用 mock 替换。我不主张拦截太多内部实现细节重点是断言“给定输入handler 返回什么”。这样业务逻辑调整了只要返回结构不变测试就不会频繁废掉。还有一个我强烈建议加的测试test_schemas.py专门测试参数 Schema 的校验行为比如必填字段缺失会不会报错、默认值是否正确、非法枚举值会不会被拒绝。这个测试能帮你在 AI 调用出错时快速定位是 Schema 的问题还是 AI 的问题。5. 关于 MCP Server 工程结构的一些长线思考做 MCP Server 跟做普通后端服务最大的不同就是它的“用户”里多了一个 AI。这个 AI 不像人类用户那样能灵活应对各种意外情况它完全依赖你给的工具描述、参数定义和返回结构来理解这个世界。你代码里模糊的地方在普通 API 里可能只是一个小瑕疵在 MCP Server 里就是一次失败的调用、一个错误的回答。工程结构之所以重要是因为它直接决定了“当 AI 的需求变化时你能不能快速响应”。今天你要加一个工具如果结构清晰你可能只需要十分钟如果结构混乱你可能要花半天解决代码纠缠带来的副作用。在这个 AI 能力快速迭代的时代快速响应不是优势而是底线。再说一句关于标准的问题。MCP 协议本身还在快速演进SDK 也在不断更新。工程结构上做好分层、模块化、配置外部化之后将来协议升级、SDK 切换都只需要动入口层和协议适配层你的业务代码可以基本不动。这一点我在一次 SDK 大版本升级时体验特别深刻因为结构清晰整个升级过程只花了一个小时而且没有引入新的故障。最后分享一个我个人的小习惯每次新做一个 MCP Server 项目我都会把第一周的时间专门用于“设计结构”包括工具目录怎么分、基础设施需要哪些能力、鉴权放哪一层、日志记录哪些字段。这些设计不一定多优雅但一定要想清楚。因为 MCP Server 这个项目的本质是给 AI 构建一个稳定、可靠、可控的工具中枢而这个中枢的地基就是工程结构。地基扎实了后面怎么盖楼都不慌。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价