Python MCP服务器开发指南：为LLM构建标准化工具调用接口

1. 项目概述一个Python MCP服务器的诞生最近在折腾AI应用开发特别是想让大语言模型LLM能更“接地气”直接操作我本地或远程的工具和数据。这让我想到了一个概念模型上下文协议。简单来说它就像给AI装上了一套标准化的“手”和“眼睛”让它能安全、可控地调用外部功能。在GitHub上LinkupPlatform/python-mcp-server这个项目就是一个用Python实现的MCP服务器框架。这个项目是干什么的它不是一个具体的工具而是一个脚手架。如果你想让你的AI助手比如Claude Desktop、Cursor等能够查询你数据库里的数据、控制你的智能家居、或者执行一个你自定义的脚本你就需要一个MCP服务器来提供这些能力。python-mcp-server就是帮你快速搭建这个服务器的工具箱。它封装了MCP协议通信的底层细节让你可以专注于定义“你的AI助手能做什么”而不用去头疼怎么和AI客户端握手、传数据。对于开发者尤其是那些正在构建AI原生应用或希望深度集成AI能力到现有工作流的朋友这个项目价值很大。它降低了为LLM构建“工具调用”能力的门槛。你不用从零开始设计一套复杂的API和授权机制MCP协议提供了一套现成的标准而这个Python实现则让遵循标准变得异常简单。接下来我会拆解它的核心设计、手把手带你实现一个自己的服务器并分享我在集成和调试过程中踩过的那些坑。2. 核心设计思路与架构拆解2.1 为什么是MCP协议层的价值在深入代码之前我们必须先理解MCP要解决什么问题。LLM本身是“大脑”但它没有“肢体”。传统的做法是通过Function Calling函数调用向LLM描述一堆API然后LLM生成JSON去调用。但这有几个痛点首先API描述如OpenAI的tools格式和实际后端服务是紧耦合的每次变更都需要同步其次身份验证、资源发现、标准化输入输出都是开发者自己处理容易混乱。MCP协议的出现就是为了将“工具能力的描述”和“工具的执行”标准化、解耦。它定义了一套基于JSON-RPC的通信规范。一个MCP服务器启动后会向客户端如AI应用宣告“我这里有哪些工具Tools、有哪些只读的数据资源Resources”。客户端获取到这个清单后当用户提出需求时LLM就可以从清单中选择合适的工具或资源来使用。整个通信过程是双向、持续的支持流式响应。python-mcp-server的核心价值就是把这套协议的客户端Server部分用清晰、Pythonic的方式实现了。你不需要去手动拼接JSON-RPC消息处理传输层stdio或SSE只需要继承它提供的基类用装饰器声明你的工具函数框架就会自动处理剩下的所有事情。这种设计思路非常符合Python的哲学——让开发者关注业务逻辑而非底层协议。2.2 项目架构与核心模块这个项目的结构非常清晰主要围绕几个核心类展开Server类这是整个框架的入口和核心。它管理着所有已注册的工具、资源并维护着与客户端的连接。你需要实例化一个Server对象然后将你的工具函数注册上去。Transport层负责底层的消息传输。框架默认支持两种方式标准输入输出stdio和服务器发送事件SSE。Stdio模式最常见用于与本地AI桌面应用如Claude Desktop集成你的服务器作为一个独立的子进程启动通过管道通信。SSE模式则适用于远程服务器场景允许通过HTTP进行长连接通信。装饰器与注册机制这是框架最优雅的部分。通过server.tool()装饰器你可以将一个普通的Python函数“变成”一个MCP工具。装饰器会自动解析函数的类型注解Type Hints和文档字符串Docstring生成符合MCP标准的工具描述名称、描述、参数schema。这极大地减少了样板代码。类型与消息定义在mcp.types等模块中定义了所有MCP协议中涉及的数据结构如Tool、Resource、Argument等。这些使用了Pydantic模型确保了数据验证的严谨性也方便了IDE的自动补全。这种架构带来的好处是高内聚、低耦合。传输层的变化不会影响你的工具逻辑你工具逻辑的迭代也基本不需要改动通信代码。当你需要新增一个功能时95%的工作就是写一个新的Python函数然后用装饰器装饰它。3. 从零开始构建你的第一个MCP服务器理论讲得再多不如动手做一遍。我们来构建一个实用的服务器它提供两个功能1. 查询指定城市的实时天气模拟2. 对一段文本进行情感分析。3.1 环境准备与项目初始化首先确保你的Python环境是3.8或更高版本。创建一个新的项目目录并安装核心依赖。# 创建项目目录并进入 mkdir my-weather-sentiment-server cd my-weather-sentiment-server # 创建虚拟环境推荐 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate # 安装 python-mcp-server pip install mcp注意项目在PyPI上的包名就是mcp。直接pip install mcp即可安装LinkupPlatform/python-mcp-server这个库。这是最规范的方式避免从GitHub源码安装可能带来的版本管理问题。接下来创建我们的主文件server.py。3.2 核心工具函数实现在server.py中我们将实现两个工具。这里的关键是充分利用类型注解和文档字符串。# server.py import asyncio from typing import Any from mcp import ClientSession, StdioServerParameters from mcp.server import Server from mcp.server.models import InitializationOptions import mcp.server.stdio # 创建Server实例 server Server(weather-sentiment-server) # 工具1获取天气 server.tool() async def get_weather(city: str) - str: 获取指定城市的当前天气情况。 Args: city (str): 城市名称例如 北京, 上海, New York。 Returns: str: 该城市的模拟天气报告。 # 这里是一个模拟实现。真实场景下你会调用如OpenWeatherMap的API。 # 模拟根据城市返回不同的天气 weather_map { 北京: 晴15°C西北风2级, 上海: 多云18°C东南风1级, 纽约: 阴10°C东北风3级可能有小雨, } report weather_map.get(city, f未找到{city}的天气信息当前模拟为晴20°C微风。) # 模拟一点网络延迟让响应更真实 await asyncio.sleep(0.5) return f{city}的天气{report} # 工具2情感分析 server.tool() async def analyze_sentiment(text: str) - dict[str, Any]: 分析一段文本的情感倾向。 Args: text (str): 需要分析的文本内容。 Returns: dict: 包含情感分析结果的字典包括情感极性polarity和主观性subjectivity。 polarity范围在-1极度负面到1极度正面之间。 subjectivity范围在0非常客观到1非常主观之间。 # 这是一个极其简化的模拟分析。生产环境应使用NLP库如TextBlob, NLTK或调用AI服务。 positive_words [好, 开心, 优秀, 美丽, 感谢] negative_words [差, 伤心, 糟糕, 讨厌, 失望] text_lower text.lower() positive_count sum(word in text_lower for word in positive_words) negative_count sum(word in text_lower for word in negative_words) total positive_count negative_count if total 0: polarity 0.0 else: polarity (positive_count - negative_count) / total # 简单限制在[-1, 1]区间 polarity max(-1.0, min(1.0, polarity)) # 主观性模拟文本越长可能包含更多个人观点这里用一个简单启发式规则 subjectivity min(0.3 (len(text) / 500) * 0.5, 0.9) return { polarity: round(polarity, 2), subjectivity: round(subjectivity, 2), verdict: 正面 if polarity 0.05 else (负面 if polarity -0.05 else 中性) } # 主异步函数用于启动服务器 async def main(): 启动MCP服务器的主函数。 配置为使用stdio传输这是与Claude Desktop等客户端集成的标准方式。 # 创建stdio服务器参数这里使用默认配置 server_params StdioServerParameters( commandpython, # 解释器命令 args[server.py], # 脚本参数这里就是它自己 # 注意在真实集成中客户端如Claude Desktop会用自己的命令来启动这个脚本。 # 我们这里用同一个脚本做演示实际运行时客户端命令可能不同。 ) # 使用stdio运行服务器 async with mcp.server.stdio.stdio_server(server_params) as (read_stream, write_stream): async with ClientSession(read_stream, write_stream) as session: # 初始化会话告诉客户端我们服务器的能力 await session.initialize( InitializationOptions( server_nameserver.name, server_version0.1.0, capabilitiesserver.get_capabilities( # 可以在这里配置服务器支持哪些MCP特性 # 例如是否支持工具调用、资源列表等 tool_callingTrue, ) ) ) # 将我们server实例中注册的工具告知客户端 await session.list_tools() # 开始处理来自客户端的请求工具调用等 await server.run( session, # 可以传递自定义的请求处理器这里用默认的 ) if __name__ __main__: asyncio.run(main())这段代码已经是一个功能完整的MCP服务器了。server.tool()装饰器是魔法发生的地方。它会读取get_weather函数的签名——参数city: str和返回类型str以及函数下的文档字符串。然后自动生成一个MCP工具定义包括名称函数名get_weather、描述来自文档字符串的第一行、以及一个符合JSON Schema的参数定义。analyze_sentiment工具同理它的返回类型是dict框架也能很好地处理。实操心得文档字符串Docstring的质量至关重要。AI客户端如Claude会读取这个描述来理解工具用途。描述应简洁、准确参数说明要清晰。好的文档能让LLM更准确地决定何时、如何调用你的工具。3.3 运行与基础测试由于我们的服务器设计为通过stdio与客户端通信直接运行python server.py会启动服务器并等待来自标准输入的请求但因为没有客户端连接它看起来会“挂起”。为了测试我们需要模拟一个客户端。最简单的方法是使用框架自带的测试工具或编写一个简单的测试脚本。不过更实用的方法是使用mcpCLI工具安装mcp包后自带。它可以作为一个通用的MCP客户端来测试服务器。首先我们需要一个简单的服务器启动配置。创建一个server-config.json{ mcpServers: { my-local-server: { command: python, args: [/绝对路径/to/your/project/server.py], env: { PYTHONPATH: /绝对路径/to/your/project } } } }然后可以通过CLI来列出工具进行测试但这需要更复杂的设置。对于快速验证我们可以写一个极简的模拟测试脚本test_server.py# test_server.py - 一个非常基础的模拟测试 import asyncio import json import subprocess import sys async def test_tool_call(): # 启动服务器进程建立管道 proc await asyncio.create_subprocess_exec( sys.executable, server.py, stdinasyncio.subprocess.PIPE, stdoutasyncio.subprocess.PIPE, stderrasyncio.subprocess.PIPE ) # 模拟一个简单的“初始化”和“工具调用”请求序列 # 注意这是一个高度简化的演示真实MCP协议消息更复杂。 init_request { jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 0.1.0, capabilities: {}, clientInfo: {name: test-client} } } # 发送初始化请求 proc.stdin.write((json.dumps(init_request) \n).encode()) await proc.stdin.drain() # ... 这里省略了读取响应、处理等复杂步骤 # 实际上你需要完整实现一个简单的MCP客户端来解析服务器输出。 print(测试启动完成。在实际集成中请使用Claude Desktop等客户端进行连接。) # 清理进程 proc.terminate() await proc.wait() if __name__ __main__: asyncio.run(test_tool_call())这个测试脚本并不完整因为它需要完整解析双向的JSON-RPC消息流。它只是为了展示服务器进程是如何启动的。真正的集成测试强烈建议直接与目标客户端如Claude Desktop进行。4. 高级功能与生产级考量一个基础的玩具服务器很容易但要投入实际使用还需要考虑更多。4.1 资源Resources的声明与使用除了工具ToolsMCP另一个核心概念是资源Resources。工具代表“可执行的操作”而资源代表“可读取的数据”。例如你的服务器可以提供“当前待办事项列表”作为一个资源AI客户端可以读取它但无法通过工具修改它除非你另外提供update_todo工具。在python-mcp-server中你可以使用server.resource()装饰器来声明一个资源并通过实现read_resource等方法来处理资源读取请求。资源通常用URI来标识如file:///path/to/file或自定义的myapp://todos。# 在server.py中添加资源声明 server.resource(todo://list) async def get_todo_list() - str: 获取当前的待办事项列表只读资源。 # 这里可以从数据库或文件读取 todos [完成MCP服务器文档, 测试天气工具, 购买 groceries] return \n.join(f- {item} for item in todos)声明资源后支持资源发现的客户端就可以在上下文中加载这些只读数据为LLM提供更丰富的背景信息。4.2 错误处理与健壮性在生产环境中你的工具函数可能会失败网络超时、API限流、无效输入等。MCP协议要求服务器对错误进行恰当的响应。框架会自动捕获工具函数抛出的异常并将其转换为MCP错误响应。但你可以做得更好输入验证虽然类型注解提供基础验证但更复杂的业务规则如城市名白名单应在函数内部进行。server.tool() async def get_weather_v2(city: str) - str: allowed_cities [北京, 上海, 广州, 深圳, 纽约] if city not in allowed_cities: # 抛出一个清晰的异常框架会将其转化为客户端可理解的错误 raise ValueError(f暂不支持城市{city}。支持的城市有{, .join(allowed_cities)}) # ... 原有逻辑优雅降级与重试对于依赖外部API的工具实现重试逻辑和缓存。import aiohttp from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) async def fetch_external_weather(city: str): async with aiohttp.ClientSession() as session: async with session.get(fhttps://api.weatherapi.com/v1/...?keyYOUR_KEYq{city}) as resp: resp.raise_for_status() return await resp.json() server.tool() async def get_weather_pro(city: str) - str: try: data await fetch_external_weather(city) return f{city}: {data[current][temp_c]}°C, {data[current][condition][text]} except aiohttp.ClientError as e: # 网络错误返回降级信息 return f无法获取实时天气请检查网络。最后已知信息{city}天气通常良好。超时控制使用asyncio.wait_for为工具执行设置超时防止一个长时间运行的工具阻塞整个服务器。server.tool() async def long_running_task(query: str): try: result await asyncio.wait_for(_internal_long_task(query), timeout30.0) return result except asyncio.TimeoutError: return 任务执行超时请简化查询或稍后重试。4.3 安全与权限管理MCP服务器可能访问敏感数据或执行重要操作安全至关重要。传输安全如果使用SSE传输务必通过HTTPSWSS运行。对于stdio模式本地集成确保启动服务器的进程是可信的。工具级权限不是所有工具都应对所有用户开放。你可以在服务器层面维护一个简单的权限映射或者在工具函数开头进行检查。# 假设有一个简单的用户上下文在实际中可能来自客户端初始化参数 class UserContext: def __init__(self, user_id, role): self.user_id user_id self.role role # 在工具中检查 server.tool() async def delete_database(table: str, context: UserContext): if context.role ! admin: raise PermissionError(只有管理员可以执行此操作。) # ... 删除逻辑注意MCP协议本身不规定身份验证和授权机制这需要你根据集成环境如Claude for Teams的企业SSO来设计和实现。一种常见模式是客户端在初始化请求中携带一个令牌服务器验证该令牌并建立用户上下文。输入净化永远不要信任来自客户端的输入。即使LLM已经做了筛选也要防止潜在的注入攻击特别是当你的工具会执行系统命令或构造数据库查询时。5. 与主流客户端集成实战让服务器跑起来只是第一步让它真正发挥作用需要与AI客户端集成。目前最流行的MCP客户端是Anthropic Claude Desktop和Cursor IDE。5.1 集成到 Claude DesktopClaude Desktop 允许通过配置文件添加自定义MCP服务器。这是最无缝的集成体验。找到配置文件位置macOS:~/Library/Application Support/Claude/claude_desktop_config.jsonWindows:%APPDATA%\Claude\claude_desktop_config.jsonLinux:~/.config/Claude/claude_desktop_config.json编辑配置文件如果文件不存在就创建它。添加你的服务器配置。{ mcpServers: { my-weather-server: { command: /绝对路径/to/your/venv/bin/python, args: [/绝对路径/to/your/project/server.py], env: { PYTHONUNBUFFERED: 1 } } } }关键点command: 必须使用虚拟环境中Python解释器的绝对路径。这是最常见的错误来源。使用which python在激活的虚拟环境中来获取。args: 你的服务器脚本的绝对路径。env: 设置PYTHONUNBUFFERED1确保日志实时输出便于调试。重启Claude Desktop保存配置文件后完全退出并重启Claude Desktop应用。验证集成重启后在Claude的聊天界面你应该能看到一个微小的变化可能是一个新的图标或提示。你可以直接问Claude“你现在有哪些工具可以用”或者“用get_weather工具查一下北京的天气”。Claude会自动发现服务器注册的工具并调用它们。5.2 集成到 Cursor IDECursor 也支持MCP服务器配置方式类似但配置文件路径不同。配置文件路径通常位于用户配置目录下如~/.cursor/mcp.jsonmacOS/Linux或%APPDATA%\Cursor\mcp.jsonWindows。配置内容格式与Claude Desktop类似。{ mcpServers: { my-weather-server: { command: /path/to/venv/bin/python, args: [/path/to/project/server.py] } } }重启Cursor修改配置后需要重启Cursor IDE。5.3 调试与日志记录集成过程很少一帆风顺。强大的日志是调试的利器。在服务器中启用日志import logging logging.basicConfig(levellogging.DEBUG, format%(asctime)s - %(name)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) server.tool() async def get_weather(city: str): logger.info(f收到天气查询请求城市: {city}) # ... 业务逻辑 logger.debug(f天气查询完成结果: {result}) return result查看客户端日志Claude Desktop: 日志文件通常位于上述配置文件的同级目录如~/Library/Logs/Claude/macOS。查看最新的日志文件搜索mcp、server或你的服务器名。通用方法将服务器的标准错误输出重定向到文件在配置中修改args或者直接在代码中配置日志输出到文件。常见集成问题排查表问题现象可能原因排查步骤客户端完全没发现工具配置文件路径错误、命令错误、服务器启动失败1. 检查配置文件路径和格式。2. 手动在终端运行配置中的命令看服务器能否正常启动不报错并等待输入。3. 检查客户端日志看是否有加载MCP配置的记录或错误信息。服务器启动后立即退出Python路径错误、脚本语法错误、依赖缺失1. 确保command中的Python路径绝对正确且虚拟环境已安装所有依赖mcp。2. 手动运行python server.py看是否有导入错误或语法错误。客户端发现工具但调用失败工具函数内部异常、协议消息格式错误1. 查看服务器日志如果已重定向。2. 在工具函数内部添加更详细的try-catch和日志定位异常点。3. 检查工具函数的输入输出类型是否可JSON序列化。工具调用超时或无响应工具函数执行时间过长、死循环1. 在工具函数内添加超时控制。2. 优化工具函数逻辑避免长时间阻塞操作。6. 性能优化与扩展方向当你的工具越来越多或者单个工具负载较重时就需要考虑性能。异步Async是核心python-mcp-server完全基于asyncio。确保你的所有工具函数都是async def并且在执行I/O操作网络请求、数据库查询、文件读写时使用异步库如aiohttp,asyncpg,aiofiles。绝对避免在工具函数内部使用同步的、会阻塞事件循环的调用如time.sleep,requests.get而不在线程池中运行。连接池与缓存对于频繁访问的外部服务如数据库、第三方API在服务器生命周期内维护连接池和缓存。from aiohttp import ClientSession from aiocache import Cache # 在Server实例化后创建全局会话和缓存需安装aiocache # server Server(...) # http_session ClientSession() # cache Cache(Cache.MEMORY) # 使用内存缓存 server.tool() async def get_weather_cached(city: str): cache_key fweather:{city} cached await cache.get(cache_key) if cached: return f缓存{cached} # ... 调用真实API await cache.set(cache_key, result, ttl300) # 缓存5分钟 return result水平扩展单个MCP服务器进程有性能上限。对于高并发场景可以考虑多进程利用uvicornasgi运行多个服务器实例如果使用SSE传输。负载均衡在多个服务器实例前放置一个负载均衡器客户端连接不同的实例。这需要更复杂的服务发现和状态管理如果工具有状态。扩展方向python-mcp-server是一个优秀的起点你可以基于它构建更复杂的系统动态工具注册根据配置或数据库内容在运行时动态添加或移除工具。工具市场构建一个中心化的服务器可以加载和管理多个第三方开发的“工具插件”。与现有后端集成将MCP服务器作为现有Web应用的一个适配层让LLM能够安全地调用你已有的业务API。构建一个稳定、高效的MCP服务器其核心在于理解协议的解耦思想利用好Python异步生态并始终将安全性、可观测性日志、监控放在重要位置。从一个小工具开始逐步迭代你会发现它为你的AI应用带来的能力提升是巨大的。