MCP服务开发实战：从零编写并挂载私有Python工具到智能体

分享一下 MCPModel Context Protocol服务开发与私有工具挂载的完整落地过程。最近在接企业内部智能体时最花时间的其实不是大模型本身而是怎么把现有 Python 脚本、内部接口、私有数据安全地暴露给智能体同时还要保证权限可控、调用可追踪。这套方案梳理清楚后整套流程可以稳定复用到多个业务场景。这篇文章会从 MCP 的核心概念讲起带大家从零编写一个私有 Python MCP 工具再把工具挂载到智能体上完成一次真实调用内容覆盖代码示例、完整配置、常见报错和工程落地建议前后端开发、运维和算法同学都可以直接参考。1. 背景与核心概念1.1 MCP 是什么MCP 全称 Model Context Protocol翻译过来是“模型上下文协议”。它解决的核心问题是大模型如何安全、标准化的访问外部数据和工具。在没有 MCP 之前要让大模型调用内部系统通常需要为每个场景单独写一套工具调用逻辑。比如 A 系统用 HTTP 接口暴露B 系统用 Python SDKC 系统只有数据库权限每一套的接入方式都不一样。大模型应用要对接 N 个系统就要维护 N 套工具接入层代码耦合高权限也不好统一控制。MCP 的思路是定义一套统一的协议层把“大模型应用”和“外部工具/数据源”隔离开。大模型应用只需要理解 MCP 协议工具提供方只需要按照 MCP 规范暴露自己的能力两者通过标准化的消息格式通信。这样新增一个工具时不需要改动大模型主程序只需要在 MCP 服务端注册一个新工具即可。1.2 私有工具与智能体挂载这里的“私有工具”指的是不对外公开、只服务于企业内部智能体的工具。它通常包含三种类型内部业务接口封装例如查询订单、创建工单。私有数据查询例如读取公司内部数据库。本地脚本能力例如执行某个服务器上的运维脚本。所谓“挂载”是指把编写好的 MCP 工具注册到智能体运行时中让大模型在对话中可以根据用户意图自动选择并调用这个工具。挂载完成后用户用自然语言提问智能体内部判断需要调用哪个工具再把结果组织成答案返回给用户。1.3 为什么需要自己开发私有工具市面上已经有不少现成的 MCP 工具集但大部分是面向公开数据源的。企业内部场景往往有很强的定制性例如数据表结构是私有的。接口鉴权方式特殊。工具输入参数需要结合业务语义。需要对接旧系统的非标准协议。这些情况都必须自己编写 MCP 工具。自研工具的好处是可以完全控制工具的输入输出格式。可以复用企业已有的 Python 代码资产。可以在工具层加入权限校验、审计日志、限流逻辑。不依赖第三方平台的稳定性。1.4 常见应用场景场景说明企业内部知识问答智能体调用搜索工具从私有知识库中检索答案工单系统自动处理智能体调用建单工具根据用户描述自动创建工单数据报表查询智能体调用 SQL 查询工具返回业务报表数据运维自动化智能体调用脚本工具完成日志检查、配置变更等操作客户服务助手智能体调用客户信息查询工具辅助客服快速定位问题这些场景的共同点是工具本身已经存在Python 脚本、API、数据库操作缺的是连接智能体的通道而 MCP 就是这条通道。2. 环境准备与版本说明2.1 基础运行环境本文的示例以 Linux / macOS 环境为主Windows 系统的差异点会在文中单独说明。整体环境如下组件说明操作系统LinuxCentOS 7或 macOS 12Python3.10 及以上MCP SDKmcpPython 包官方 SDK智能体运行时支持 MCP 协议的服务端或客户端IDEVS Code / PyCharm包管理工具pip / pipenv / poetry 均可版本说明MCP 协议目前仍在快速迭代中不同版本的 SDK 在 API 命名上略有差异。本文的示例以常见的稳定版本写法为准如果你安装的 SDK 版本较新建议对照官方文档中的迁移说明调整导入路径和方法名。本文的重点是演示 MCP 工具开发的核心思路而不是绑定某一个具体版本。2.2 安装 MCP SDK创建一个虚拟环境并安装依赖# 创建虚拟环境 python3 -m venv venv # 激活虚拟环境Linux / macOS source venv/bin/activate # Windows 系统激活命令 # venv\Scripts\activate # 安装 MCP SDK pip install mcp如果想使用开发版或最新特性可以安装 nightly 版本pip install mcp-nightly非必要不建议在生产环境使用开发版。MCP SDK 会同时安装mcp命令行工具安装完成后可以验证mcp --version如果命令输出类似0.x.x或1.x.x的版本号说明安装成功。2.3 项目结构说明本文的示例项目结构如下后续所有代码都按这个结构组织mcp-demo/ ├── venv/ # 虚拟环境 ├── server.py # MCP 服务端入口 ├── tools/ │ ├── __init__.py │ ├── calculator.py # 示例工具计算器 │ └── order.py # 示例工具订单查询 ├── requirements.txt # 依赖清单 └── scripts/ └── test_client.py # 简易测试客户端在实际项目中你可以把工具按业务模块拆分到不同文件中但入口文件统一暴露 MCP 服务。3. MCP 工具开发核心原理3.1 MCP 服务的基本构成一个 MCP 服务通常包含三层结构协议层负责 MCP 协议的握手、消息编解码、会话维护。工具层定义工具的名称、描述、输入参数和输出结构。逻辑层执行工具实际业务逻辑例如访问数据库、调用 API。在 Python SDK 中我们主要关注工具层的编写。SDK 提供了装饰器或类继承两种方式来注册工具。先看一个最简单的 MCP 服务示例# 文件路径server.py from mcp.server.fastmcp import FastMCP # 创建 MCP 服务实例 mcp FastMCP(demo-server) mcp.tool() def hello(name: str) - str: 简单的问候工具 return fHello, {name}! if __name__ __main__: mcp.run()这段代码做了三件事创建了一个名为demo-server的 MCP 服务实例。注册了一个名为hello的工具工具接收一个字符串参数name返回字符串。调用mcp.run()启动服务。FastMCP是官方 SDK 提供的高层封装对于大多数业务工具开发来说已经足够。它的内部会自动处理协议细节我们只需要关注工具函数的实现。3.2 工具函数的输入输出规范MCP 工具函数的参数是基于函数签名自动推断的。也就是说函数的参数名、类型注解、默认值、文档字符串都会成为工具协议的一部分。定义一个带多个参数的工具from typing import Optional mcp.tool() def query_order( order_id: str, customer_name: Optional[str] None, status: str all ) - dict: 查询订单信息。 Args: order_id: 订单编号。 customer_name: 客户名称可选项。 status: 订单状态筛选可选值 pending / paid / shipped / all。 # 这里省略实际业务逻辑返回模拟数据 result { order_id: order_id, customer_name: customer_name or 未知客户, status: status, items: [] } return result这里有几点需要特别注意类型注解必须写完整工具参数的 JSON Schema 是根据类型注解自动生成的。文档字符串会被用作工具描述大模型会根据这段描述判断何时调用该工具所以描述要写清楚用途。返回值建议统一使用 dict 或符合 JSON 结构的类型因为大模型消费的是结构化数据。可选参数使用 Optional 或设置默认值避免大模型因为缺少参数而调用失败。3.3 同步工具与异步工具实际业务中会遇到需要等待外部接口响应的情况。MCP SDK 同时支持同步工具和异步工具推荐使用异步方式处理 IO 密集型任务。import asyncio mcp.tool() async def fetch_remote_data(url: str) - dict: 模拟调用远程接口获取数据 await asyncio.sleep(1) # 模拟网络延迟 return {url: url, data: mock data}如果业务逻辑是 CPU 密集型例如复杂的计算任务可以使用同步函数让 SDK 自动调度到线程池执行。mcp.tool() def compute_hash(data: str) - str: 计算字符串的 SHA256 哈希值 import hashlib return hashlib.sha256(data.encode()).hexdigest()3.4 工具错误处理工具在执行过程中可能会遇到各种异常例如参数不合法。数据库连接失败。第三方接口返回错误。权限不足。错误的处理方式会直接影响大模型的行为。如果工具抛出未捕获异常大模型可能无法理解发生了什么。更推荐的做法是捕获异常后返回结构化的错误信息。mcp.tool() def divide(a: float, b: float) - dict: 两个数相除 try: result a / b return {success: True, result: result} except ZeroDivisionError: return { success: False, error: 除数不能为零, code: DIVIDE_BY_ZERO } except Exception as e: return { success: False, error: f计算失败: {str(e)}, code: UNKNOWN_ERROR }这样做的好处是大模型可以从error字段中知道失败原因。业务方可以根据code字段做后续处理。日志中保留了完整的错误上下文。3.5 服务启动与连接方式MCP 服务支持多种传输方式常用的是stdio标准输入输出传输适合本地集成。HTTP/SSE通过 HTTP 协议提供服务的流式传输适合远程部署。FastMCP的run()方法默认使用 stdio。在本地调试时可以直接运行脚本然后用客户端连接python server.py如果希望在网络中提供服务可以显式指定传输方式。以下是一个 HTTPServer 示例# 文件路径server.pyHTTP 版本 from mcp.server.fastmcp import FastMCP mcp FastMCP(demo-server, host0.0.0.0, port8000) mcp.tool() def ping() - str: 探活工具 return pong if __name__ __main__: mcp.run(transporthttp)配置传输方式时需要考虑网络环境和安全策略服务器部署方式在生产环境需要配合访问控制和 TLS。4. 完整实战编写私有 Python MCP 工具并挂载到智能体下面我们从零开始编写一个带有实际业务含义的 MCP 工具集一个“订单助手”工具包含订单查询、订单统计和客户信息查询三个工具。整个过程分为项目初始化、工具代码编写、服务启动与调试、挂载到智能体、完整调用验证。4.1 初始化项目创建项目目录并安装依赖mkdir mcp-order-demo cd mcp-order-demo python3 -m venv venv source venv/bin/activate pip install mcp创建requirements.txt文件mcp创建项目所需的目录结构mkdir tools scripts touch tools/__init__.py4.2 编写订单工具先写订单查询工具。在实际业务中这个工具可以对接内部订单数据库为了方便演示我们先用模拟数据代替。# 文件路径tools/order.py from typing import Optional # 模拟订单数据 MOCK_ORDERS [ { order_id: ORD20250101, customer_name: 张三, amount: 299.00, status: paid, create_time: 2025-01-01 10:00:00 }, { order_id: ORD20250102, customer_name: 李四, amount: 599.00, status: shipped, create_time: 2025-01-02 11:30:00 }, { order_id: ORD20250103, customer_name: 王五, amount: 1299.00, status: pending, create_time: 2025-01-03 09:15:00 } ] def query_order(order_id: str) - dict: 根据订单编号查询订单详情。 Args: order_id: 订单编号例如 ORD20250101。 for order in MOCK_ORDERS: if order[order_id] order_id: return {success: True, order: order} return {success: False, error: 订单不存在, order: None} def list_orders(status: Optional[str] all) - dict: 查询订单列表可按状态筛选。 Args: status: 筛选状态可选 all / pending / paid / shipped。 if status all: filtered MOCK_ORDERS else: filtered [o for o in MOCK_ORDERS if o[status] status] return {success: True, total: len(filtered), orders: filtered} def order_statistics() - dict: 统计订单总额和订单数量。 total_amount sum(o[amount] for o in MOCK_ORDERS) total_count len(MOCK_ORDERS) paid_amount sum(o[amount] for o in MOCK_ORDERS if o[status] paid) shipped_amount sum(o[amount] for o in MOCK_ORDERS if o[status] shipped) pending_amount sum(o[amount] for o in MOCK_ORDERS if o[status] pending) return { success: True, statistics: { total_count: total_count, total_amount: round(total_amount, 2), paid_amount: round(paid_amount, 2), shipped_amount: round(shipped_amount, 2), pending_amount: round(pending_amount, 2) } }这里把业务逻辑放在tools/order.py中没有直接使用 MCP 装饰器这是故意的。这样做的好处是业务函数可以被普通 Python 脚本复用也可以被单测直接调用。MCP 协议层逻辑单独放在服务入口中。4.3 编写 MCP 服务入口在server.py中引入上面的业务函数并注册为 MCP 工具。# 文件路径server.py from mcp.server.fastmcp import FastMCP from tools.order import query_order, list_orders, order_statistics # 创建 MCP 服务实例 mcp FastMCP(order-assistant-server) mcp.tool() async def query_order_tool(order_id: str) - dict: 根据订单编号查询订单详情包括订单状态、金额、客户名称和创建时间。 Args: order_id: 订单编号例如 ORD20250101。 return query_order(order_id) mcp.tool() async def list_orders_tool(status: str all) - dict: 查询订单列表可按订单状态筛选。 Args: status: 筛选状态可选 all / pending / paid / shipped。 return list_orders(status) mcp.tool() async def order_statistics_tool() - dict: 统计全部订单的总金额、总数量和不同状态的金额分布。 return order_statistics() if __name__ __main__: # 默认以 stdio 方式启动调试时可观察日志输出 mcp.run()代码说明mcp.tool()装饰器接收一个异步函数自动注册为 MCP 工具。每个工具的名称默认等于函数名例如query_order_tool。工具描述直接取函数文档字符串建议在文档中写清楚参数含义和取值示例。实际逻辑仍然调用tools/order.py中的普通函数保持逻辑层纯净。4.4 本地启动 MCP 服务在项目根目录激活虚拟环境后运行python server.py如果一切正常服务会进入监听状态但 stdio 模式下不会有明显的输出。此时可以编写一个简单的测试客户端来验证服务是否可用。先安装客户端依赖pip install mcp[cli]然后使用mcp命令行工具连接mcp dev server.pymcp dev是 SDK 提供的开发模式命令会启动一个带调试面板的服务可以在浏览器中查看工具列表并手动测试工具调用。如果没有图形界面环境可以通过下面这个简单的 Python 客户端脚本验证# 文件路径scripts/test_client.py import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): # 传入服务端启动命令 server_params StdioServerParameters( commandpython, args[server.py] ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: # 初始化会话 await session.initialize() # 获取工具列表 tools await session.list_tools() print(可用工具列表) for tool in tools.tools: print(f - {tool.name}: {tool.description}) # 调用订单查询工具 result await session.call_tool( query_order_tool, {order_id: ORD20250101} ) print(调用结果) print(result) if __name__ __main__: asyncio.run(main())运行测试客户端python scripts/test_client.py预期会输出类似下面这样的内容可用工具列表 - query_order_tool: 根据订单编号查询订单详情包括订单状态、金额、客户名称和创建时间。 - list_orders_tool: 查询订单列表可按订单状态筛选。 - order_statistics_tool: 统计全部订单的总金额、总数量和不同状态的金额分布。 调用结果 content[TextContent(text{success: true, order: ...}, typetext)]到这里本地 MCP 服务已经开发完成并且可以通过客户端正常调用。4.5 挂载到智能体运行时“挂载”这一步取决于你使用的智能体运行时不同平台的挂载方式不同但核心逻辑一致让智能体知道“提供 MCP 服务的进程地址”和“可以调用哪些工具”。方式一本地智能体直接连接如果你的智能体运行在本地通常支持在配置文件中声明 MCP 服务地址。例如某智能体客户端的配置文件{ mcpServers: { order-assistant: { command: python, args: [/path/to/mcp-order-demo/server.py], env: {} } } }配置完成后重启智能体它就会通过 stdio 与你的 MCP 服务建立连接。方式二通过 HTTP 方式挂载如果需要远程挂载将 MCP 服务改为 HTTP 模式。修改server.py的启动部分if __name__ __main__: # 绑定到 0.0.0.0 以允许外部访问注意生产环境需要配合防火墙和鉴权 mcp.run(transporthttp)启动后服务监听在 8000 端口python server.py在智能体配置中将 MCP 服务地址指向 HTTP 接口{ mcpServers: { order-assistant: { url: http://your-server-ip:8000/mcp } } }需要注意不同智能体运行时对 HTTP 模式下 MCP endpoint 的路径规则可能不同具体路径要以运行时文档为准。4.6 智能体会话中的完整调用验证挂载完成后使用自然语言与智能体对话。下面是完整会话示例用户输入帮我查一下订单 ORD20250101 的详情。智能体内部会执行类似这样的流程理解用户意图判断需要订单查询工具。从可用工具中选择query_order_tool。将order_id ORD20250101填入工具参数。调用 MCP 服务获得工具返回结果。将工具返回的 JSON 数据整理成自然语言回复。智能体最终回复订单 ORD20250101 详情如下 - 客户姓名张三 - 订单金额299.00 元 - 订单状态paid已支付 - 创建时间2025-01-01 10:00:00再试一个统计类问题用户输入我们目前的订单总体情况怎么样智能体会调用order_statistics_tool然后回复截至目前共有 3 笔订单订单总金额为 2197.00 元。其中 - 已支付299.00 元 - 已发货599.00 元 - 待处理1299.00 元到这里从编写 Python 工具到挂载智能体整个流程已经跑通。4.7 完整代码组织回顾最后把完整的项目文件梳理一下mcp-order-demo/ ├── venv/ ├── server.py ├── requirements.txt ├── tools/ │ ├── __init__.py │ └── order.py └── scripts/ └── test_client.pyserver.py完整内容如下# 文件路径server.py from mcp.server.fastmcp import FastMCP from tools.order import query_order, list_orders, order_statistics mcp FastMCP(order-assistant-server) mcp.tool() async def query_order_tool(order_id: str) - dict: 根据订单编号查询订单详情包括订单状态、金额、客户名称和创建时间。 Args: order_id: 订单编号例如 ORD20250101。 return query_order(order_id) mcp.tool() async def list_orders_tool(status: str all) - dict: 查询订单列表可按订单状态筛选。 Args: status: 筛选状态可选 all / pending / paid / shipped。 return list_orders(status) mcp.tool() async def order_statistics_tool() - dict: 统计全部订单的总金额、总数量和不同状态的金额分布。 return order_statistics() if __name__ __main__: mcp.run()tools/order.py的内容在前面已经给出这里不再重复。实际项目中tools/order.py里的模拟数据应该替换为数据库查询或内部 API 调用。5. 常见问题与排查思路5.1 工具调用时报错 “Tool not found”这个错误说明智能体运行时没有找到对应的工具。可能原因MCP 服务没有成功注册该工具。智能体的工具缓存的还是旧列表。工具名称拼写不一致。排查步骤确认服务端启动时没有报错。使用mcp dev server.py查看当前工具列表。检查智能体配置文件中的服务地址是否正确。有些智能体运行时会缓存工具列表重启后再试。5.2 MCP 服务连接成功但调用超时可能原因工具内部有网络请求而超时时间设置过短。工具内部死循环或阻塞调用。服务端处理能力不足。解决思路将耗时操作改为异步实现。在工具内部设置合理的连接超时。将耗时长任务拆分为“提交任务 查询结果”两步。查看服务端日志确认卡在哪个环节。5.3 中文参数或返回值乱码如果工具返回的中文内容出现乱码通常和编码有关。排查方向确认终端编码为 UTF-8。在服务端脚本开头声明# -*- coding: utf-8 -*-。检查数据库连接字符串中是否包含正确字符集参数。需要注意有些智能体运行时的协议层默认使用 UTF-8只要服务端返回的 JSON 是 UTF-8 编码一般不会出现乱码。测试客户端打印result时如果乱码可以改成repr(result)查看原始结构。5.4 参数类型不匹配智能体传给工具的参数有时是字符串而工具函数期望数字。例如mcp.tool() def divide(a: float, b: float) - dict: ...实际调用时可能传入{a: 10, b: 3}解决方案有下面几种在函数内部做类型转换。提供更明确的参数 Schema说明参数类型。在工具入口做参数兼容处理。推荐在工具函数内部统一做一次类型校验和转换def safe_float(value): try: return float(value) except (TypeError, ValueError): return None5.5 常见问题速查表问题现象常见原因解决思路MCP 服务启动报 “ModuleNotFoundError”缺少依赖包激活虚拟环境后执行pip install -r requirements.txt工具注册失败装饰器写法不正确确认使用了mcp.tool()且函数在服务实例创建之后定义智能体无法发现工具挂载配置未刷新重启智能体运行时或清除工具缓存工具调用结果为空业务函数返回 None确保工具返回 dict 或可序列化对象异步工具报错Python 版本低于 3.8升级 Python 至 3.10Windows 下启动失败路径或编码问题检查 Python 解释器路径确认工作目录正确5.6 排查清单按顺序排查可以定位大多数问题[ ] MCP SDK 是否安装成功。[ ] 服务端能否独立启动。[ ] 客户端能否获取工具列表。[ ] 客户端能否直接调用工具。[ ] 智能体配置中的服务地址是否正确。[ ] 智能体工具缓存是否刷新。[ ] 服务端日志是否有异常堆栈。[ ] 工具参数是否与智能体传参匹配。6. 最佳实践与工程建议6.1 工具设计规范工具的设计直接影响智能体的调用成功率。建议遵循以下几点工具名称使用动词开头例如query_order、create_ticket、send_message让大模型更容易理解工具用途。工具描述要包含适用场景描述越具体大模型越容易在正确场景下调用。参数尽量少且语义明确不要设计大量可空参数。参数越多大模型越容易填错。返回值结构化统一使用success、data、error、code这样的字段结构。一个工具只做一件事便于测试和复用。6.2 错误处理策略工具层是智能体与外部系统的边界错误处理不能图省事。推荐做法在工具入口捕获所有异常避免异常直接抛到协议层。错误信息要区分用户可读信息和内部日志信息不要把敏感错误细节直接返回给用户。对数据库、网络等外部依赖设置超时。对于操作类工具要考虑数据写入失败时的回滚。6.3 安全边界与权限控制私有 MCP 工具暴露给智能体后相当于打开了一条新的调用通道。安全必须放在第一位。# 示例在工具内部实现简单的调用方校验 import os def check_caller_allowed(caller_token: str) - bool: expected_token os.getenv(MCP_CALLER_TOKEN, ) return caller_token expected_token更完整的做法是MCP 服务启动时使用环境变量管理密钥不硬编码在代码中。工具内部二次校验是否有权限调用。操作类工具记录操作人、操作时间、变更内容。服务部署在内网环境不直接暴露到公网。通过 HTTP 传输时务必启用 TLS配合网关鉴权。对调用频率做限制防止智能体在循环调用时打爆后端接口。6.4 日志与审计智能体调用的工具往往是业务敏感操作日志是排障和审计的基础。每次工具调用建议记录调用时间。工具名称。入参脱敏后。出参脱敏后。耗时。调用方来源。简单实现import logging import time logging.basicConfig(levellogging.INFO, format%(asctime)s %(levelname)s %(message)s) def log_tool_call(tool_name: str, params: dict, result: dict, elapsed_ms: float): logging.info( f[MCP_TOOL] {tool_name} params{params} fsuccess{result.get(success)} elapsed{elapsed_ms:.2f}ms )6.5 测试策略MCP 工具必须有测试不能只靠智能体手工验证。至少覆盖单元测试直接测试业务函数确保逻辑正确。调用测试通过 MCP 客户端调用工具验证协议层正常。边界测试空参数、缺少参数、超长字符串、非法值。故障测试数据库不可用、接口超时、服务重启后能否恢复。6.6 版本管理与部署requirements.txt或pyproject.toml固定 SDK 版本范围。服务代码使用 Git 管理每次修改都关联工具变更说明。部署时先在一台测试服务器上启动用测试客户端验证后再接入生产智能体。多个智能体共用服务时在服务层加租户隔离逻辑。7. 总结与学习路线这篇文章从 MCP 协议的核心概念出发完整演示了私有 Python 工具的编写、服务启动、客户端测试和智能体挂载流程。掌握的核心内容包括MCP 服务的目录结构和启动方式FastMCP装饰器的工具注册方法工具函数的输入输出规范与错误处理本地调试和智能体挂载的完整链路。接下来可以继续深入的方向包括学习 MCP 协议中 Resource 与 Prompt 两种能力类型补充工具之外的模型上下文能力。研究 MCP SDK 的底层消息格式理解协议握手和会话管理。尝试接入真实 MySQL 或 Redis让工具操作真实业务数据补齐事务和连接池管理经验。熟悉主流智能体运行时的 MCP 挂载配置和权限模型适配公司内部的平台。关注 MCP 官方版本迭代及时跟进接口变更和安全建议。在实际项目中接入私有 MCP 工具时优先关注三件事工具入参出参是否结构化、错误信息是否能被大模型理解、调用链路是否有审计日志。把这三件事做好开发和排障成本会大幅降低。如果这篇指南对你有帮助收藏备用也可以在评论区聊聊你在 MCP 开发中遇到的坑。