1. 项目概述MCP Maker一个为AI智能体打造通用工具链的框架最近在折腾AI智能体Agent开发的朋友估计都听过MCPModel Context Protocol这个概念。简单来说MCP是Anthropic提出的一套协议旨在让AI模型比如Claude能够安全、标准化地调用外部工具、访问数据和执行操作。你可以把它想象成给AI模型装上了一双“手”和“眼睛”让它不再局限于文本对话而是能真正操作你电脑里的文件、查询数据库、控制智能家居甚至执行复杂的业务流程。而我今天要聊的MrAliHasan/mcp-maker就是一个专门用来快速创建、管理和部署MCP服务器的开源框架。如果你觉得从头手写一个符合MCP协议的服务器太麻烦或者想把自己的一些脚本、API快速“包装”成AI能用的工具那么这个项目就是你需要的“脚手架”和“工具箱”。它抽象了协议通信、生命周期管理、工具注册等底层细节让开发者能聚焦在核心工具逻辑的实现上。我自己在尝试将内部几个数据查询脚本AI化时就用了这个框架效率提升非常明显从协议学习到第一个工具上线只用了不到半天时间。2. 核心设计思路为什么我们需要一个MCP框架在深入代码之前我们先聊聊为什么直接手搓MCP服务器可能不是最佳选择。MCP协议本身基于JSON-RPC over stdio/SSE这意味着你需要处理进程间通信、消息序列化、错误处理、资源清理等一系列繁琐但至关重要的问题。一个健壮的MCP服务器还需要考虑工具的动态注册、参数验证、权限控制以及日志监控。mcp-maker框架的核心设计思路正是为了解决这些工程化痛点。它采用了类似现代Web框架如Express、FastAPI的设计哲学通过提供高层次抽象和约定大于配置的原则降低开发门槛。2.1 核心架构拆解框架的架构可以清晰地分为三层协议通信层这一层完全由框架封装。它负责启动子进程、建立stdio通信管道、解析和封装符合MCP协议规范的JSON-RPC消息。作为开发者你几乎不需要关心消息是如何被发送和接收的就像使用Web框架时不需要手动解析HTTP报文一样。工具管理层这是框架的核心。它提供了一个装饰器如tool和一套类如ToolRegistry让你能以极其简洁的方式声明一个工具。你只需要定义工具的名称、描述、参数模式遵循JSON Schema以及具体的执行函数。框架会自动处理工具的注册、列表返回并在收到调用请求时将参数反序列化后传递给你的函数。生命周期与资源层框架提供了清晰的初始化initialize和清理shutdown钩子。这对于需要连接数据库、加载大模型、打开文件等操作的场景至关重要。框架确保这些资源在服务器启动时正确初始化并在退出时无论是正常退出还是崩溃被妥善清理避免资源泄漏。2.2 与直接实现协议的对比为了更直观地理解框架的价值我们对比一下两种方式事项直接实现MCP协议使用mcp-maker框架协议处理需手动实现JSON-RPC over stdio的完整读写、解析、路由逻辑。框架内置开发者无需关心。工具注册需维护一个工具列表手动实现tools/list和tools/call的响应。使用装饰器或API自动注册自动生成列表和调用路由。错误处理需在多个层级协议、工具逻辑捕获异常并转换为MCP标准错误响应。框架提供统一的错误处理中间件可将异常自动包装。资源管理需自行确保在initialize和shutdown信号中正确初始化和释放资源。提供明确的钩子函数资源管理逻辑更集中、安全。开发速度慢需要深入理解协议细节。快聚焦业务逻辑快速原型开发。代码维护性协议逻辑与业务逻辑耦合代码冗长不易维护。关注点分离业务逻辑清晰代码简洁。实操心得在评估是否使用框架时一个简单的判断标准是如果你的工具数量超过3个或者工具涉及外部资源网络、数据库、文件那么使用框架带来的可维护性和健壮性提升将远远超过其微小的学习成本。mcp-maker的API设计非常直观看过一两个例子就能上手。3. 从零开始使用 mcp-maker 创建你的第一个MCP服务器理论说得再多不如动手试一下。接下来我将带你一步步创建一个具有实际功能的MCP服务器一个本地文件内容搜索工具。这个工具允许AI智能体在你指定的目录中根据关键词搜索文件内容。3.1 环境准备与项目初始化首先确保你的开发环境已安装Python 3.8。然后通过pip安装mcp-maker框架。目前它可能尚未发布到PyPI所以我们需要从GitHub仓库克隆并安装。# 克隆仓库 git clone https://github.com/MrAliHasan/mcp-maker.git cd mcp-maker # 使用pip从本地安装开发模式便于修改 pip install -e . # 或者如果你只是想使用稳定版且作者已发布到PyPI则可以直接 # pip install mcp-maker安装完成后我们创建一个新的项目目录。mkdir my_file_search_server cd my_file_search_server3.2 定义核心工具文件内容搜索在项目根目录下创建主文件server.py。我们将在这里定义我们的工具。# server.py import os from pathlib import Path from typing import List, Optional import mcp_maker as mcp from mcp_maker.types import ToolResult # 初始化一个MCP服务器应用 app mcp.MCPMaker() # 使用装饰器定义我们的第一个工具search_files app.tool( namesearch_files_in_directory, description在指定的目录中递归搜索包含特定关键词的文本文件。, # 定义输入参数的模式遵循JSON Schema input_schema{ type: object, properties: { directory_path: { type: string, description: 要搜索的根目录的绝对路径。 }, keyword: { type: string, description: 要搜索的关键词。 }, file_extension: { type: string, description: 可选限制搜索特定扩展名的文件例如 .txt, .py。默认为 .txt。, default: .txt } }, required: [directory_path, keyword] # 必填参数 } ) async def search_files( directory_path: str, keyword: str, file_extension: str .txt ) - ToolResult: 工具的具体实现函数。 参数名必须与 input_schema 中定义的 properties 键名一致。 results [] path Path(directory_path) # 1. 参数验证与安全边界检查非常重要 if not path.exists(): return ToolResult( content[{type: text, text: f错误目录 {directory_path} 不存在。}], is_errorTrue ) if not path.is_dir(): return ToolResult( content[{type: text, text: f错误{directory_path} 不是一个目录。}], is_errorTrue ) # 简单的路径遍历限制可根据需要增强 try: path.resolve().relative_to(Path.home()) # 示例限制在用户家目录下 except ValueError: return ToolResult( content[{type: text, text: f错误出于安全考虑只能搜索用户家目录下的文件。}], is_errorTrue ) # 2. 执行搜索逻辑 try: # 递归遍历目录 for file_path in path.rglob(f*{file_extension}): if file_path.is_file(): try: # 以只读方式打开文件避免意外修改 with open(file_path, r, encodingutf-8, errorsignore) as f: content f.read() if keyword.lower() in content.lower(): # 简单的大小写不敏感搜索 # 构造一个格式化的结果项 results.append({ file: str(file_path.relative_to(path)), # 相对路径 path: str(file_path), matches: content.count(keyword.lower()) # 粗略匹配计数 }) except (IOError, PermissionError) as e: # 记录无法读取的文件但继续搜索其他文件 print(f警告无法读取文件 {file_path}: {e}) continue except Exception as e: # 捕获遍历过程中的意外错误 return ToolResult( content[{type: text, text: f搜索过程中发生错误{str(e)}}], is_errorTrue ) # 3. 格式化并返回结果 if results: result_text f在目录 {directory_path} 中搜索关键词 {keyword} 共找到 {len(results)} 个文件\n\n for idx, r in enumerate(results, 1): result_text f{idx}. **文件**: {r[file]}\n result_text f **完整路径**: {r[path]}\n result_text f **匹配次数**: {r[matches]}\n\n else: result_text f在目录 {directory_path} 中未找到包含关键词 {keyword} 的 {file_extension} 文件。 return ToolResult( content[{type: text, text: result_text}] ) # 服务器初始化钩子可选 app.on_initialize async def initialize_server(): 服务器启动时调用可用于建立数据库连接、加载配置等。 print(文件搜索MCP服务器正在初始化...) # 例如可以在这里加载一个允许搜索的目录白名单 # global ALLOWED_DIRS # ALLOWED_DIRS load_allowed_dirs_from_config() print(初始化完成。) # 服务器关闭钩子可选 app.on_shutdown async def shutdown_server(): 服务器关闭时调用用于清理资源。 print(文件搜索MCP服务器正在关闭执行清理...) # 例如关闭数据库连接 # await database.close() print(清理完成。) # 主入口点 if __name__ __main__: # 运行服务器。框架会处理所有与MCP客户端的通信。 app.run()3.3 配置与运行服务器MCP服务器通常作为子进程被AI客户端如Claude Desktop、第三方Agent平台调用。客户端需要一个标准的方式来定位和启动你的服务器。这是通过一个简单的JSON配置文件完成的。在server.py同级目录下创建一个名为mcp.json的配置文件{ mcpServers: { file-search: { command: python, args: [/ABSOLUTE/PATH/TO/YOUR/server.py], env: { PYTHONPATH: /ABSOLUTE/PATH/TO/YOUR/PROJECT } } } }关键配置说明command: 启动服务器的命令这里是python。args: 命令的参数第一个是server.py的绝对路径。这是最容易出错的地方务必使用绝对路径。env: 可选的环境变量。如果你的代码依赖项目内的其他模块可能需要设置PYTHONPATH。注意事项这个mcp.json文件通常不是放在你的项目里而是放在AI客户端的配置目录下。例如对于Claude Desktop你需要将此文件放在~/Library/Application Support/Claude/claude_desktop_config.jsonMac或%APPDATA%\Claude\claude_desktop_config.jsonWindows所指向的目录中。具体请参考你所使用的AI客户端的文档。配置好后当AI客户端启动时它会读取这个配置并执行指定的命令来启动你的MCP服务器。你的工具search_files_in_directory就会出现在AI的工具列表里了。4. 进阶实践构建更复杂的工具与系统一个文件搜索工具只是开始。mcp-maker的真正威力在于它能帮你轻松构建包含多个工具、具有状态和复杂交互的MCP服务。下面我们探讨几个进阶主题。4.1 工具间的依赖与共享状态假设我们除了搜索还想添加一个“标记重要文件”的工具。这个工具需要知道之前搜索到了哪些文件。我们可以通过框架的上下文Context或共享状态来实现。一种简单的方式是使用一个全局的、内存中的数据结构对于简单场景或外部存储如小型数据库。在mcp-maker的设计中更优雅的方式可能是利用其依赖注入系统如果支持或创建一个共享的服务类。# 示例一个简单的内存存储服务 class SearchResultStore: def __init__(self): self.recent_results {} # 例如{session_id: [file_paths]} def save_results(self, session_id: str, results: list): self.recent_results[session_id] results def get_results(self, session_id: str) - list: return self.recent_results.get(session_id, []) # 在初始化时创建这个存储实例 store SearchResultStore() app.tool( namesearch_and_save, description搜索文件并保存结果供后续工具使用。, input_schema{...} # 类似之前的搜索参数 ) async def search_and_save_tool(directory_path: str, keyword: str): # ... 执行搜索逻辑得到 results ... # 假设我们有一个会话ID实际中可能来自MCP请求的上下文 session_id default_session store.save_results(session_id, results) # ... 返回结果 ... app.tool( namemark_important, description从最近一次搜索结果中标记某个文件为重要。, input_schema{ type: object, properties: { file_index: { type: integer, description: 要标记的文件在最近结果列表中的索引从1开始。 } }, required: [file_index] } ) async def mark_important_tool(file_index: int): session_id default_session recent_files store.get_results(session_id) if not recent_files: return ToolResult(..., is_errorTrue) try: target_file recent_files[file_index - 1] # 执行标记逻辑例如写入一个标记文件或更新数据库 # ... return ToolResult(content[{type: text, text: f已成功标记文件{target_file}}]) except IndexError: return ToolResult(..., is_errorTrue)4.2 异步工具与长时间运行任务有些工具操作可能很耗时比如处理一个大文件或调用一个慢速API。MCP协议支持异步工具调用通过tools/call返回一个result为null并附带isProgress或后续轮询机制。mcp-maker框架应该能很好地支持异步函数async def。对于长时间运行的任务最佳实践是立即返回工具函数快速返回一个“任务已接收”的响应。后台执行在后台启动一个异步任务或使用任务队列如asyncio.create_task,celery。状态通知通过MCP协议可能的进度报告机制或另一个“检查任务状态”的工具让客户端查询结果。import asyncio app.tool( nameprocess_large_dataset, description启动一个长时间运行的数据集处理任务。, input_schema{...} ) async def process_large_dataset_tool(dataset_id: str): # 立即返回告知任务已开始 task_id ftask_{dataset_id}_{int(time.time())} # 在后台启动处理任务避免阻塞当前调用 asyncio.create_task(_background_processing(task_id, dataset_id)) return ToolResult( content[{ type: text, text: f已开始处理数据集 {dataset_id}。任务ID: {task_id}。请使用 check_task_status 工具查询进度。 }] ) async def _background_processing(task_id: str, dataset_id: str): # 这里是实际耗时的处理逻辑 await asyncio.sleep(60) # 模拟长时间工作 # 处理完成后将结果存入某个地方如数据库、内存缓存供 check_task_status 查询 # task_results[task_id] {status: completed, result: ...}4.3 安全性考量与最佳实践将本地能力暴露给AI是一个需要慎之又慎的操作。mcp-maker框架本身不强制安全策略这需要开发者自己实现。输入验证与净化这是第一道防线。就像我们在search_files工具里做的必须验证路径是否存在、是否为目录。更要防止目录遍历攻击如../../../etc/passwd。使用pathlib.Path.resolve()并检查路径是否在允许的白名单内。权限最小化MCP服务器进程应该以尽可能低的权限运行。不要用root或管理员权限。工具粒度控制不是所有工具都需要暴露给所有用户或所有会话。可以在工具装饰器或工具函数内部实现基于上下文如用户身份、会话ID的权限检查。审计日志记录所有工具调用包括参数、调用者如果可能、时间和结果。这对于调试和事后安全审计至关重要。网络隔离如果你的MCP服务器需要访问网络确保其访问范围是受控的避免成为内部网络攻击的跳板。5. 调试、测试与问题排查实录开发MCP服务器时你并非在一个真空中工作。你需要与AI客户端交互这给调试带来了一些挑战。以下是我在实践中总结的一些方法和常见问题。5.1 独立测试你的工具逻辑在集成到MCP框架之前先单独测试你的工具函数。这可以排除业务逻辑错误。# test_tool_logic.py import asyncio from server import search_files # 导入你的工具函数 async def test(): # 模拟调用 result await search_files( directory_path/Users/yourname/Documents, keywordTODO, file_extension.py ) print(Is Error:, result.is_error) for content in result.content: if content[type] text: print(content[text]) if __name__ __main__: asyncio.run(test())5.2 模拟MCP客户端进行端到端测试你可以编写一个简单的脚本模拟MCP客户端通过stdio与你的服务器通信。这能测试协议层是否正常工作。# simulate_client.py import subprocess import json import sys def run_mcp_server_test(): # 启动你的服务器进程 proc subprocess.Popen( [sys.executable, /path/to/your/server.py], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue ) # 1. 发送初始化请求 init_request { jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 1.0, capabilities: {}, clientInfo: {name: TestClient} } } proc.stdin.write(json.dumps(init_request) \n) proc.stdin.flush() # 读取初始化响应 init_response proc.stdout.readline() print(Init Response:, init_response) # 2. 请求工具列表 list_tools_request { jsonrpc: 2.0, id: 2, method: tools/list, params: {} } proc.stdin.write(json.dumps(list_tools_request) \n) proc.stdin.flush() list_response proc.stdout.readline() print(Tools List:, list_response) # 3. 调用一个工具这里需要根据上一步返回的工具名调整 call_request { jsonrpc: 2.0, id: 3, method: tools/call, params: { name: search_files_in_directory, arguments: { directory_path: /tmp/test_dir, keyword: hello } } } proc.stdin.write(json.dumps(call_request) \n) proc.stdin.flush() call_response proc.stdout.readline() print(Tool Call Result:, call_response) # 关闭进程 proc.terminate() proc.wait() if __name__ __main__: run_mcp_server_test()5.3 常见问题排查表问题现象可能原因排查步骤与解决方案AI客户端完全找不到工具1.mcp.json配置文件路径错误。2. 配置文件语法错误。3. 服务器启动命令失败。1. 检查客户端文档确认配置文件放置的绝对正确路径。2. 使用jsonlint验证mcp.json格式。3. 手动在终端运行配置中的command和args看服务器能否正常启动并打印日志。工具列表为空或缺少某个工具1. 工具定义装饰器app.tool未正确应用。2. 工具函数存在语法错误导致模块加载失败。3. 服务器初始化 (initialize) 钩子中发生未捕获异常。1. 确保工具函数定义在app mcp.MCPMaker()之后。2. 单独运行server.py脚本看是否有导入或语法错误。3. 在initialize和工具函数内添加更详细的print日志观察服务器启动时的输出。调用工具时返回协议错误1. 工具输入参数与input_schema不匹配。2. 工具函数返回值不是ToolResult类型。3. 工具函数内部抛出未处理的异常。1. 使用模拟客户端发送精确符合input_schema的请求参数。2. 确保工具函数返回return ToolResult(...)。3. 在工具函数内部用try...except包裹核心逻辑并返回错误类型的ToolResult。服务器进程意外退出1. 代码中存在导致进程崩溃的严重错误如段错误。2. 资源泄漏如打开文件未关闭被操作系统终止。3.shutdown钩子中有错误。1. 查看客户端或系统的错误日志stderr。2. 确保所有文件操作、网络连接都在with语句或finally块中正确关闭。3. 简化shutdown逻辑确保其不会抛出异常。工具执行速度慢或超时1. 工具本身是同步阻塞操作。2. 网络或IO延迟高。3. AI客户端设置了较短的超时时间。1. 将工具函数改为async def并在其中使用asyncio.to_thread运行CPU密集型任务或使用异步IO库如aiofiles,httpx。2. 对于确实耗时的操作考虑实现异步任务模式见4.2节。3. 检查客户端是否有超时配置可调整。5.4 日志记录是生命线由于MCP服务器运行在后台没有控制台输出因此一个健壮的日志系统至关重要。不要只用print。import logging # 在 server.py 开头配置日志 logging.basicConfig( levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(mcp_server.log), # 输出到文件 logging.StreamHandler() # 同时输出到控制台如果通过stdio运行客户端可能捕获 ] ) logger logging.getLogger(__name__) # 在代码中使用 app.on_initialize async def initialize_server(): logger.info(服务器初始化开始...) # ... logger.info(服务器初始化完成。) app.tool(...) async def some_tool(param): logger.info(f工具被调用参数: {param}) try: # ... logger.info(工具执行成功。) except Exception as e: logger.error(f工具执行失败: {e}, exc_infoTrue) # ...通过查看mcp_server.log文件你可以清晰地追踪服务器的生命周期和每一次工具调用这在排查复杂问题时非常有用。