1. 项目概述一个为AI应用量身定制的“工具箱”构建框架最近在折腾AI应用开发特别是那些需要让大模型LLM与外部工具、数据源进行深度交互的场景时我发现了一个绕不开的痛点如何高效、标准化地让模型“理解”并“调用”我们自定义的功能无论是让AI去查询数据库、调用一个内部API还是执行一个复杂的脚本都需要一套清晰的“沟通协议”。正是在这个背景下我深入研究了mcp-custom-dev这个项目。简单来说它不是一个现成的工具集而是一个开发框架专门用于帮助你为AI模型尤其是遵循MCP协议的模型或客户端创建自定义的、可被模型理解和使用的“工具”Tools或“资源”Resources。你可以把它想象成给AI模型打造一个专属的、可扩展的“瑞士军刀”的“工厂”和“说明书生成器”。这个项目本身不提供具体的螺丝刀或剪刀但它提供了制造这些工具的标准流程、接口规范以及最重要的——如何生成一份AI能读懂的“工具使用说明书”。这对于想要将企业内部系统、私有API或特定业务流程接入AI助手的开发者来说价值巨大。它解决了从“我有一个功能”到“AI能安全、准确地使用这个功能”之间的标准化桥梁问题。2. MCP协议核心AI与工具对话的“普通话”要理解mcp-custom-dev必须先搞懂它背后的MCPModel Context Protocol。你可以把MCP理解为AI模型与外部工具之间进行对话的“普通话”或“标准通信协议”。在没有MCP之前每个AI应用想要连接外部工具都可能需要自定义一套复杂的提示词工程、函数调用描述和结果解析逻辑这就像两个人用各自方言沟通效率低且容易出错。MCP协议的核心思想是标准化工具的描述与调用。它主要定义了几种关键组件工具Tools一个可供模型调用的具体操作比如“查询天气”、“发送邮件”。每个工具需要有明确的名称、描述、输入参数参数名、类型、描述和输出格式。资源Resources模型可以读取的静态或动态数据源比如一个文件、一个数据库表的视图。资源有唯一的URI、描述和MIME类型。提示词模板Prompts预定义的、可参数化的提示词片段方便模型快速获取特定上下文。mcp-custom-dev项目正是基于MCP协议为开发者提供了一套便捷的脚手架和开发范式让你无需从零开始理解协议的所有细节就能快速构建出符合MCP标准的工具服务器Server。这个服务器启动后就能被任何兼容MCP的AI客户端例如某些集成了MCP的AI IDE或智能体平台发现并调用。2.1 为什么需要自定义MCP服务器你可能会问市面上不是已经有一些现成的MCP工具服务器了吗比如连接GitHub、Notion的。确实但对于企业级、个性化的需求自定义开发是必然选择连接内部系统你的CRM、ERP、内部工单系统、监控平台这些不可能有公开的通用MCP服务器。封装复杂流程将多个API调用、数据处理步骤封装成一个简单的工具比如“一键生成周报并发送给领导”这个工具背后可能调用了数据查询、文本生成、邮件发送等多个服务。安全与权限控制在工具服务器层面实现精细化的权限验证、审计日志确保AI只能在授权范围内操作。性能与稳定性优化针对高并发或延迟敏感的内部服务可以自定义连接池、缓存和重试机制。mcp-custom-dev的价值就在于它降低了构建这样一个标准化、专业化工具服务器的门槛。3. 项目架构与核心设计思路mcp-custom-dev作为一个开发框架其设计思路非常清晰以工具Tool和资源Resource为核心通过标准的MCP协议接口暴露给客户端。整个架构通常遵循客户端-服务器模型但这里的“客户端”是AI模型或AI应用平台。3.1 核心组件拆解一个基于mcp-custom-dev构建的项目通常会包含以下几个核心部分工具定义模块这是项目的核心。你需要在这里用代码声明每一个工具。声明内容包括name: 工具的唯一标识符如get_weather。description: 给AI模型看的自然语言描述这至关重要例如“根据城市名称查询当前天气情况和温度。城市名称应为中文。” 清晰准确的描述能极大提升模型调用的准确性。inputSchema: 定义输入参数的JSON Schema。这是强类型约束确保传入的参数格式正确。例如为city参数定义类型为string并可能提供enum枚举值限制。# 示例性代码结构非项目原码 from mcp_sdk import Tool Tool( namequery_sales_data, description查询指定区域和时间段的销售总额。, inputSchema{ type: object, properties: { region: {type: string, description: 销售区域如华北、华东}, start_date: {type: string, format: date}, end_date: {type: string, format: date} }, required: [region] } ) async def query_sales_data_tool(region: str, start_date: str None, end_date: str None): # 实际的业务逻辑调用内部API、查询数据库等 data await internal_api.get_sales(region, start_date, end_date) return {total_sales: data.total, unit: CNY}资源定义模块类似地定义可供读取的资源。例如你可以定义一个资源其URI为dynamic://internal-dashboard/active-users当AI客户端请求该资源时服务器端动态查询数据库并返回当前活跃用户数的HTML或Markdown片段。from mcp_sdk import Resource Resource( uridynamic://metrics/server_health, description实时服务器健康状态概览 ) async def get_server_health(): cpu, memory, disk await monitor.get_system_metrics() return f ## 服务器健康状态 - **CPU使用率**: {cpu}% - **内存使用率**: {memory}% - **磁盘剩余空间**: {disk}GB 协议适配与服务器启动框架会帮你处理MCP协议的底层通信细节通常基于STDIO或HTTP。你只需要将定义好的工具和资源“注册”到服务器实例然后启动它。服务器启动后会等待兼容MCP的客户端连接。配置与依赖管理项目通常包含配置文件如config.yaml或.env文件用于管理数据库连接字符串、API密钥、服务端口等。依赖管理则确保项目所需的所有Python库如mcpSDK、httpx、pydantic等被正确安装。3.2 设计模式与最佳实践在组织代码时推荐采用分层或模块化的设计工具层集中存放所有工具函数保持函数功能单一、纯净。服务层封装对内部系统或第三方API的调用工具函数调用服务层。数据模型层使用Pydantic等库定义输入输出的数据模型便于验证和序列化。配置层统一管理所有配置项。实操心得描述就是生产力在定义工具的description和参数的description时一定要站在AI模型而不是人类程序员的角度去写。要清晰、无歧义、包含示例。例如“请输入日期”是糟糕的描述“请输入日期格式为YYYY-MM-DD例如2023-10-27”是好的描述。这能直接减少模型调用错误。4. 从零开始构建你的第一个自定义MCP工具理论说了这么多我们来动手实现一个具体的例子构建一个“公司内部知识库问答”工具。这个工具允许AI模型根据用户问题查询我们内部的文档库假设是一个Elasticsearch索引并返回最相关的答案片段。4.1 环境准备与项目初始化首先确保你的开发环境已就绪# 1. 创建项目目录并进入 mkdir my-mcp-knowledge-server cd my-mcp-knowledge-server # 2. 创建虚拟环境推荐 python -m venv venv # Windows: venv\Scripts\activate # Mac/Linux: source venv/bin/activate # 3. 初始化项目并安装核心依赖 # 假设 mcp-custom-dev 或其SDK可以通过pip安装 pip install mcp # 这是MCP的官方Python SDK是构建的基础 pip install pydantic httpx elasticsearch python-dotenv pip install uvloop # 可选用于提升异步性能 # 4. 创建基础项目结构 touch main.py touch tools/ touch config.py touch .env4.2 定义核心工具知识库查询在tools/knowledge.py中我们定义核心工具# tools/knowledge.py import logging from typing import Optional from mcp import Tool from pydantic import BaseModel, Field # 假设我们有一个封装好的ES客户端 from services.elasticsearch_client import es_client logger logging.getLogger(__name__) class KnowledgeQueryInput(BaseModel): 知识库查询工具的输入参数模型 question: str Field( ..., description用户提出的具体问题例如我们的产品退款政策是什么 或 如何配置XX服务器的网络, min_length5 ) max_results: Optional[int] Field( 3, description返回最相关文档片段的最大数量默认为3。, ge1, le10 ) Tool( namequery_company_knowledge_base, description在公司内部知识库中搜索与用户问题相关的文档片段。 知识库涵盖了产品手册、内部流程、技术文档和常见问题解答。 请确保问题描述具体清晰以获得更准确的结果。, # inputSchema 可以由 Pydantic 模型自动生成框架通常支持 ) async def query_company_knowledge_base(input_data: KnowledgeQueryInput) - dict: 执行知识库搜索的核心函数。 question input_data.question max_results input_data.max_results logger.info(f正在知识库中搜索: {question}) # 1. 构建Elasticsearch查询DSL # 这里使用简单的match查询实际生产环境可能要用更复杂的multi-match或BM25调优 search_body { query: { match: { content: question } }, size: max_results, _source: [title, content_snippet, url, last_updated] } try: # 2. 执行搜索 response await es_client.search(indexcompany-knowledge, bodysearch_body) hits response.get(hits, {}).get(hits, []) # 3. 格式化结果使其对AI模型友好 if not hits: return { answer: 在知识库中未找到直接相关的信息。, suggestions: [尝试使用更具体的关键词重新提问。, 或联系相关部门的同事获取帮助。] } formatted_results [] for hit in hits: source hit[_source] formatted_results.append({ 标题: source.get(title, 无标题), 相关片段: source.get(content_snippet, )[:200] ..., # 截取片段 来源链接: source.get(url, #), 相关性得分: round(hit[_score], 2) }) # 4. 返回结构化结果 return { answer: f根据你的问题『{question}』我在知识库中找到以下{len(formatted_results)}条最相关的内容, results: formatted_results, search_performed: True } except Exception as e: logger.error(f知识库查询失败: {e}, exc_infoTrue) # 返回一个清晰的错误信息而不是抛出异常避免服务器崩溃 return { answer: 查询知识库时遇到内部错误暂时无法获取信息。, error_detail: str(e), search_performed: False }4.3 集成工具并启动MCP服务器在main.py中我们将工具集成并启动服务器# main.py import asyncio import logging from contextlib import asynccontextmanager from mcp import Server, StdioServerParameters from mcp.server import NotificationOptions import tools.knowledge # 导入工具模块完成装饰器注册 # 通常框架会有自动发现工具的机制这里演示显式导入 from tools.knowledge import query_company_knowledge_base # 配置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) # 假设我们还有其他工具... # from tools.calendar import schedule_meeting_tool # from tools.jira import create_jira_ticket_tool async def main(): 启动MCP服务器的主函数 # 1. 创建Server实例 # 这里需要根据你使用的具体mcp库的API来调整 server Server( namecompany-internal-tools, version0.1.0, notification_optionsNotificationOptions(), ) # 2. 注册工具 # 方式取决于框架可能是自动扫描也可能是手动注册 # 手动注册示例如果框架支持 server.register_tool(query_company_knowledge_base) # server.register_tool(schedule_meeting_tool) # server.register_tool(create_jira_ticket_tool) # 3. 配置服务器参数使用标准输入输出这是MCP常见方式 server_params StdioServerParameters() logger.info(开始启动 Company Internal Tools MCP 服务器...) # 4. 运行服务器 async with server.run_stdio(server_params) as (read_stream, write_stream): logger.info(MCP服务器已就绪正在等待客户端连接...) # 服务器进入事件循环处理客户端请求 await server.wait_for_disconnect() logger.info(MCP服务器已停止。) if __name__ __main__: asyncio.run(main())4.4 配置与运行创建.env文件存储敏感配置# .env ELASTICSEARCH_HOSTShttps://your-es-cluster.internal:9200 ELASTICSEARCH_USERyour_service_account ELASTICSEARCH_PASSWORDyour_password KNOWLEDGE_INDEXcompany-knowledge创建config.py加载配置# config.py import os from dotenv import load_dotenv load_dotenv() ELASTICSEARCH_CONFIG { hosts: [os.getenv(ELASTICSEARCH_HOSTS)], http_auth: (os.getenv(ELASTICSEARCH_USER), os.getenv(ELASTICSEARCH_PASSWORD)), verify_certs: True, # 生产环境应为True }最后运行你的服务器python main.py服务器启动后它会等待兼容MCP的客户端如 Claude Desktop、Cursor等配置了MCP的AI应用通过标准输入输出与其建立连接。一旦连接客户端就能发现并调用你定义的query_company_knowledge_base工具了。注意事项错误处理与超时在工具函数中务必要有完善的错误处理try-except和超时控制。AI客户端调用工具时如果工具长时间无响应或崩溃会导致整个交互体验变差。对于网络请求如调用ES务必设置合理的超时时间并使用asyncio.wait_for或httpx.Timeout进行控制。5. 高级特性与生产级考量当你掌握了基础工具开发后为了将其用于生产环境还需要考虑以下几个关键方面5.1 工具的动态注册与发现在更复杂的场景中你的工具列表可能不是静态的而是根据配置、数据库内容或用户权限动态变化的。mcp-custom-dev类框架通常支持动态工具注册。你可以在服务器启动后根据条件向服务器实例动态添加或移除工具。例如根据当前登录用户的角色决定是否注册“删除数据库”这类高危工具async def initialize_tools_based_on_user(user_role: str): if user_role admin: server.register_tool(dangerous_admin_tool) # 注册其他通用工具...5.2 认证、授权与审计AAA这是企业级应用的核心。MCP协议本身可能不直接处理认证但这部分必须在你的工具服务器中实现。认证客户端连接时如何验证其身份可以通过启动参数传递令牌或在首次握手时进行认证。一种常见模式是使用API密钥或OAuth2.0客户端凭证。授权即使认证通过用户/客户端是否有权调用某个工具这需要在工具函数内部或通过装饰器进行权限检查。审计谁在什么时候调用了什么工具输入输出是什么注意脱敏这些日志对于安全追溯和问题排查至关重要。可以在工具装饰器或一个公共的拦截器Middleware中实现这些逻辑def audit_log(tool_name): def decorator(func): wraps(func) async def wrapper(*args, **kwargs): user get_current_user() logger.info(fAUDIT: User {user} invoking tool {tool_name} with args {kwargs}) start time.time() try: result await func(*args, **kwargs) logger.info(fAUDIT: Tool {tool_name} executed successfully in {time.time()-start:.2f}s) # 注意记录结果时可能需要脱敏避免日志泄露敏感数据 return result except Exception as e: logger.error(fAUDIT: Tool {tool_name} failed with error: {e}) raise return wrapper return decorator # 在工具上使用 Tool(...) audit_log(query_sales_data) async def query_sales_data_tool(...): ...5.3 性能优化异步、缓存与连接池异步AsyncMCP服务器和工具函数强烈建议使用异步编程asyncio。这能保证在等待I/O如网络请求、数据库查询时服务器可以处理其他请求提高并发能力。确保你使用的所有客户端库如httpx.AsyncClient,aiopg,asyncpg都支持异步。缓存对于查询类、结果变化不频繁的工具引入缓存可以极大提升响应速度并降低后端压力。可以使用aiocache或redis配合异步客户端。from aiocache import cached cached(ttl300) # 缓存5分钟 Tool(...) async def get_department_list(...): # 昂贵的数据库查询 ...连接池对于数据库、Elasticsearch等外部服务务必使用连接池而不是为每次调用创建新连接。5.4 测试策略单元测试与集成测试为MCP工具编写测试至关重要。单元测试直接测试工具函数本身模拟mock所有外部依赖如ES客户端、API调用。确保各种输入正常、边界、异常下函数行为符合预期。集成测试/端到端测试启动一个测试版的MCP服务器使用一个MCP客户端测试库或模拟客户端实际连接并调用工具验证整个流程。这能发现协议层或集成上的问题。6. 常见问题与实战排坑记录在实际开发和部署mcp-custom-dev类项目的过程中我踩过不少坑这里总结几个典型问题和解决方案问题现象可能原因排查步骤与解决方案AI客户端无法发现工具1. MCP服务器未正确启动或通信协议不一致。2. 工具定义不符合MCP协议规范如缺少必需字段。3. 客户端配置的服务器路径或参数错误。1. 检查服务器日志确认已成功启动并监听。2. 使用mcpSDK自带的CLI工具如mcp dev或编写一个简单的测试脚本来连接服务器列出工具验证服务器本身是否正常。3. 仔细核对客户端如Claude Desktop的配置JSON文件确保command和args指向正确的Python解释器和脚本路径。工具调用超时或无响应1. 工具函数内部有同步阻塞操作如requests.get而非httpx.AsyncClient。2. 工具函数陷入死循环或等待资源死锁。3. 网络问题导致调用外部服务超时。1.将所有I/O操作改为异步。这是最常见的原因。检查代码确保没有使用requests,subprocess.run等同步库。2. 在工具函数中添加超时控制async with asyncio.timeout(30): ...。3. 增加详细的日志定位卡在哪一步。对外部服务调用配置合理的超时和重试。工具返回结果但AI模型“不理解”1. 工具返回的数据结构过于复杂或非结构化。2. 工具描述description不清晰导致模型误用。3. 错误信息不够友好。1.返回结构化的、简洁的JSON对象。优先使用字符串、数字、布尔值、简单数组和对象。避免返回多层嵌套的复杂对象或自定义类实例。2.优化工具和参数的描述。用自然语言明确说明工具的用途、限制和参数示例。这是提示词工程的一部分。3. 错误时返回一个包含error或message字段的JSON对象而不是抛出未处理的异常。权限错误或认证失败1. 环境变量或配置文件未正确加载。2. 密钥过期或权限不足。3. 服务器端认证逻辑有bug。1. 在服务器启动时打印关键配置脱敏后以确认加载成功。2. 实现一个简单的“健康检查”工具或接口测试对外部服务的连接状态。3. 认证逻辑单独编写单元测试模拟各种令牌场景。服务器内存泄漏或CPU占用高1. 工具函数内有资源未释放如文件句柄、数据库连接。2. 缓存策略不当缓存无限增长。3. 某个工具计算过于密集阻塞事件循环。1. 使用async with确保客户端资源正确关闭。2. 为缓存设置大小限制和过期时间TTL。3. 对于CPU密集型任务考虑使用asyncio.to_thread或ProcessPoolExecutor将其放到单独线程/进程中执行避免阻塞主事件循环。一个关键的避坑技巧善用“模拟客户端”进行开发调试。在开发初期不要急于对接复杂的AI客户端。可以写一个简单的Python脚本模拟MCP客户端与你的服务器通信直接调用工具并打印结果。这能让你快速验证工具逻辑是否正确而不受客户端复杂性的干扰。许多MCP SDK都提供了用于测试的客户端库。7. 扩展思路不止于工具服务器当你熟练使用mcp-custom-dev模式后可以将其思路扩展到更广阔的领域构建工具市场/仓库将公司内各部门开发的优质MCP工具标准化、版本化形成一个内部工具市场。AI助手可以根据用户需求动态加载和组合不同的工具。实现复杂工作流单个工具能力有限但你可以开发一个“工作流编排”工具。这个工具本身接受一个复杂目标然后在内部按顺序或条件调用其他多个基础工具最终返回整合结果。这相当于让AI具备了执行多步骤任务的能力。与低代码平台结合将MCP工具作为低代码平台的后端“积木”。业务人员通过界面配置流程实际上生成的是对一系列MCP工具的调用序列。监控与可观测性为你的MCP服务器集成监控如Prometheus指标、分布式追踪如OpenTelemetry。监控每个工具的调用次数、延迟、错误率这对于维护一个稳定的AI工具生态至关重要。回过头看mcp-custom-dev这类项目代表的是一种范式转变从“让人去适应系统”到“让系统AI来服务人”。它通过一个轻量级、标准化的协议将企业内部浩瀚的数字能力封装成AI可以理解和操作的“技能”。开发这样的工具服务器技术难点并不在于协议本身而在于如何设计出语义清晰、边界明确、安全可靠的工具这需要开发者对业务有深刻理解并具备良好的软件工程和提示词工程能力。我的体会是成功的AI工具集成30%在技术70%在设计和沟通——既包括与AI模型的“沟通”工具描述也包括与业务方的沟通明确需求和边界。
基于MCP协议构建AI工具开发框架：从原理到企业级应用实践
1. 项目概述一个为AI应用量身定制的“工具箱”构建框架最近在折腾AI应用开发特别是那些需要让大模型LLM与外部工具、数据源进行深度交互的场景时我发现了一个绕不开的痛点如何高效、标准化地让模型“理解”并“调用”我们自定义的功能无论是让AI去查询数据库、调用一个内部API还是执行一个复杂的脚本都需要一套清晰的“沟通协议”。正是在这个背景下我深入研究了mcp-custom-dev这个项目。简单来说它不是一个现成的工具集而是一个开发框架专门用于帮助你为AI模型尤其是遵循MCP协议的模型或客户端创建自定义的、可被模型理解和使用的“工具”Tools或“资源”Resources。你可以把它想象成给AI模型打造一个专属的、可扩展的“瑞士军刀”的“工厂”和“说明书生成器”。这个项目本身不提供具体的螺丝刀或剪刀但它提供了制造这些工具的标准流程、接口规范以及最重要的——如何生成一份AI能读懂的“工具使用说明书”。这对于想要将企业内部系统、私有API或特定业务流程接入AI助手的开发者来说价值巨大。它解决了从“我有一个功能”到“AI能安全、准确地使用这个功能”之间的标准化桥梁问题。2. MCP协议核心AI与工具对话的“普通话”要理解mcp-custom-dev必须先搞懂它背后的MCPModel Context Protocol。你可以把MCP理解为AI模型与外部工具之间进行对话的“普通话”或“标准通信协议”。在没有MCP之前每个AI应用想要连接外部工具都可能需要自定义一套复杂的提示词工程、函数调用描述和结果解析逻辑这就像两个人用各自方言沟通效率低且容易出错。MCP协议的核心思想是标准化工具的描述与调用。它主要定义了几种关键组件工具Tools一个可供模型调用的具体操作比如“查询天气”、“发送邮件”。每个工具需要有明确的名称、描述、输入参数参数名、类型、描述和输出格式。资源Resources模型可以读取的静态或动态数据源比如一个文件、一个数据库表的视图。资源有唯一的URI、描述和MIME类型。提示词模板Prompts预定义的、可参数化的提示词片段方便模型快速获取特定上下文。mcp-custom-dev项目正是基于MCP协议为开发者提供了一套便捷的脚手架和开发范式让你无需从零开始理解协议的所有细节就能快速构建出符合MCP标准的工具服务器Server。这个服务器启动后就能被任何兼容MCP的AI客户端例如某些集成了MCP的AI IDE或智能体平台发现并调用。2.1 为什么需要自定义MCP服务器你可能会问市面上不是已经有一些现成的MCP工具服务器了吗比如连接GitHub、Notion的。确实但对于企业级、个性化的需求自定义开发是必然选择连接内部系统你的CRM、ERP、内部工单系统、监控平台这些不可能有公开的通用MCP服务器。封装复杂流程将多个API调用、数据处理步骤封装成一个简单的工具比如“一键生成周报并发送给领导”这个工具背后可能调用了数据查询、文本生成、邮件发送等多个服务。安全与权限控制在工具服务器层面实现精细化的权限验证、审计日志确保AI只能在授权范围内操作。性能与稳定性优化针对高并发或延迟敏感的内部服务可以自定义连接池、缓存和重试机制。mcp-custom-dev的价值就在于它降低了构建这样一个标准化、专业化工具服务器的门槛。3. 项目架构与核心设计思路mcp-custom-dev作为一个开发框架其设计思路非常清晰以工具Tool和资源Resource为核心通过标准的MCP协议接口暴露给客户端。整个架构通常遵循客户端-服务器模型但这里的“客户端”是AI模型或AI应用平台。3.1 核心组件拆解一个基于mcp-custom-dev构建的项目通常会包含以下几个核心部分工具定义模块这是项目的核心。你需要在这里用代码声明每一个工具。声明内容包括name: 工具的唯一标识符如get_weather。description: 给AI模型看的自然语言描述这至关重要例如“根据城市名称查询当前天气情况和温度。城市名称应为中文。” 清晰准确的描述能极大提升模型调用的准确性。inputSchema: 定义输入参数的JSON Schema。这是强类型约束确保传入的参数格式正确。例如为city参数定义类型为string并可能提供enum枚举值限制。# 示例性代码结构非项目原码 from mcp_sdk import Tool Tool( namequery_sales_data, description查询指定区域和时间段的销售总额。, inputSchema{ type: object, properties: { region: {type: string, description: 销售区域如华北、华东}, start_date: {type: string, format: date}, end_date: {type: string, format: date} }, required: [region] } ) async def query_sales_data_tool(region: str, start_date: str None, end_date: str None): # 实际的业务逻辑调用内部API、查询数据库等 data await internal_api.get_sales(region, start_date, end_date) return {total_sales: data.total, unit: CNY}资源定义模块类似地定义可供读取的资源。例如你可以定义一个资源其URI为dynamic://internal-dashboard/active-users当AI客户端请求该资源时服务器端动态查询数据库并返回当前活跃用户数的HTML或Markdown片段。from mcp_sdk import Resource Resource( uridynamic://metrics/server_health, description实时服务器健康状态概览 ) async def get_server_health(): cpu, memory, disk await monitor.get_system_metrics() return f ## 服务器健康状态 - **CPU使用率**: {cpu}% - **内存使用率**: {memory}% - **磁盘剩余空间**: {disk}GB 协议适配与服务器启动框架会帮你处理MCP协议的底层通信细节通常基于STDIO或HTTP。你只需要将定义好的工具和资源“注册”到服务器实例然后启动它。服务器启动后会等待兼容MCP的客户端连接。配置与依赖管理项目通常包含配置文件如config.yaml或.env文件用于管理数据库连接字符串、API密钥、服务端口等。依赖管理则确保项目所需的所有Python库如mcpSDK、httpx、pydantic等被正确安装。3.2 设计模式与最佳实践在组织代码时推荐采用分层或模块化的设计工具层集中存放所有工具函数保持函数功能单一、纯净。服务层封装对内部系统或第三方API的调用工具函数调用服务层。数据模型层使用Pydantic等库定义输入输出的数据模型便于验证和序列化。配置层统一管理所有配置项。实操心得描述就是生产力在定义工具的description和参数的description时一定要站在AI模型而不是人类程序员的角度去写。要清晰、无歧义、包含示例。例如“请输入日期”是糟糕的描述“请输入日期格式为YYYY-MM-DD例如2023-10-27”是好的描述。这能直接减少模型调用错误。4. 从零开始构建你的第一个自定义MCP工具理论说了这么多我们来动手实现一个具体的例子构建一个“公司内部知识库问答”工具。这个工具允许AI模型根据用户问题查询我们内部的文档库假设是一个Elasticsearch索引并返回最相关的答案片段。4.1 环境准备与项目初始化首先确保你的开发环境已就绪# 1. 创建项目目录并进入 mkdir my-mcp-knowledge-server cd my-mcp-knowledge-server # 2. 创建虚拟环境推荐 python -m venv venv # Windows: venv\Scripts\activate # Mac/Linux: source venv/bin/activate # 3. 初始化项目并安装核心依赖 # 假设 mcp-custom-dev 或其SDK可以通过pip安装 pip install mcp # 这是MCP的官方Python SDK是构建的基础 pip install pydantic httpx elasticsearch python-dotenv pip install uvloop # 可选用于提升异步性能 # 4. 创建基础项目结构 touch main.py touch tools/ touch config.py touch .env4.2 定义核心工具知识库查询在tools/knowledge.py中我们定义核心工具# tools/knowledge.py import logging from typing import Optional from mcp import Tool from pydantic import BaseModel, Field # 假设我们有一个封装好的ES客户端 from services.elasticsearch_client import es_client logger logging.getLogger(__name__) class KnowledgeQueryInput(BaseModel): 知识库查询工具的输入参数模型 question: str Field( ..., description用户提出的具体问题例如我们的产品退款政策是什么 或 如何配置XX服务器的网络, min_length5 ) max_results: Optional[int] Field( 3, description返回最相关文档片段的最大数量默认为3。, ge1, le10 ) Tool( namequery_company_knowledge_base, description在公司内部知识库中搜索与用户问题相关的文档片段。 知识库涵盖了产品手册、内部流程、技术文档和常见问题解答。 请确保问题描述具体清晰以获得更准确的结果。, # inputSchema 可以由 Pydantic 模型自动生成框架通常支持 ) async def query_company_knowledge_base(input_data: KnowledgeQueryInput) - dict: 执行知识库搜索的核心函数。 question input_data.question max_results input_data.max_results logger.info(f正在知识库中搜索: {question}) # 1. 构建Elasticsearch查询DSL # 这里使用简单的match查询实际生产环境可能要用更复杂的multi-match或BM25调优 search_body { query: { match: { content: question } }, size: max_results, _source: [title, content_snippet, url, last_updated] } try: # 2. 执行搜索 response await es_client.search(indexcompany-knowledge, bodysearch_body) hits response.get(hits, {}).get(hits, []) # 3. 格式化结果使其对AI模型友好 if not hits: return { answer: 在知识库中未找到直接相关的信息。, suggestions: [尝试使用更具体的关键词重新提问。, 或联系相关部门的同事获取帮助。] } formatted_results [] for hit in hits: source hit[_source] formatted_results.append({ 标题: source.get(title, 无标题), 相关片段: source.get(content_snippet, )[:200] ..., # 截取片段 来源链接: source.get(url, #), 相关性得分: round(hit[_score], 2) }) # 4. 返回结构化结果 return { answer: f根据你的问题『{question}』我在知识库中找到以下{len(formatted_results)}条最相关的内容, results: formatted_results, search_performed: True } except Exception as e: logger.error(f知识库查询失败: {e}, exc_infoTrue) # 返回一个清晰的错误信息而不是抛出异常避免服务器崩溃 return { answer: 查询知识库时遇到内部错误暂时无法获取信息。, error_detail: str(e), search_performed: False }4.3 集成工具并启动MCP服务器在main.py中我们将工具集成并启动服务器# main.py import asyncio import logging from contextlib import asynccontextmanager from mcp import Server, StdioServerParameters from mcp.server import NotificationOptions import tools.knowledge # 导入工具模块完成装饰器注册 # 通常框架会有自动发现工具的机制这里演示显式导入 from tools.knowledge import query_company_knowledge_base # 配置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) # 假设我们还有其他工具... # from tools.calendar import schedule_meeting_tool # from tools.jira import create_jira_ticket_tool async def main(): 启动MCP服务器的主函数 # 1. 创建Server实例 # 这里需要根据你使用的具体mcp库的API来调整 server Server( namecompany-internal-tools, version0.1.0, notification_optionsNotificationOptions(), ) # 2. 注册工具 # 方式取决于框架可能是自动扫描也可能是手动注册 # 手动注册示例如果框架支持 server.register_tool(query_company_knowledge_base) # server.register_tool(schedule_meeting_tool) # server.register_tool(create_jira_ticket_tool) # 3. 配置服务器参数使用标准输入输出这是MCP常见方式 server_params StdioServerParameters() logger.info(开始启动 Company Internal Tools MCP 服务器...) # 4. 运行服务器 async with server.run_stdio(server_params) as (read_stream, write_stream): logger.info(MCP服务器已就绪正在等待客户端连接...) # 服务器进入事件循环处理客户端请求 await server.wait_for_disconnect() logger.info(MCP服务器已停止。) if __name__ __main__: asyncio.run(main())4.4 配置与运行创建.env文件存储敏感配置# .env ELASTICSEARCH_HOSTShttps://your-es-cluster.internal:9200 ELASTICSEARCH_USERyour_service_account ELASTICSEARCH_PASSWORDyour_password KNOWLEDGE_INDEXcompany-knowledge创建config.py加载配置# config.py import os from dotenv import load_dotenv load_dotenv() ELASTICSEARCH_CONFIG { hosts: [os.getenv(ELASTICSEARCH_HOSTS)], http_auth: (os.getenv(ELASTICSEARCH_USER), os.getenv(ELASTICSEARCH_PASSWORD)), verify_certs: True, # 生产环境应为True }最后运行你的服务器python main.py服务器启动后它会等待兼容MCP的客户端如 Claude Desktop、Cursor等配置了MCP的AI应用通过标准输入输出与其建立连接。一旦连接客户端就能发现并调用你定义的query_company_knowledge_base工具了。注意事项错误处理与超时在工具函数中务必要有完善的错误处理try-except和超时控制。AI客户端调用工具时如果工具长时间无响应或崩溃会导致整个交互体验变差。对于网络请求如调用ES务必设置合理的超时时间并使用asyncio.wait_for或httpx.Timeout进行控制。5. 高级特性与生产级考量当你掌握了基础工具开发后为了将其用于生产环境还需要考虑以下几个关键方面5.1 工具的动态注册与发现在更复杂的场景中你的工具列表可能不是静态的而是根据配置、数据库内容或用户权限动态变化的。mcp-custom-dev类框架通常支持动态工具注册。你可以在服务器启动后根据条件向服务器实例动态添加或移除工具。例如根据当前登录用户的角色决定是否注册“删除数据库”这类高危工具async def initialize_tools_based_on_user(user_role: str): if user_role admin: server.register_tool(dangerous_admin_tool) # 注册其他通用工具...5.2 认证、授权与审计AAA这是企业级应用的核心。MCP协议本身可能不直接处理认证但这部分必须在你的工具服务器中实现。认证客户端连接时如何验证其身份可以通过启动参数传递令牌或在首次握手时进行认证。一种常见模式是使用API密钥或OAuth2.0客户端凭证。授权即使认证通过用户/客户端是否有权调用某个工具这需要在工具函数内部或通过装饰器进行权限检查。审计谁在什么时候调用了什么工具输入输出是什么注意脱敏这些日志对于安全追溯和问题排查至关重要。可以在工具装饰器或一个公共的拦截器Middleware中实现这些逻辑def audit_log(tool_name): def decorator(func): wraps(func) async def wrapper(*args, **kwargs): user get_current_user() logger.info(fAUDIT: User {user} invoking tool {tool_name} with args {kwargs}) start time.time() try: result await func(*args, **kwargs) logger.info(fAUDIT: Tool {tool_name} executed successfully in {time.time()-start:.2f}s) # 注意记录结果时可能需要脱敏避免日志泄露敏感数据 return result except Exception as e: logger.error(fAUDIT: Tool {tool_name} failed with error: {e}) raise return wrapper return decorator # 在工具上使用 Tool(...) audit_log(query_sales_data) async def query_sales_data_tool(...): ...5.3 性能优化异步、缓存与连接池异步AsyncMCP服务器和工具函数强烈建议使用异步编程asyncio。这能保证在等待I/O如网络请求、数据库查询时服务器可以处理其他请求提高并发能力。确保你使用的所有客户端库如httpx.AsyncClient,aiopg,asyncpg都支持异步。缓存对于查询类、结果变化不频繁的工具引入缓存可以极大提升响应速度并降低后端压力。可以使用aiocache或redis配合异步客户端。from aiocache import cached cached(ttl300) # 缓存5分钟 Tool(...) async def get_department_list(...): # 昂贵的数据库查询 ...连接池对于数据库、Elasticsearch等外部服务务必使用连接池而不是为每次调用创建新连接。5.4 测试策略单元测试与集成测试为MCP工具编写测试至关重要。单元测试直接测试工具函数本身模拟mock所有外部依赖如ES客户端、API调用。确保各种输入正常、边界、异常下函数行为符合预期。集成测试/端到端测试启动一个测试版的MCP服务器使用一个MCP客户端测试库或模拟客户端实际连接并调用工具验证整个流程。这能发现协议层或集成上的问题。6. 常见问题与实战排坑记录在实际开发和部署mcp-custom-dev类项目的过程中我踩过不少坑这里总结几个典型问题和解决方案问题现象可能原因排查步骤与解决方案AI客户端无法发现工具1. MCP服务器未正确启动或通信协议不一致。2. 工具定义不符合MCP协议规范如缺少必需字段。3. 客户端配置的服务器路径或参数错误。1. 检查服务器日志确认已成功启动并监听。2. 使用mcpSDK自带的CLI工具如mcp dev或编写一个简单的测试脚本来连接服务器列出工具验证服务器本身是否正常。3. 仔细核对客户端如Claude Desktop的配置JSON文件确保command和args指向正确的Python解释器和脚本路径。工具调用超时或无响应1. 工具函数内部有同步阻塞操作如requests.get而非httpx.AsyncClient。2. 工具函数陷入死循环或等待资源死锁。3. 网络问题导致调用外部服务超时。1.将所有I/O操作改为异步。这是最常见的原因。检查代码确保没有使用requests,subprocess.run等同步库。2. 在工具函数中添加超时控制async with asyncio.timeout(30): ...。3. 增加详细的日志定位卡在哪一步。对外部服务调用配置合理的超时和重试。工具返回结果但AI模型“不理解”1. 工具返回的数据结构过于复杂或非结构化。2. 工具描述description不清晰导致模型误用。3. 错误信息不够友好。1.返回结构化的、简洁的JSON对象。优先使用字符串、数字、布尔值、简单数组和对象。避免返回多层嵌套的复杂对象或自定义类实例。2.优化工具和参数的描述。用自然语言明确说明工具的用途、限制和参数示例。这是提示词工程的一部分。3. 错误时返回一个包含error或message字段的JSON对象而不是抛出未处理的异常。权限错误或认证失败1. 环境变量或配置文件未正确加载。2. 密钥过期或权限不足。3. 服务器端认证逻辑有bug。1. 在服务器启动时打印关键配置脱敏后以确认加载成功。2. 实现一个简单的“健康检查”工具或接口测试对外部服务的连接状态。3. 认证逻辑单独编写单元测试模拟各种令牌场景。服务器内存泄漏或CPU占用高1. 工具函数内有资源未释放如文件句柄、数据库连接。2. 缓存策略不当缓存无限增长。3. 某个工具计算过于密集阻塞事件循环。1. 使用async with确保客户端资源正确关闭。2. 为缓存设置大小限制和过期时间TTL。3. 对于CPU密集型任务考虑使用asyncio.to_thread或ProcessPoolExecutor将其放到单独线程/进程中执行避免阻塞主事件循环。一个关键的避坑技巧善用“模拟客户端”进行开发调试。在开发初期不要急于对接复杂的AI客户端。可以写一个简单的Python脚本模拟MCP客户端与你的服务器通信直接调用工具并打印结果。这能让你快速验证工具逻辑是否正确而不受客户端复杂性的干扰。许多MCP SDK都提供了用于测试的客户端库。7. 扩展思路不止于工具服务器当你熟练使用mcp-custom-dev模式后可以将其思路扩展到更广阔的领域构建工具市场/仓库将公司内各部门开发的优质MCP工具标准化、版本化形成一个内部工具市场。AI助手可以根据用户需求动态加载和组合不同的工具。实现复杂工作流单个工具能力有限但你可以开发一个“工作流编排”工具。这个工具本身接受一个复杂目标然后在内部按顺序或条件调用其他多个基础工具最终返回整合结果。这相当于让AI具备了执行多步骤任务的能力。与低代码平台结合将MCP工具作为低代码平台的后端“积木”。业务人员通过界面配置流程实际上生成的是对一系列MCP工具的调用序列。监控与可观测性为你的MCP服务器集成监控如Prometheus指标、分布式追踪如OpenTelemetry。监控每个工具的调用次数、延迟、错误率这对于维护一个稳定的AI工具生态至关重要。回过头看mcp-custom-dev这类项目代表的是一种范式转变从“让人去适应系统”到“让系统AI来服务人”。它通过一个轻量级、标准化的协议将企业内部浩瀚的数字能力封装成AI可以理解和操作的“技能”。开发这样的工具服务器技术难点并不在于协议本身而在于如何设计出语义清晰、边界明确、安全可靠的工具这需要开发者对业务有深刻理解并具备良好的软件工程和提示词工程能力。我的体会是成功的AI工具集成30%在技术70%在设计和沟通——既包括与AI模型的“沟通”工具描述也包括与业务方的沟通明确需求和边界。
