使用mcp-maker快速构建AI工具集成服务器：从MCP协议到实践

1. 项目概述一个为AI应用注入“超能力”的MCP服务器工厂如果你最近在折腾AI应用开发特别是想给ChatGPT、Claude这类大模型配上“手和脚”让它们能操作你的本地文件、查询数据库甚至控制你的智能家居那你大概率已经听说过MCPModel Context Protocol这个概念了。简单来说MCP就是一套标准协议它让大模型能够安全、可控地调用外部工具和资源极大地扩展了AI的能力边界。今天要聊的这个项目MrAliHasan/mcp-maker就是一个专门用来“制造”MCP服务器的工具。你可以把它理解为一个“MCP服务器工厂”。它的核心价值在于极大地简化了MCP服务器的开发流程。过去你要想为你的数据库、你的API或者你的本地工具创建一个MCP服务器可能需要从零开始理解协议规范、处理连接、定义工具、管理资源过程相当繁琐。而有了mcp-maker你只需要用简单的Python代码描述你的工具和资源它就能帮你自动生成一个功能完整、符合MCP标准的服务器。这就像是你想开一家餐厅mcp-maker不是给你一块地和一堆建材让你自己盖房子而是直接给了你一个已经搭好水电、装修完毕的厨房框架你只需要专注于设计菜单也就是你的工具逻辑就行了。对于开发者、技术爱好者和希望快速将现有服务AI化的团队来说这无疑是一个效率神器。2. 核心设计思路为什么我们需要一个“MCP Maker”在深入代码之前我们先得搞清楚为什么MCP服务器的开发需要被“简化”这得从MCP协议本身和开发现状说起。2.1 MCP协议的核心与开发痛点MCP协议的设计目标是标准化AI模型与外部工具之间的交互。一个标准的MCP服务器需要实现几个核心部分工具Tools定义告诉AI模型“我能做什么”。比如一个“读取文件”的工具需要定义名称、描述、输入参数如文件路径。资源Resources管理告诉AI模型“我能提供什么数据”。资源可以是一个数据库表、一个API端点甚至是一个实时数据流。需要定义URI模式、元数据等。协议通信基于JSON-RPC over stdio标准输入输出或SSE服务器发送事件进行双向通信处理模型的请求并返回结果。清单Manifest声明一个配置文件声明服务器实现了哪些能力工具、资源、提示词模板等。从零开始实现这些意味着你要处理网络通信、协议解析、错误处理、生命周期管理等一系列底层细节。对于只想快速暴露一两个工具的开发者来说这个学习成本和开发成本太高了。2.2 mcp-maker的解决方案约定优于配置mcp-maker的核心思路是“约定优于配置”和“声明式编程”。它提供了一个高级的、Pythonic的抽象层。你声明“做什么”用Python函数和装饰器来定义你的工具。你只需要关心这个工具的业务逻辑比如“怎么查询数据库”、“怎么调用某个API”。它处理“怎么做”mcp-maker负责将你的函数包装成符合MCP协议的工具定义自动生成清单文件并启动一个处理所有JSON-RPC通信的服务器进程。这种设计带来了几个显著优势开发速度极快几分钟内就能将一个简单的Python函数变成AI可调用的工具。代码更清晰业务逻辑和协议胶水代码分离你的工具函数就是普通的、可测试的Python函数。降低入门门槛开发者无需深入理解MCP协议的每个细节就能构建出可用的服务器。易于集成生成的服务器可以无缝接入支持MCP的AI客户端如Claude Desktop、Cursor等。3. 从零开始手把手搭建你的第一个MCP服务器理论说再多不如动手做一遍。我们假设一个最常见的场景我想让AI能查询我本地的SQLite数据库。我们将用mcp-maker一步步实现这个“数据库查询工具”。3.1 环境准备与项目初始化首先确保你的Python环境是3.8或更高版本。然后通过pip安装mcp-makerpip install mcp-maker注意建议在虚拟环境如venv或conda中操作避免污染全局Python环境。可以使用python -m venv .venv创建并激活虚拟环境。安装完成后我们创建一个新的项目目录比如my_mcp_server并在其中开始编写我们的服务器代码。3.2 定义你的第一个工具数据库查询在项目目录下创建一个名为server.py的文件。我们将在这里编写核心代码。# server.py import sqlite3 from typing import Any from mcp import ClientSession, StdioServerParameters from mcp.server import Server from mcp.server.models import InitializationOptions import mcp.server.stdio # 1. 创建MCP服务器实例 server Server(my-database-server) # 2. 使用装饰器定义工具 server.list_tools() async def list_tools() - list: 列出本服务器提供的所有工具。 # mcp-maker会自动处理这里返回空列表即可工具信息由装饰器提供 return [] server.tool() async def query_database(sql_query: str) - str: 执行一条SQL查询语句并返回结果。 Args: sql_query: 要执行的SQL查询语句例如 SELECT * FROM users LIMIT 5; Returns: 查询结果的字符串表示如果出错则返回错误信息。 # 这里是你的业务逻辑 db_path my_database.db # 假设你的数据库文件在此 try: conn sqlite3.connect(db_path) cursor conn.cursor() cursor.execute(sql_query) # 获取查询结果 results cursor.fetchall() column_names [description[0] for description in cursor.description] if cursor.description else [] conn.close() # 格式化输出 if not results: return 查询成功但未返回任何数据。 # 简单制表符格式化 output_lines [\t.join(column_names)] if column_names else [] for row in results: output_lines.append(\t.join(str(item) for item in row)) return \n.join(output_lines) except sqlite3.Error as e: return f数据库查询出错: {e} except Exception as e: return f执行过程中发生未知错误: {e} # 3. 定义资源可选例如暴露数据库中的表作为资源 server.list_resources() async def list_resources() - list: 列出可用的资源比如数据库中的表。 # 这里我们可以动态读取数据库表名 # 为了示例简单我们返回一个静态资源列表 return [ { uri: db://schema/tables, name: 数据库表列表, description: 列出数据库中所有的表, mimeType: text/plain } ] server.read_resource() async def read_resource(uri: str) - str: 读取指定URI的资源内容。 if uri db://schema/tables: try: conn sqlite3.connect(my_database.db) cursor conn.cursor() # 查询SQLite中所有用户表 cursor.execute(SELECT name FROM sqlite_master WHERE typetable;) tables cursor.fetchall() conn.close() table_list \n.join([table[0] for table in tables]) return f数据库中的表有\n{table_list} if table_list else 数据库中没有用户表。 except sqlite3.Error as e: return f无法读取表列表: {e} return f未知资源URI: {uri} # 4. 服务器运行入口 async def main(): 运行MCP服务器。 # 配置stdio通信参数 server_params StdioServerParameters( commandpython, args[server.py] # 这里就是自身当通过stdio调用时会执行这个脚本 ) # 使用mcp-maker提供的便捷运行函数 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_nameMy Database MCP Server, server_version0.1.0, capabilitiesserver.get_capabilities() ) ) # 运行服务器主循环等待并处理来自客户端的请求 await server.run( session, read_stream, write_stream, # 这里可以传入你的工具和资源列表但因为我们用了装饰器通常不需要 ) if __name__ __main__: import asyncio asyncio.run(main())3.3 配置与清单生成mcp-maker的一个便利之处是它通常能根据你的代码自动推断或生成部分配置。但为了更精细的控制我们可以创建一个pyproject.toml或mcp.json文件来声明服务器信息。不过对于许多简单场景像上面那样在代码中使用装饰器已经足够了。当AI客户端如Claude Desktop连接时它会调用list_tools和list_resources来发现你的服务器能力。3.4 运行与测试现在我们如何测试这个服务器呢最直接的方式是使用一个MCP客户端来连接它。首先确保你有一个SQLite数据库文件my_database.db里面有一张表例如users包含id,name,email字段并插入一些测试数据。运行服务器在终端中你可以直接运行python server.py。但是一个纯粹的MCP服务器设计是通过stdio与父进程通信所以直接运行可能只会启动并等待连接。通常我们需要在支持MCP的AI应用中进行测试。在Claude Desktop中测试推荐打开Claude Desktop应用。进入设置Settings- 开发者Developer- MCP服务器配置。点击“添加MCP服务器”Add MCP Server。在配置中选择“命令”Command类型。命令填写python参数填写你的server.py文件的绝对路径例如/Users/yourname/projects/my_mcp_server/server.py。给它起个名字比如 “My Database Query”。保存并重启Claude Desktop。重启后你就可以在Claude的对话中使用了。尝试输入“请使用My Database Query工具查询users表的前10条数据。” Claude应该能识别到这个工具并调用它将SQL查询语句SELECT * FROM users LIMIT 10;发送给你的服务器然后将返回的结果展示给你。4. 进阶技巧构建更复杂、更实用的MCP服务器掌握了基础工具创建后我们可以探索mcp-maker更强大的功能构建真正实用的集成。4.1 处理复杂参数与类型提示MCP协议支持丰富的参数类型。mcp-maker充分利用Python的类型提示Type Hints来生成更准确的工具定义。from pydantic import BaseModel from typing import List, Optional from enum import Enum class FileFormat(str, Enum): CSV csv JSON json EXCEL excel class ExportDataToolInput(BaseModel): 导出数据的工具参数模型 table_name: str format: FileFormat FileFormat.CSV limit: Optional[int] 100 columns: Optional[List[str]] None server.tool() async def export_table_data(args: ExportDataToolInput) - str: 将指定表的数据导出为指定格式。 Args: args: 包含导出参数的对象。 # 通过args.table_name, args.format等访问参数 # ... 你的导出逻辑 ... return f成功将表 {args.table_name} 导出为 {args.format} 格式共 {rows_exported} 行。使用pydantic模型和EnumAI客户端能获得清晰的结构化参数定义用户输入时也会有更好的引导和验证。4.2 实现资源Resources与提示词模板Prompts除了工具资源是MCP的另一大核心。资源可以是静态的参考数据也可以是动态生成的内容。server.list_resources() async def list_my_resources(): return [ { uri: resource://docs/api-overview, name: API概览文档, description: 本系统主要API的简要说明, mimeType: text/markdown }, { uri: resource://system/health, name: 系统健康状态, description: 实时系统健康检查信息, mimeType: application/json } ] server.read_resource() async def read_my_resource(uri: str): if uri resource://docs/api-overview: return # API 概览\n\n- /users: 用户管理\n- /posts: 内容管理\n... elif uri resource://system/health: import psutil, json, datetime health_info { timestamp: datetime.datetime.now().isoformat(), cpu_percent: psutil.cpu_percent(), memory_percent: psutil.virtual_memory().percent, status: healthy if psutil.cpu_percent() 80 else degraded } return json.dumps(health_info, indent2) else: raise ValueError(fUnknown resource: {uri})提示词模板Prompts则允许你预定义一些高质量的提示供AI直接调用或组合确保任务执行的准确性和一致性。server.list_prompts() async def list_prompts(): return [ { name: analyze_sales_trend, description: 分析指定时间段内的销售趋势, arguments: [ {name: start_date, description: 开始日期 (YYYY-MM-DD), required: True}, {name: end_date, description: 结束日期 (YYYY-MM-DD), required: True} ] } ] server.get_prompt() async def get_prompt(name: str, arguments: dict): if name analyze_sales_trend: start arguments.get(start_date) end arguments.get(end_date) # 这里可以动态生成提示词甚至结合查询数据库 prompt_text f请分析从{start}到{end}期间的销售数据。请按以下步骤进行 1. 计算总销售额和订单量。 2. 找出销售额最高的前5个产品类别。 3. 分析日销售额的趋势指出是否有明显的高峰或低谷。 4. 给出下一阶段的销售策略建议。 请以清晰、有条理的报告形式呈现。 return {messages: [{role: user, content: prompt_text}]}4.3 错误处理与日志记录健壮的生产级服务器必须有良好的错误处理和日志。import logging logging.basicConfig(levellogging.INFO, format%(asctime)s - %(name)s - %(levelname)s - %(message)s) logger logging.getLogger(__name__) server.tool() async def risky_operation(file_path: str) - str: try: # 模拟可能失败的操作 with open(file_path, r) as f: content f.read() logger.info(f成功读取文件: {file_path}) return f文件内容长度: {len(content)} except FileNotFoundError: error_msg f文件未找到: {file_path} logger.error(error_msg) # 返回给AI的错误信息应清晰、可操作 return f操作失败{error_msg}。请检查文件路径是否正确。 except Exception as e: logger.exception(f读取文件时发生未知错误: {e}) return f操作失败系统内部错误。4.4 安全性考量将内部工具暴露给AI存在安全风险必须谨慎。输入验证与净化永远不要相信来自AI的输入。像之前的SQL查询工具直接执行用户输入的SQL是极度危险的SQL注入。应该进行严格的白名单验证或者只提供预定义的查询选项。ALLOWED_QUERIES { “get_user_count”: “SELECT COUNT(*) FROM users;”, “get_recent_orders”: “SELECT * FROM orders ORDER BY created_at DESC LIMIT 10;” } server.tool() async def safe_query(query_key: str) - str: if query_key not in ALLOWED_QUERIES: return f“不允许的查询类型{query_key}。请使用{list(ALLOWED_QUERIES.keys())}” sql ALLOWED_QUERIES[query_key] # ... 执行安全的sql ...权限控制服务器运行在什么权限下它能访问哪些文件系统、网络资源遵循最小权限原则。操作确认对于删除、修改等危险操作可以在工具逻辑中设计二次确认机制或者仅在受信任的环境中使用。5. 实战构建一个多功能个人助理MCP服务器现在我们综合运用以上知识构建一个更实用的“个人助理”MCP服务器集成日历、待办事项和文件搜索功能。# personal_assistant_server.py import json import os from pathlib import Path from typing import List, Optional from datetime import datetime, date from pydantic import BaseModel from mcp import Server server Server(personal-assistant) # --- 日历工具 --- CALENDAR_FILE Path.home() / .personal_assistant / calendar.json CALENDAR_FILE.parent.mkdir(parentsTrue, exist_okTrue) class CalendarEvent(BaseModel): title: str date: str # YYYY-MM-DD time: Optional[str] None description: Optional[str] None def load_calendar() - List[dict]: if CALENDAR_FILE.exists(): with open(CALENDAR_FILE, r) as f: return json.load(f) return [] def save_calendar(events: List[dict]): with open(CALENDAR_FILE, w) as f: json.dump(events, f, indent2) server.tool() async def add_calendar_event(event: CalendarEvent) - str: 添加一个新的日历事件。 events load_calendar() event_dict event.dict() event_dict[id] len(events) 1 events.append(event_dict) save_calendar(events) return f已添加事件{event.title} 于 {event.date}。 server.tool() async def get_events_today() - str: 获取今天的所有日历事件。 today date.today().isoformat() events load_calendar() todays_events [e for e in events if e[date] today] if not todays_events: return 今天没有安排任何事件。 result [今天的安排] for e in todays_events: result.append(f- {e[time] or 全天}: {e[title]}) if e.get(description): result.append(f 描述{e[description]}) return \n.join(result) # --- 文件搜索工具 --- server.tool() async def find_files(keyword: str, search_dir: Optional[str] None) - str: 在指定目录默认为Home目录下递归搜索包含关键字的文件名。 Args: keyword: 搜索关键词不区分大小写。 search_dir: 搜索的起始目录路径。 root_dir Path(search_dir) if search_dir else Path.home() if not root_dir.exists() or not root_dir.is_dir(): return f错误目录 {root_dir} 不存在或不可访问。 matches [] try: # 限制搜索深度和数量避免性能问题 for p in root_dir.rglob(*): if keyword.lower() in p.name.lower(): # 显示相对路径更简洁 try: rel_path p.relative_to(root_dir) except ValueError: rel_path p matches.append(str(rel_path)) if len(matches) 50: # 限制结果数量 matches.append(... (结果超过50条已截断)) break except PermissionError: return f在搜索目录 {root_dir} 时遇到权限错误。 if not matches: return f在 {root_dir} 及其子目录下未找到包含 {keyword} 的文件或文件夹。 return f找到 {len(matches)} 个结果\n \n.join(matches[:10]) # 只展示前10个 # --- 资源提供使用指南 --- server.list_resources() async def list_assistant_resources(): return [ { uri: resource://assistant/guide, name: 个人助理使用指南, description: 本服务器提供的所有功能说明, mimeType: text/markdown } ] server.read_resource() async def read_assistant_resource(uri: str): if uri resource://assistant/guide: return # 个人助理 MCP 服务器指南 ## 可用工具 1. **日历管理** - add_calendar_event: 添加新事件。需要标题、日期。 - get_events_today: 查看今日安排。 2. **文件搜索** - find_files: 按文件名关键词搜索文件。 ## 使用示例 - “帮我添加一个明天下午3点的会议标题是‘项目评审’。” - “我今天有什么安排” - “在我的文档文件夹里找一下包含‘报告’关键词的PDF文件。” return 未知资源。 # 运行部分与之前示例类似此处省略... if __name__ __main__: # ... 运行服务器的异步代码 ... pass这个服务器虽然简单但已经具备了实用的雏形。你可以在此基础上继续扩展比如集成邮件发送、调用天气预报API、控制智能家居设备等真正打造一个属于你自己的、可被AI驱动的数字助理。6. 部署、调试与最佳实践6.1 调试你的MCP服务器调试stdio服务器有点特殊因为它的输入输出不是直接面向终端的。使用日志如前所述添加详细的logging记录是调试的基石。将日志输出到文件。logging.basicConfig( levellogging.DEBUG, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(mcp_server.log), logging.StreamHandler() # 同时输出到控制台如果直接运行的话 ] )模拟客户端测试你可以编写一个简单的测试脚本模拟MCP客户端向你的服务器发送请求。# test_client.py import asyncio import json import subprocess import sys async def test_tool(): # 启动服务器进程 proc await asyncio.create_subprocess_exec( sys.executable, server.py, stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE ) # 模拟初始化请求 init_request { jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 0.1.0, capabilities: {}, clientInfo: {name: TestClient} } } proc.stdin.write((json.dumps(init_request) \n).encode()) await proc.stdin.drain() # 读取响应简化处理实际协议更复杂 # ... 解析响应然后测试工具调用 ... proc.terminate() asyncio.run(test_tool())使用MCP Inspector工具社区有一些工具可以帮助你检查和调试MCP服务器例如一些图形化的调试客户端可以直观地看到通信过程。6.2 性能与可扩展性异步是核心MCP服务器本质是I/O密集型的网络、数据库、文件。确保你的工具函数都使用async/await并选择合适的异步库如aiohttp用于HTTP请求aiosqlite用于数据库。避免阻塞操作不要在工具函数中执行长时间同步的CPU密集型计算这会阻塞整个服务器。如果必须考虑使用asyncio.to_thread在单独线程中运行。连接池与缓存对于数据库、API客户端等使用连接池和适当的缓存机制如lru_cache来提升性能。6.3 部署考量打包可以使用pyinstaller或shiv将你的服务器打包成可执行文件简化部署避免环境依赖问题。进程管理在生产环境中你需要确保服务器进程的稳定运行。可以使用systemd(Linux)、launchd(macOS) 或进程管理器如supervisord、pm2来管理。配置化将数据库连接字符串、API密钥、文件路径等敏感或可配置信息从代码中抽离使用环境变量或配置文件管理。6.4 与不同AI客户端的兼容性虽然MCP是协议但不同客户端Claude Desktop、Cursor、第三方应用的实现可能略有差异。测试时最好在你目标使用的客户端中进行。关注客户端日志它们通常会显示与MCP服务器通信的错误信息。7. 避坑指南与常见问题在实际使用mcp-maker和开发MCP服务器的过程中我踩过不少坑这里总结一下希望能帮你绕过去。问题1工具在AI客户端中不显示或调用失败。检查清单首先确认你的服务器是否正确实现了initialize和list_tools调用。在服务器日志中查看初始化是否成功。检查装饰器确保你的工具函数正确使用了server.tool()装饰器并且函数有清晰的文档字符串docstringAI客户端依赖这些信息。检查参数定义工具参数是否使用了简单的、可JSON序列化的类型复杂的自定义类型可能需要额外的模式定义。优先使用str,int,float,bool,List,Optional和pydantic.BaseModel。重启客户端很多AI客户端如Claude Desktop只在启动时加载MCP服务器配置修改服务器代码后需要重启客户端。问题2服务器启动后立即退出或报错。检查入口点确保if __name__ __main__:块正确并且调用了asyncio.run(main())。检查导入确保所有依赖库mcp,pydantic等都已正确安装。查看stderr服务器通过stdio通信错误信息通常会打印到标准错误流。在配置AI客户端时有时可以勾选“显示服务器输出”来查看日志。问题3工具执行超时或无响应。工具函数是否异步确保工具函数是async def定义的并且在执行I/O操作时使用了await。是否有死循环或长时间操作工具函数应在较短时间内几秒内返回。长时间任务应考虑异步通知或状态查询机制。客户端超时设置有些客户端有默认的超时时间如30秒如果你的工具确实需要更长时间可能需要调整客户端设置如果支持的话。问题4如何让工具返回结构化的数据如JSON供AI进一步处理返回字符串化的JSON在工具函数中将结果字典或列表用json.dumps()转换成字符串返回。并在工具描述中说明返回的是JSON格式。使用资源Resource对于更大的或结构固定的数据可以考虑通过资源Resource暴露。工具执行成功后返回一个资源URIAI可以再通过read_resource来获取结构化内容。这更符合MCP的设计模式。问题5我想动态添加或删除工具可以吗mcp-maker的装饰器模式通常在服务器启动时静态定义工具。对于动态性要求高的场景你可能需要更底层地操作Server实例的tools列表或者设计一个“元工具”由这个元工具来管理其他工具的生命周期。这属于更高级的用法需要你对MCP协议有更深的理解。开发MCP服务器的过程是一个让AI能力“落地”的过程。mcp-maker这个工具极大地降低了门槛但它只是一个起点。真正的挑战和乐趣在于如何设计出直观、安全、强大的工具让AI成为你工作流中真正得力的助手。从自动化一个简单的数据查询开始逐步构建起属于你自己的智能工具生态这个过程本身就充满了创造性和实用性。