基于classmcp构建AI本地工具：Python类封装与MCP协议实践

1. 项目概述与核心价值最近在折腾AI应用开发特别是想让大语言模型LLM能更“主动”地操作电脑上的各种软件比如打开浏览器查资料、用Excel处理数据或者控制音乐播放器。这听起来像是科幻电影里的场景但在实际开发中我遇到了一个核心难题如何让AI安全、可控地调用本地应用程序的API直接给AI系统权限去执行任意命令风险太高而针对每个软件都写一套复杂的适配代码工作量又大得惊人。正是在这个背景下我深入研究了GitHub上一个名为classmcp的项目。这个项目由TheDecipherist开源它提供了一种非常巧妙的思路利用Python的类Class来定义和管理AI可调用的工具Tools。简单来说它不是一个庞大的框架而是一个轻量级的“粘合剂”和“规范制定者”。它帮你把你想让AI执行的操作比如“打开文件”、“发送邮件”封装成一个个结构清晰、自带描述的Python类方法。然后它遵循MCPModel Context Protocol协议将这些工具暴露给支持该协议的AI客户端例如Claude Desktop、Cursor等从而实现AI对本地功能的精细调用。对我而言classmcp的核心价值在于**“降低开发门槛”和“提升安全性”**。它通过约定大于配置的方式让开发者能用最熟悉的Python面向对象思维来构建AI工具无需深入理解复杂的协议细节。同时由于工具的执行逻辑完全由你编写的类方法控制你可以在其中加入权限检查、输入验证、操作日志等安全措施避免了AI的“野蛮操作”。接下来我将详细拆解如何使用classmcp从设计思路到实操落地分享我趟过的一些坑和总结的经验。2. 核心设计思路与架构解析2.1 什么是MCPModel Context Protocol在深入classmcp之前必须先理解它背后的协议——MCP。你可以把MCP想象成AI世界里的“USB协议”。在没有统一协议之前每个AI应用客户端想连接本地资源服务器都需要自己定制驱动混乱且低效。MCP的出现就是为了标准化AI客户端与资源服务器之间的通信方式。MCP定义了一套简单的JSON-RPC标准规定了工具Tools服务器可以提供哪些可调用的函数包括函数名、描述、参数格式。资源Resources服务器可以提供哪些可读的数据源如文件、数据库内容。提示Prompts服务器可以预定义一些提示词模板。classmcp扮演的角色就是一个MCP服务器的快速构建器。它让你不用从零开始实现JSON-RPC通信、工具发现等底层逻辑而是专注于用Python类来定义“工具”本身。2.2classmcp的巧妙之处以类为中心的工具封装传统的MCP服务器开发你可能需要分别定义工具schema、工具处理函数然后再进行注册代码会比较分散。classmcp的创新在于它利用了Python类的天然结构来组织工具。一个类对应一个逻辑模块例如文件操作、网络搜索类中的每一个async def方法在满足特定条件后就会自动被注册为一个MCP工具。这种方法的好处显而易见高内聚相关的工具都放在同一个类里管理起来非常方便。比如FileManager类里放read_file,write_file,list_directory等方法。自描述性通过Python的docstring方法文档字符串和类型注解就能自动生成工具的详细描述和参数格式减少了额外编写元数据的工作。易于扩展要增加新工具只需在对应的类里添加一个新方法。要增加一个新功能模块就新建一个类。2.3 项目架构与工作流程classmcp的架构非常清晰主要包含以下几个核心部分装饰器 (mcp_tool)这是核心魔法所在。你只需要在类的方法上添加这个装饰器classmcp就会在后台自动收集该方法的信息名称、描述、参数并将其包装成一个符合MCP标准的工具。服务器类 (MCPServer)这是服务器的运行时容器。你创建MCPServer实例然后将你定义的工具类注册进去。它负责启动一个遵循MCP协议的服务器监听来自AI客户端的请求。工具执行引擎当AI客户端通过MCP协议调用某个工具时MCPServer会接收到请求找到对应的类和方法传入参数并执行最后将结果返回给客户端。整个工作流程可以概括为定义工具类 - 注册到服务器 - 启动服务器 - AI客户端连接并调用。这种设计使得开发者的心智负担大大降低可以将精力完全集中在工具的业务逻辑实现上。3. 从零开始构建你的第一个MCP工具3.1 环境准备与安装首先确保你的Python环境在3.8以上。创建一个新的虚拟环境是一个好习惯可以避免包依赖冲突。# 创建并激活虚拟环境以venv为例 python -m venv .venv source .venv/bin/activate # Linux/macOS # .venv\Scripts\activate # Windows # 安装 classmcp pip install classmcpclassmcp的依赖非常干净主要是用于HTTP通信的httpx和用于进程通信的pydantic安装很快。3.2 定义你的第一个工具类我们来创建一个最简单的工具一个计算器。新建一个文件比如my_tools.py。from classmcp import mcp_tool from pydantic import BaseModel # 定义输入参数的模型可选但强烈推荐 class AddNumbersInput(BaseModel): a: float b: float # 定义工具类 class CalculatorTools: # 使用 mcp_tool 装饰器标记这是一个MCP工具 mcp_tool async def add_numbers(self, input: AddNumbersInput) - float: 将两个数字相加。 Args: input: 包含两个数字a和b的输入对象。 Returns: 两个数字的和。 return input.a input.b mcp_tool async def multiply_numbers(self, a: float, b: float) - float: 将两个数字相乘。 Args: a: 第一个乘数。 b: 第二个乘数。 Returns: 两个数字的乘积。 return a * b关键点解析mcp_tool装饰器这是关键。它告诉classmcp这个方法需要被暴露为AI可调用的工具。异步方法工具方法必须是async def。这是因为MCP通信通常是异步I/O操作使用异步可以提高服务器并发处理能力。参数与类型注解classmcp会利用类型注解来生成工具的参数schema。你可以像add_numbers方法一样使用Pydantic模型来定义复杂的结构化输入这会让工具的描述更清晰也可以像multiply_numbers一样直接使用基本类型。Docstring文档方法的文档字符串会被自动用作工具的“描述”。AI客户端如Claude在决定是否调用该工具时会参考这个描述。因此清晰、准确的描述至关重要。3.3 创建并启动MCP服务器接下来我们创建一个主程序文件main.py来启动服务器。import asyncio from classmcp import MCPServer from my_tools import CalculatorTools # 导入刚才定义的工具类 async def main(): # 1. 创建MCP服务器实例 server MCPServer() # 2. 注册工具类。可以注册多个类。 server.register_tool_class(CalculatorTools()) # 3. 启动服务器。 # 这里使用stdio传输这是与Claude Desktop等客户端通信的常用方式。 # 客户端会启动这个Python脚本并通过标准输入输出进行通信。 await server.run(transportstdio) if __name__ __main__: asyncio.run(main())3.4 配置AI客户端进行连接服务器准备好了还需要一个支持MCP的AI客户端来调用它。这里以Claude Desktop为例找到Claude Desktop的配置文件夹。macOS:~/Library/Application Support/Claude/claude_desktop_config.jsonWindows:%APPDATA%\Claude\claude_desktop_config.json编辑或创建claude_desktop_config.json文件添加你的MCP服务器配置{ mcpServers: { my-calculator: { command: /path/to/your/.venv/bin/python, args: [/absolute/path/to/your/main.py] } } }重要提示command必须是你虚拟环境中Python解释器的绝对路径。args是你的主程序main.py的绝对路径。配置完成后重启Claude Desktop。重启后你可以在Claude的输入框里尝试说“请调用计算器工具帮我计算123加456。” Claude会识别到可用的add_numbers工具并请求你提供参数a和b最终返回结果。4. 进阶实践构建实用的文件管理器工具单一的计算器工具演示了基础流程但classmcp的真正威力在于构建复杂的本地交互工具。下面我们构建一个更实用的、具备基础安全性的文件管理器。4.1 设计带安全边界的工具类我们不想让AI能随意删除系统文件或访问敏感目录。因此在设计时就要加入“工作目录”的概念和路径安全检查。import os import shutil from pathlib import Path from typing import List from classmcp import mcp_tool from pydantic import BaseModel, Field class FileManagerInput(BaseModel): # 使用Field来提供更详细的参数描述 filepath: str Field(..., description相对于工作目录的文件或目录路径) class FileManager: def __init__(self, workspace_root: str): # 初始化时设定一个安全的工作空间根目录 self.workspace_root Path(workspace_root).resolve() # 确保工作目录存在 self.workspace_root.mkdir(parentsTrue, exist_okTrue) print(f文件管理器工作空间已锁定在: {self.workspace_root}) def _safe_path(self, user_path: str) - Path: 将用户提供的相对路径解析为绝对路径并确保其位于工作空间内防止路径遍历攻击。 requested_path (self.workspace_root / user_path).resolve() # 检查请求的路径是否在工作空间根目录之下 try: requested_path.relative_to(self.workspace_root) except ValueError: raise PermissionError(f访问路径 {user_path} 被拒绝它超出了允许的工作空间范围。) return requested_path mcp_tool async def list_files(self, directory: str .) - List[str]: 列出指定目录下的文件和子目录。 Args: directory: 要列出的目录路径默认为当前工作目录。 Returns: 文件名和目录名的列表。 target_dir self._safe_path(directory) if not target_dir.is_dir(): return [f错误路径 {directory} 不是一个目录。] items [] for item in target_dir.iterdir(): items.append(f[DIR] {item.name} if item.is_dir() else item.name) return items mcp_tool async def read_file(self, input: FileManagerInput) - str: 读取文本文件的内容。 Args: input: 包含文件路径的对象。 Returns: 文件的内容。如果文件不存在或读取失败返回错误信息。 file_path self._safe_path(input.filepath) try: return file_path.read_text(encodingutf-8) except FileNotFoundError: return f错误文件 {input.filepath} 未找到。 except UnicodeDecodeError: return f错误无法以UTF-8编码读取文件 {input.filepath}它可能是一个二进制文件。 mcp_tool async def write_file(self, filepath: str, content: str) - str: 将内容写入文件。如果文件已存在将被覆盖如果不存在将被创建。 Args: filepath: 要写入的文件路径。 content: 要写入的文本内容。 Returns: 操作结果信息。 file_path self._safe_path(filepath) try: # 确保父目录存在 file_path.parent.mkdir(parentsTrue, exist_okTrue) file_path.write_text(content, encodingutf-8) return f成功内容已写入 {filepath}。 except Exception as e: return f错误写入文件 {filepath} 时失败 - {str(e)} mcp_tool async def delete_file(self, input: FileManagerInput) - str: 删除一个文件。这是一个危险操作 Args: input: 包含要删除的文件路径的对象。 Returns: 操作结果信息。 target_path self._safe_path(input.filepath) if not target_path.exists(): return f错误路径 {input.filepath} 不存在。 if target_path.is_dir(): return f错误路径 {input.filepath} 是一个目录请使用删除目录工具。 try: target_path.unlink() return f成功文件 {input.filepath} 已被删除。 except Exception as e: return f错误删除文件 {input.filepath} 时失败 - {str(e)}4.2 在主服务器中集成并初始化修改你的main.py集成这个更强大的工具。import asyncio from classmcp import MCPServer from my_tools import CalculatorTools from file_manager import FileManager # 假设文件管理器保存在file_manager.py async def main(): server MCPServer() # 注册计算器工具 server.register_tool_class(CalculatorTools()) # 初始化文件管理器并指定一个安全的工作空间例如用户目录下的一个特定文件夹 import os workspace os.path.expanduser(~/ai_workspace) # 例如/Users/yourname/ai_workspace server.register_tool_class(FileManager(workspace_rootworkspace)) await server.run(transportstdio) if __name__ __main__: asyncio.run(main())4.3 实操心得与安全强化建议在实现上述文件管理器的过程中我总结了几点关键经验路径安全是重中之重_safe_path方法是整个工具类的安全基石。它使用pathlib.Path.resolve()来解析绝对路径并使用relative_to()来确保目标路径不会通过../../../这样的方式逃逸出工作空间。永远不要相信AI客户端直接传来的路径。详细的错误反馈工具返回的错误信息应当对AI友好这样AI才能理解哪里出了问题并可能尝试纠正或提示用户。例如返回“不是一个目录”比返回一个Python的NotADirectoryError异常堆栈更有用。操作确认机制对于delete_file这类危险操作理想的实现应该包含一个“确认”步骤。虽然MCP协议本身是单次请求-响应但我们可以通过设计让工具返回一个需要确认的提示或者在实际项目中结合客户端的UI能力来实现二次确认。资源清理如果工具会创建临时文件或网络连接记得在类中实现清理逻辑或者使用上下文管理器避免资源泄漏。5. 高级技巧与性能优化5.1 工具依赖注入与共享状态有时多个工具类需要共享一些状态或服务比如数据库连接池、配置信息、或一个共享的HTTP客户端。classmcp允许你在注册工具类时传递初始化参数也支持更复杂的依赖注入模式。from classmcp import MCPServer import aiohttp class WebSearchTools: def __init__(self, http_session: aiohttp.ClientSession, api_key: str): self.session http_session self.api_key api_key mcp_tool async def search_web(self, query: str): # 使用共享的session进行网络请求效率更高 async with self.session.get(fhttps://api.example.com/search?q{query}key{self.api_key}) as resp: return await resp.json() async def main(): server MCPServer() # 创建共享的aiohttp会话 async with aiohttp.ClientSession() as session: # 将会话和API密钥注入到工具类中 search_tools WebSearchTools(session, api_keyyour_api_key_here) server.register_tool_class(search_tools) # 注意这里需要确保session在服务器运行期间保持有效 # 一种方法是将server.run放在with块内或者使用生命周期钩子管理session await server.run(transportstdio)5.2 异步工具与长时间运行任务如果你的工具需要执行长时间运行的任务如训练模型、处理大文件直接同步执行会阻塞整个服务器影响其他工具的响应。正确的做法是使用asyncio的机制来避免阻塞。import asyncio from classmcp import mcp_tool class LongRunningTools: mcp_tool async def process_large_data(self, data_id: str) - str: 模拟一个长时间的数据处理任务。 Args: data_id: 数据ID。 Returns: 处理结果。 # 立即返回一个任务已接受的消息 # 在实际场景中你可能需要将任务ID存入队列或数据库 task_id ftask_{data_id} # 在后台异步执行耗时任务不阻塞当前函数返回 asyncio.create_task(self._real_processing(task_id, data_id)) return f数据处理任务 {task_id} 已开始在后台执行。完成后请查询结果。 async def _real_processing(self, task_id: str, data_id: str): 实际执行耗时任务的内部方法。 await asyncio.sleep(10) # 模拟10秒的耗时操作 # ... 实际处理逻辑 ... print(f任务 {task_id} 处理完成) # 这里可以将结果存储起来供另一个查询结果的工具读取5.3 工具的动态注册与卸载在某些高级应用场景你可能需要在服务器运行时动态地添加或移除工具。classmcp的MCPServer实例提供了相应的方法。async def main(): server MCPServer() # ... 初始注册一些工具 ... # 假设某个条件触发后需要动态添加一个新工具类 def on_condition_met(): dynamic_tool_instance SomeDynamicTool() server.register_tool_class(dynamic_tool_instance) # 注意动态注册后需要通知客户端刷新工具列表这取决于客户端实现 # MCP协议支持工具列表变更通知但需要客户端配合。 # 动态卸载工具相对复杂因为需要明确工具标识。 # 更常见的模式是通过工具类内部的逻辑开关来“禁用”某个工具而非从服务器移除。6. 调试、问题排查与最佳实践6.1 常见问题与解决方案在开发和集成classmcp工具时我遇到了以下几个典型问题问题现象可能原因排查步骤与解决方案Claude Desktop 无法连接/找不到工具1. 配置文件路径错误。2. Python或脚本路径不是绝对路径。3. 虚拟环境未激活或依赖未安装。4. 服务器脚本启动即报错退出。1.检查配置文件确保claude_desktop_config.json格式正确路径无误。2.使用绝对路径在command和args中务必使用绝对路径。3.查看日志在终端手动运行配置中的命令如/path/to/python /path/to/main.py查看是否有导入错误或运行时错误。4.简化测试先创建一个最简单的“echo”工具确保基础通信正常。工具调用失败返回参数错误1. AI客户端传递的参数格式与工具定义不匹配。2. Pydantic模型验证失败。1.检查工具定义确认mcp_tool装饰的方法参数名和类型是否清晰。使用PydanticBaseModel可以极大增强健壮性。2.查看服务器日志在main.py中增加日志记录打印收到的请求看参数是否正确传递。工具执行超时或无响应1. 工具方法是同步的执行了阻塞操作如time.sleep。2. 工具内部有死循环或长时间计算。1.确保工具异步所有mcp_tool装饰的方法必须是async def。2.避免阻塞将耗时的I/O操作文件读写、网络请求使用异步库aiofiles,aiohttp。3.设置超时在客户端或工具逻辑内部考虑增加超时机制。工具描述在客户端显示不全或混乱工具方法的docstring格式不符合预期。规范Docstring使用标准的Python文档字符串格式。第一行写简要概述Args:部分列出参数Returns:部分说明返回值。保持简洁明了。6.2 调试技巧启用日志在服务器启动前配置Python的logging模块将日志输出到文件或控制台这对于追踪问题至关重要。import logging logging.basicConfig(levellogging.DEBUG, format%(asctime)s - %(name)s - %(levelname)s - %(message)s)使用Stdio调试在开发时可以暂时修改main.py不使用server.run(transportstdio)而是自己模拟一个简单的输入输出循环来测试工具逻辑排除协议通信问题。分步验证先确保工具类的方法在纯Python环境下能正常工作再集成到MCP服务器中测试。6.3 最佳实践总结经过多个项目的实践我总结了以下使用classmcp的最佳实践单一职责每个工具类应专注于一个特定的领域如文件操作、网络请求、数据查询。这使代码更清晰也便于维护。防御性编程对所有外部输入来自AI的参数进行严格的验证和清理特别是涉及文件系统、网络访问或系统命令的操作。详尽文档为每个工具方法编写清晰、完整的docstring。这不仅是给AI看的也是给未来维护代码的你自己看的。错误处理工具方法内部应使用try...except捕获可能出现的异常并返回对AI友好的错误信息而不是抛出未处理的异常导致服务器崩溃。性能考量对于可能耗时的操作设计成异步非阻塞模式或者提供任务状态查询接口。配置化将工作空间路径、API密钥等可变参数通过配置文件或环境变量管理而不是硬编码在代码中。classmcp这个项目本质上提供了一种优雅的“思维转换”方式。它将构建AI智能体周边工具链的复杂性封装成了熟悉的Python类开发体验。当你不再需要纠结于协议细节和通信底层时就能更专注于思考“我希望我的AI助手具备什么样的能力”然后像编写普通工具函数一样将这些能力实现出来。从简单的文本处理到复杂的系统交互边界只取决于你的想象力与对安全性的把控。