1. 项目概述与核心价值最近在折腾AI应用开发特别是想给大语言模型LLM装上更强大的“工具箱”让它们能直接操作外部系统和数据。在这个过程中我反复遇到了一个痛点如何高效、安全地让AI调用各种工具Tools传统的做法要么是把所有工具一股脑塞给模型导致上下文窗口爆炸、成本飙升要么是手动编写复杂的逻辑来判断何时调用哪个工具费时费力且难以维护。直到我深入研究了Model Context Protocol并发现了currenjin/mcp-context-template这个项目才感觉找到了一个优雅的解决方案。这个项目本质上是一个为MCPModel Context Protocol设计的上下文管理模板它帮你搭建了一个结构化的“工具调用脚手架”。简单来说MCP是一个新兴的协议旨在标准化LLM与外部工具如数据库、API、文件系统之间的交互方式。而mcp-context-template则是一个具体的实现模板它预先定义了一套最佳实践教你如何组织工具、管理上下文、处理权限从而让你的AI应用能更智能、更可控地使用外部能力。它解决的不仅仅是“能调用”的问题更是“如何高效、安全、可维护地调用”的问题。无论你是想构建一个能分析数据库的AI助手还是一个能操作云资源的智能体这个模板都能提供一个坚实的起点帮你省去大量重复的底层架构工作。2. MCP协议核心思想与模板设计解析2.1 为什么需要MCP从“硬编码”到“协议化”的演进在MCP出现之前我们给LLM集成外部功能常见的方式是“硬编码”或使用特定框架的SDK。比如你想让ChatGPT能查询天气就需要在后台写一个get_weather函数然后通过OpenAI的Function Calling机制暴露给它。这种方式有几个明显的弊端工具定义碎片化每个项目、每个框架如LangChain、LlamaIndex都有自己定义工具的方式缺乏统一标准。从一个项目迁移到另一个项目工具层几乎要重写。上下文管理混乱工具的描述、参数schema、执行结果都混杂在主要的对话上下文中。工具一多宝贵的Token就被大量工具描述占用影响核心对话质量。安全与权限控制薄弱通常是在工具执行函数内部进行权限判断缺乏统一的、声明式的权限管理机制。开发体验不友好开发者需要关心网络通信、序列化、错误处理等底层细节不能专注于工具本身的业务逻辑。MCP的提出正是为了应对这些问题。它定义了一套标准的JSON-RPC over SSE/HTTP协议将工具Tools、资源Resources如只读数据的提供者Server与消费者Client通常是LLM应用解耦。currenjin/mcp-context-template这个项目就是基于MCP协议为你预先搭建了一个Server端的样板工程。它采用了“上下文模板”的设计理念即不是提供一个万能工具箱而是提供一个如何组织和管理“工具上下文”的范式。2.2 模板的核心架构分层与模块化这个模板的代码结构清晰地体现了其设计思想。它通常包含以下几个核心层次协议层基于MCP的SDK例如modelcontextprotocol/sdk实现标准的ListTools、CallTool、ReadResource等请求处理。这一层是固定的由模板帮你实现你一般无需修改。上下文管理层这是模板的核心价值所在。它定义了如何将你的业务工具分类、组装并注入到MCP Server中。例如可能会有一个ContextBuilder或ToolRegistry类负责收集所有工具模块并统一设置它们的元信息名称、描述、参数schema。工具模块层你的业务工具按领域被划分到不同的模块中。例如database_tools.py包含所有数据库操作工具file_system_tools.py包含文件操作工具。每个工具都是一个独立的函数并附带标准的描述性注解或Pydantic模型来定义输入输出。配置与安全层通过配置文件或环境变量来管理Server的启动参数、允许访问的工具列表、权限等级等。模板可能会集成基础的认证或权限检查钩子。这种架构的好处是“关注点分离”。作为开发者你只需要在“工具模块层”专心实现你的业务逻辑函数上下文管理层会帮你自动注册和暴露它们协议层则负责处理与LLM Client的标准通信。当需要增加一个新工具时你只需在相应的模块文件中添加一个函数并在上下文管理层注册即可无需触碰网络或协议相关的代码。注意模板的具体实现可能因版本和作者偏好有所不同但“分层”和“基于协议”的核心思想是不变的。理解这一点比记忆具体文件结构更重要。3. 基于模板的实战构建一个智能数据查询助手假设我们要构建一个AI助手它能够根据自然语言查询对公司内部的销售数据库进行安全的数据分析和汇总。我们将使用mcp-context-template作为基础。3.1 环境准备与项目初始化首先克隆模板仓库并安装依赖。模板通常是基于Node.jsTypeScript或Python实现的这里我们以Python版本为例进行说明。# 克隆模板仓库 git clone https://github.com/currenjin/mcp-context-template.git my-mcp-data-assistant cd my-mcp-data-assistant # 安装依赖 (假设使用 poetry 或 pip) pip install -r requirements.txt # 通常依赖会包括 mcp SDK, pydantic, sqlalchemy 等接下来我们需要审视项目结构。关键文件通常包括server.py或main.py: 应用入口启动MCP Server。context/或tools/目录存放工具模块。config.yaml或.env配置文件。models/可能包含Pydantic模型用于定义工具输入输出的数据结构。3.2 定义数据模型与核心工具我们的目标是查询销售数据所以先定义核心的数据模型。在models/sales.py中from pydantic import BaseModel, Field from datetime import date from typing import Optional class SalesQuery(BaseModel): 销售数据查询参数 start_date: date Field(..., description查询开始日期格式YYYY-MM-DD) end_date: date Field(..., description查询结束日期格式YYYY-MM-DD) region: Optional[str] Field(None, description销售区域如‘North America’不指定则查询所有区域) product_category: Optional[str] Field(None, description产品类别如‘Electronics’) class SalesSummary(BaseModel): 销售数据汇总结果 total_revenue: float total_units_sold: int average_order_value: float top_products: list[dict] # 例如 [{product: Laptop, revenue: 50000}]然后在tools/database_tools.py中实现核心查询工具。这里是一个极易踩坑的点数据库连接的管理。import sqlalchemy as sa from sqlalchemy.orm import sessionmaker from models.sales import SalesQuery, SalesSummary from mcp import Tool # 数据库引擎应在应用启动时创建全局共享 # 切忌在每个工具调用内部创建新引擎会导致连接池耗尽 engine None SessionLocal None def init_db_connection(connection_string: str): 初始化数据库连接应在server启动时调用一次 global engine, SessionLocal engine sa.create_engine(connection_string, pool_pre_pingTrue, pool_recycle3600) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) # 使用MCP SDK的装饰器或类来声明一个工具 Tool( namequery_sales_summary, description根据时间范围、区域和产品类别查询销售数据汇总。, input_modelSalesQuery, output_modelSalesSummary ) async def query_sales_summary_tool(query: SalesQuery) - SalesSummary: 执行销售汇总查询。 注意此函数应专注于业务逻辑SQL拼接和权限验证最好在更底层处理。 if not engine: raise RuntimeError(数据库连接未初始化) # 构建动态SQL查询示例使用SQLAlchemy Core with SessionLocal() as session: # 基础查询 stmt sa.text( SELECT SUM(revenue) as total_revenue, SUM(quantity) as total_units, AVG(revenue) as avg_order_value FROM sales_records WHERE sale_date BETWEEN :start_date AND :end_date ).bindparams(start_datequery.start_date, end_datequery.end_date) # 动态添加过滤条件 if query.region: stmt stmt.where(sa.column(region) query.region) if query.product_category: stmt stmt.where(sa.column(product_category) query.product_category) result session.execute(stmt).fetchone() # 查询畅销产品另一个简化查询 top_prod_stmt sa.text( SELECT product_name, SUM(revenue) as prod_revenue FROM sales_records WHERE sale_date BETWEEN :start_date AND :end_date GROUP BY product_name ORDER BY prod_revenue DESC LIMIT 5 ).bindparams(start_datequery.start_date, end_datequery.end_date) top_products session.execute(top_prod_stmt).fetchall() # 构造返回结果 return SalesSummary( total_revenueresult.total_revenue or 0.0, total_units_soldresult.total_units or 0, average_order_valueresult.avg_order_value or 0.0, top_products[{product: row.product_name, revenue: row.prod_revenue} for row in top_products] )实操心得数据库工具最怕连接泄漏和SQL注入。模板项目通常不会强制你如何管理连接但最佳实践是使用连接池如SQLAlchemy的create_engine默认就带连接池。会话生命周期管理在每个工具函数内部获取和关闭会话确保会话不会跨请求泄露。使用参数化查询绝对不要用字符串拼接f-string来构造SQL必须像上面一样使用:param语法或SQLAlchemy的表达式语言从根源上杜绝SQL注入。初始化分离像init_db_connection这样的初始化操作应该放在Server启动脚本中而不是工具模块里。3.3 集成工具并配置上下文接下来我们需要在模板的上下文管理层注册这个工具。通常在context/__init__.py或tool_registry.py中from tools.database_tools import query_sales_summary_tool from mcp import Server class MyContext: def __init__(self): self.tools [query_sales_summary_tool] # 未来可以在这里加载更多工具如 file_tools, api_tools 等 async def get_tools_for_session(self, session_metadata: dict) - list: 这是一个关键钩子可以根据会话元数据如用户身份动态返回可用的工具列表。 实现基础的权限控制。 # 示例假设metadata中包含用户角色 user_role session_metadata.get(role, guest) if user_role analyst: return self.tools # 分析师拥有所有工具 elif user_role viewer: # 只返回只读工具过滤掉写操作工具 return [t for t in self.tools if t.name in [query_sales_summary]] else: return [] # 无权限用户 # 在server.py中 from context import MyContext import asyncio from mcp import StdioServerTransport, Server from mcp.server import Server async def main(): # 1. 初始化业务上下文 context MyContext() # 初始化数据库连接读取配置 init_db_connection(os.getenv(DATABASE_URL)) # 2. 创建MCP Server实例 server Server() # 3. 注册工具列表提供方法 server.list_tools() async def handle_list_tools(session_metadata: dict) - list: # 调用上下文的权限过滤方法 return await context.get_tools_for_session(session_metadata) # 4. 注册工具调用处理器 server.call_tool() async def handle_call_tool(name: str, arguments: dict, session_metadata: dict) - dict: # 根据工具名找到对应的工具函数 available_tools await context.get_tools_for_session(session_metadata) tool_map {t.name: t for t in available_tools} if name not in tool_map: raise ValueError(fTool not found or not authorized: {name}) tool tool_map[name] # 调用工具函数传入参数 result await tool(**arguments) # 将结果转换为字典如果工具返回的是Pydantic模型需.dict() return result.dict() if hasattr(result, dict) else result # 5. 启动Server这里以Stdio为例用于CLI集成 async with StdioServerTransport() as transport: await server.run(transport) if __name__ __main__: asyncio.run(main())这个流程展示了模板如何将你的业务工具与MCP协议桥接起来。MyContext类是你的“工具管家”负责工具的收集和基于会话的过滤。server.py中的装饰器函数则是标准的MCP请求处理器。4. 权限控制与安全增强实践MCP协议本身不强制规定权限模型这给了模板和开发者极大的灵活性。mcp-context-template通常会在上下文管理层预留钩子就像上面的get_tools_for_session。我们可以在此基础上进行强化。4.1 基于角色的工具访问控制RBAC我们可以设计一个更完善的权限系统。在config/permissions.yaml中定义角色与工具的映射roles: viewer: tools: - query_sales_summary - get_report analyst: inherits: viewer tools: - execute_data_pipeline - update_dataset admin: inherits: analyst tools: - manage_users - shutdown_server然后在上下文管理中读取这个配置import yaml from pathlib import Path class RBACContext: def __init__(self, config_path: Path): with open(config_path) as f: self.config yaml.safe_load(f) self.all_tools self._load_all_tools() # 加载所有工具对象 async def get_tools_for_session(self, session_metadata: dict) - list: user_role session_metadata.get(role, viewer) role_config self.config[roles].get(user_role, {}) allowed_tool_names set(role_config.get(tools, [])) # 处理继承 if inherits in role_config: parent_tools self.config[roles][role_config[inherits]].get(tools, []) allowed_tool_names.update(parent_tools) return [t for t in self.all_tools if t.name in allowed_tool_names]4.2 工具级参数校验与操作审计除了谁能调用工具我们还需要关心工具被调用时做了什么。可以在工具调用处理器中加入审计日志和更细粒度的参数校验。import logging logger logging.getLogger(__name__) server.call_tool() async def handle_call_tool(name: str, arguments: dict, session_metadata: dict) - dict: user_id session_metadata.get(user_id, anonymous) # 1. 权限检查同上 available_tools await context.get_tools_for_session(session_metadata) tool_map {t.name: t for t in available_tools} if name not in tool_map: logger.warning(fUnauthorized tool access attempt: user{user_id}, tool{name}) raise PermissionError(fTool {name} is not authorized.) tool tool_map[name] # 2. 审计日志记录谁在什么时候调用了什么工具参数是什么注意过滤敏感参数 safe_args arguments.copy() safe_args.pop(password, None) # 示例过滤密码字段 logger.info(fTool invoked: user{user_id}, tool{name}, args{safe_args}) # 3. 额外的参数业务校验例如查询时间范围不能超过365天 if name query_sales_summary: from datetime import date, timedelta start date.fromisoformat(arguments[start_date]) end date.fromisoformat(arguments[end_date]) if (end - start).days 365: raise ValueError(查询时间范围不能超过365天。) # 4. 执行工具 try: result await tool(**arguments) logger.info(fTool succeeded: user{user_id}, tool{name}) return result.dict() if hasattr(result, dict) else result except Exception as e: logger.error(fTool execution failed: user{user_id}, tool{name}, error{str(e)}) raise注意事项审计日志非常重要尤其是在生产环境。但要注意日志中不能记录敏感信息如完整的SQL语句、API密钥、个人身份信息。建议对参数进行清洗或只记录工具名和部分非敏感元数据。5. 客户端集成与调试技巧Server搭建好后我们需要一个MCP Client来测试它。目前许多AI应用框架和平台正在逐步支持MCP Client。5.1 使用MCP Inspector进行调试最直接的调试方法是使用官方或社区的MCP Inspector工具。它是一个独立的桌面应用或CLI工具可以连接到你的MCP Server列出所有可用的工具和资源并手动发起调用观察请求和响应。# 假设你的Server通过stdio启动 npx modelcontextprotocol/inspector python server.py运行后Inspector会打开一个图形界面或提供命令行交互让你可以浏览工具、填写参数并执行是开发和调试阶段不可或缺的利器。5.2 集成到Claude Desktop或Cursor等AI应用一些先进的AI应用已经开始支持MCP。例如在Claude Desktop中你可以通过编辑配置文件来添加自定义的MCP Server。找到Claude Desktop的配置文件夹。编辑claude_desktop_config.json添加你的Server配置{ mcpServers: { my-data-assistant: { command: python, args: [/absolute/path/to/your/server.py], env: { DATABASE_URL: your_db_connection_string } } } }重启Claude Desktop。之后Claude AI在对话中就能“看到”并使用你定义的query_sales_summary等工具了你可以直接说“帮我分析一下上季度的北美电子产品销售情况”。5.3 在自定义AI应用中集成如果你在构建自己的AI应用可以使用MCP Client SDK来连接你的Server。import asyncio from mcp import Client, StdioClientTransport async def main(): # 启动Server进程并创建Client传输层 transport await StdioClientTransport.create([python, server.py]) client Client(transport) await client.initialize() # 列出可用工具 tools await client.list_tools() print(Available tools:, [t.name for t in tools.tools]) # 调用工具 result await client.call_tool( namequery_sales_summary, arguments{ start_date: 2024-01-01, end_date: 2024-03-31, region: North America } ) print(Query result:, result) await client.finalize() asyncio.run(main())6. 性能优化与生产部署考量当工具数量和复杂度增长后性能和维护性成为关键。6.1 工具懒加载与缓存不是所有工具都需要在Server启动时就全部加载进内存。对于一些初始化成本高、使用频率低的工具可以采用懒加载。class LazyTool: def __init__(self, name, loader_func): self.name name self._loader_func loader_func self._tool_instance None property async def tool(self): if self._tool_instance is None: self._tool_instance await self._loader_func() return self._tool_instance async def __call__(self, **kwargs): inst await self.tool return await inst(**kwargs) # 在上下文中注册懒加载工具 async def load_heavy_ml_tool(): # 模拟加载一个大型机器学习模型 import torch model torch.load(large_model.pt) return model.predict context.tools.append(LazyTool(namepredict_with_ml, loader_funcload_heavy_ml_tool))同时对于查询类工具的结果如果参数相同可以考虑在短时间内缓存结果避免重复查询数据库。from functools import lru_cache import asyncio # 注意缓存需要根据业务场景设置合理的过期时间和大小 lru_cache(maxsize128) async def cached_query_sales_summary(start_date_str: str, end_date_str: str, region: str None): # 实际的数据库查询逻辑 pass # 在工具函数中调用缓存版本 async def query_sales_summary_tool_cached(query: SalesQuery): key (query.start_date.isoformat(), query.end_date.isoformat(), query.region) return await cached_query_sales_summary(*key)6.2 错误处理与健壮性生产环境的工具必须健壮。模板应鼓励全面的错误处理。输入验证充分利用Pydantic模型的验证能力。对于更复杂的业务规则在工具函数开头进行校验。超时控制为可能长时间运行的工具设置超时。import asyncio async def call_tool_with_timeout(tool_func, args, timeout30): try: return await asyncio.wait_for(tool_func(**args), timeouttimeout) except asyncio.TimeoutError: logger.error(fTool {tool_func.__name__} timed out after {timeout}s) raise TimeoutError(Tool execution timed out.)优雅降级当依赖的外部服务如某个API不可用时工具应返回有意义的错误信息或尝试使用备用方案而不是让整个Server崩溃。6.3 监控与可观测性除了审计日志还需要添加性能指标监控。可以集成像Prometheus这样的监控系统。from prometheus_client import Counter, Histogram TOOL_CALL_COUNT Counter(mcp_tool_calls_total, Total tool calls, [tool_name, status]) TOOL_CALL_DURATION Histogram(mcp_tool_call_duration_seconds, Tool call duration, [tool_name]) server.call_tool() async def handle_call_tool_with_metrics(name: str, arguments: dict, session_metadata: dict) - dict: with TOOL_CALL_DURATION.labels(tool_namename).time(): try: result await call_tool_impl(name, arguments, session_metadata) # 原有的调用逻辑 TOOL_CALL_COUNT.labels(tool_namename, statussuccess).inc() return result except Exception as e: TOOL_CALL_COUNT.labels(tool_namename, statuserror).inc() raise这样你就能清晰地看到每个工具的调用频率、成功率和延迟分布便于发现性能瓶颈和异常。7. 常见问题排查与模板定制7.1 工具未在客户端显示检查点1Server是否正常启动并输出了初始化日志确保没有端口冲突或依赖缺失。检查点2list_tools方法是否正确返回了工具列表在Inspector中查看或添加调试日志。检查点3权限过滤是否过于严格检查get_tools_for_session逻辑确保当前会话的元数据能通过过滤。临时可以注释掉权限逻辑进行测试。检查点4工具描述格式是否正确MCP对工具描述的schema有要求确保name、description、inputSchema字段完整且格式正确。使用Pydantic模型能自动生成合规的JSON Schema。7.2 工具调用失败返回参数错误检查点1客户端传入的参数格式是否与schema匹配在Inspector中仔细对照工具定义的输入模型。日期字段是否是字符串格式枚举值是否正确检查点2Pydantic模型验证是否失败在工具函数内部或调用前手动用模型验证一下输入SalesQuery(**arguments)。检查点3工具函数本身是否有未处理的异常查看Server端的错误日志。可能是数据库连接失败、网络超时等。7.3 性能瓶颈分析使用监控指标如上节所述通过指标定位是哪个工具慢。数据库查询分析对于数据库工具检查SQL是否使用了索引。可以在工具函数中记录或输出执行的SQL语句生产环境需谨慎用EXPLAIN分析。并发问题如果工具涉及共享状态如全局缓存、文件写入需考虑线程安全或异步锁。asyncio.Lock可以用于保护异步环境下的临界区。7.4 如何扩展模板添加新类型的工具模板是起点不是终点。添加新工具遵循固定模式在tools/目录下创建新模块如external_api_tools.py。实现具体的工具函数用Tool装饰器装饰并定义好输入输出模型。在上下文管理类中注册将新工具添加到MyContext.tools列表中。考虑权限更新权限配置如permissions.yaml决定哪些角色可以访问新工具。编写测试为新的工具函数编写单元测试模拟API调用。例如添加一个调用天气预报API的工具# tools/weather_tools.py import httpx from pydantic import BaseModel from mcp import Tool class WeatherQuery(BaseModel): city: str country_code: str US Tool(nameget_weather, description获取指定城市的当前天气, input_modelWeatherQuery) async def get_weather_tool(query: WeatherQuery) - str: async with httpx.AsyncClient() as client: # 使用一个模拟的天气API resp await client.get(fhttps://api.weather.example.com?city{query.city}country{query.country_code}) resp.raise_for_status() data resp.json() return fThe weather in {query.city} is {data[condition]} with a temperature of {data[temp_c]}°C.将这个工具导入并注册到上下文你的AI助手立刻就拥有了查询天气的能力。这种模块化的扩展方式正是mcp-context-template带来的最大便利。
基于MCP协议构建AI工具调用框架：从原理到实战部署
发布时间：2026/5/16 0:47:09
1. 项目概述与核心价值最近在折腾AI应用开发特别是想给大语言模型LLM装上更强大的“工具箱”让它们能直接操作外部系统和数据。在这个过程中我反复遇到了一个痛点如何高效、安全地让AI调用各种工具Tools传统的做法要么是把所有工具一股脑塞给模型导致上下文窗口爆炸、成本飙升要么是手动编写复杂的逻辑来判断何时调用哪个工具费时费力且难以维护。直到我深入研究了Model Context Protocol并发现了currenjin/mcp-context-template这个项目才感觉找到了一个优雅的解决方案。这个项目本质上是一个为MCPModel Context Protocol设计的上下文管理模板它帮你搭建了一个结构化的“工具调用脚手架”。简单来说MCP是一个新兴的协议旨在标准化LLM与外部工具如数据库、API、文件系统之间的交互方式。而mcp-context-template则是一个具体的实现模板它预先定义了一套最佳实践教你如何组织工具、管理上下文、处理权限从而让你的AI应用能更智能、更可控地使用外部能力。它解决的不仅仅是“能调用”的问题更是“如何高效、安全、可维护地调用”的问题。无论你是想构建一个能分析数据库的AI助手还是一个能操作云资源的智能体这个模板都能提供一个坚实的起点帮你省去大量重复的底层架构工作。2. MCP协议核心思想与模板设计解析2.1 为什么需要MCP从“硬编码”到“协议化”的演进在MCP出现之前我们给LLM集成外部功能常见的方式是“硬编码”或使用特定框架的SDK。比如你想让ChatGPT能查询天气就需要在后台写一个get_weather函数然后通过OpenAI的Function Calling机制暴露给它。这种方式有几个明显的弊端工具定义碎片化每个项目、每个框架如LangChain、LlamaIndex都有自己定义工具的方式缺乏统一标准。从一个项目迁移到另一个项目工具层几乎要重写。上下文管理混乱工具的描述、参数schema、执行结果都混杂在主要的对话上下文中。工具一多宝贵的Token就被大量工具描述占用影响核心对话质量。安全与权限控制薄弱通常是在工具执行函数内部进行权限判断缺乏统一的、声明式的权限管理机制。开发体验不友好开发者需要关心网络通信、序列化、错误处理等底层细节不能专注于工具本身的业务逻辑。MCP的提出正是为了应对这些问题。它定义了一套标准的JSON-RPC over SSE/HTTP协议将工具Tools、资源Resources如只读数据的提供者Server与消费者Client通常是LLM应用解耦。currenjin/mcp-context-template这个项目就是基于MCP协议为你预先搭建了一个Server端的样板工程。它采用了“上下文模板”的设计理念即不是提供一个万能工具箱而是提供一个如何组织和管理“工具上下文”的范式。2.2 模板的核心架构分层与模块化这个模板的代码结构清晰地体现了其设计思想。它通常包含以下几个核心层次协议层基于MCP的SDK例如modelcontextprotocol/sdk实现标准的ListTools、CallTool、ReadResource等请求处理。这一层是固定的由模板帮你实现你一般无需修改。上下文管理层这是模板的核心价值所在。它定义了如何将你的业务工具分类、组装并注入到MCP Server中。例如可能会有一个ContextBuilder或ToolRegistry类负责收集所有工具模块并统一设置它们的元信息名称、描述、参数schema。工具模块层你的业务工具按领域被划分到不同的模块中。例如database_tools.py包含所有数据库操作工具file_system_tools.py包含文件操作工具。每个工具都是一个独立的函数并附带标准的描述性注解或Pydantic模型来定义输入输出。配置与安全层通过配置文件或环境变量来管理Server的启动参数、允许访问的工具列表、权限等级等。模板可能会集成基础的认证或权限检查钩子。这种架构的好处是“关注点分离”。作为开发者你只需要在“工具模块层”专心实现你的业务逻辑函数上下文管理层会帮你自动注册和暴露它们协议层则负责处理与LLM Client的标准通信。当需要增加一个新工具时你只需在相应的模块文件中添加一个函数并在上下文管理层注册即可无需触碰网络或协议相关的代码。注意模板的具体实现可能因版本和作者偏好有所不同但“分层”和“基于协议”的核心思想是不变的。理解这一点比记忆具体文件结构更重要。3. 基于模板的实战构建一个智能数据查询助手假设我们要构建一个AI助手它能够根据自然语言查询对公司内部的销售数据库进行安全的数据分析和汇总。我们将使用mcp-context-template作为基础。3.1 环境准备与项目初始化首先克隆模板仓库并安装依赖。模板通常是基于Node.jsTypeScript或Python实现的这里我们以Python版本为例进行说明。# 克隆模板仓库 git clone https://github.com/currenjin/mcp-context-template.git my-mcp-data-assistant cd my-mcp-data-assistant # 安装依赖 (假设使用 poetry 或 pip) pip install -r requirements.txt # 通常依赖会包括 mcp SDK, pydantic, sqlalchemy 等接下来我们需要审视项目结构。关键文件通常包括server.py或main.py: 应用入口启动MCP Server。context/或tools/目录存放工具模块。config.yaml或.env配置文件。models/可能包含Pydantic模型用于定义工具输入输出的数据结构。3.2 定义数据模型与核心工具我们的目标是查询销售数据所以先定义核心的数据模型。在models/sales.py中from pydantic import BaseModel, Field from datetime import date from typing import Optional class SalesQuery(BaseModel): 销售数据查询参数 start_date: date Field(..., description查询开始日期格式YYYY-MM-DD) end_date: date Field(..., description查询结束日期格式YYYY-MM-DD) region: Optional[str] Field(None, description销售区域如‘North America’不指定则查询所有区域) product_category: Optional[str] Field(None, description产品类别如‘Electronics’) class SalesSummary(BaseModel): 销售数据汇总结果 total_revenue: float total_units_sold: int average_order_value: float top_products: list[dict] # 例如 [{product: Laptop, revenue: 50000}]然后在tools/database_tools.py中实现核心查询工具。这里是一个极易踩坑的点数据库连接的管理。import sqlalchemy as sa from sqlalchemy.orm import sessionmaker from models.sales import SalesQuery, SalesSummary from mcp import Tool # 数据库引擎应在应用启动时创建全局共享 # 切忌在每个工具调用内部创建新引擎会导致连接池耗尽 engine None SessionLocal None def init_db_connection(connection_string: str): 初始化数据库连接应在server启动时调用一次 global engine, SessionLocal engine sa.create_engine(connection_string, pool_pre_pingTrue, pool_recycle3600) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) # 使用MCP SDK的装饰器或类来声明一个工具 Tool( namequery_sales_summary, description根据时间范围、区域和产品类别查询销售数据汇总。, input_modelSalesQuery, output_modelSalesSummary ) async def query_sales_summary_tool(query: SalesQuery) - SalesSummary: 执行销售汇总查询。 注意此函数应专注于业务逻辑SQL拼接和权限验证最好在更底层处理。 if not engine: raise RuntimeError(数据库连接未初始化) # 构建动态SQL查询示例使用SQLAlchemy Core with SessionLocal() as session: # 基础查询 stmt sa.text( SELECT SUM(revenue) as total_revenue, SUM(quantity) as total_units, AVG(revenue) as avg_order_value FROM sales_records WHERE sale_date BETWEEN :start_date AND :end_date ).bindparams(start_datequery.start_date, end_datequery.end_date) # 动态添加过滤条件 if query.region: stmt stmt.where(sa.column(region) query.region) if query.product_category: stmt stmt.where(sa.column(product_category) query.product_category) result session.execute(stmt).fetchone() # 查询畅销产品另一个简化查询 top_prod_stmt sa.text( SELECT product_name, SUM(revenue) as prod_revenue FROM sales_records WHERE sale_date BETWEEN :start_date AND :end_date GROUP BY product_name ORDER BY prod_revenue DESC LIMIT 5 ).bindparams(start_datequery.start_date, end_datequery.end_date) top_products session.execute(top_prod_stmt).fetchall() # 构造返回结果 return SalesSummary( total_revenueresult.total_revenue or 0.0, total_units_soldresult.total_units or 0, average_order_valueresult.avg_order_value or 0.0, top_products[{product: row.product_name, revenue: row.prod_revenue} for row in top_products] )实操心得数据库工具最怕连接泄漏和SQL注入。模板项目通常不会强制你如何管理连接但最佳实践是使用连接池如SQLAlchemy的create_engine默认就带连接池。会话生命周期管理在每个工具函数内部获取和关闭会话确保会话不会跨请求泄露。使用参数化查询绝对不要用字符串拼接f-string来构造SQL必须像上面一样使用:param语法或SQLAlchemy的表达式语言从根源上杜绝SQL注入。初始化分离像init_db_connection这样的初始化操作应该放在Server启动脚本中而不是工具模块里。3.3 集成工具并配置上下文接下来我们需要在模板的上下文管理层注册这个工具。通常在context/__init__.py或tool_registry.py中from tools.database_tools import query_sales_summary_tool from mcp import Server class MyContext: def __init__(self): self.tools [query_sales_summary_tool] # 未来可以在这里加载更多工具如 file_tools, api_tools 等 async def get_tools_for_session(self, session_metadata: dict) - list: 这是一个关键钩子可以根据会话元数据如用户身份动态返回可用的工具列表。 实现基础的权限控制。 # 示例假设metadata中包含用户角色 user_role session_metadata.get(role, guest) if user_role analyst: return self.tools # 分析师拥有所有工具 elif user_role viewer: # 只返回只读工具过滤掉写操作工具 return [t for t in self.tools if t.name in [query_sales_summary]] else: return [] # 无权限用户 # 在server.py中 from context import MyContext import asyncio from mcp import StdioServerTransport, Server from mcp.server import Server async def main(): # 1. 初始化业务上下文 context MyContext() # 初始化数据库连接读取配置 init_db_connection(os.getenv(DATABASE_URL)) # 2. 创建MCP Server实例 server Server() # 3. 注册工具列表提供方法 server.list_tools() async def handle_list_tools(session_metadata: dict) - list: # 调用上下文的权限过滤方法 return await context.get_tools_for_session(session_metadata) # 4. 注册工具调用处理器 server.call_tool() async def handle_call_tool(name: str, arguments: dict, session_metadata: dict) - dict: # 根据工具名找到对应的工具函数 available_tools await context.get_tools_for_session(session_metadata) tool_map {t.name: t for t in available_tools} if name not in tool_map: raise ValueError(fTool not found or not authorized: {name}) tool tool_map[name] # 调用工具函数传入参数 result await tool(**arguments) # 将结果转换为字典如果工具返回的是Pydantic模型需.dict() return result.dict() if hasattr(result, dict) else result # 5. 启动Server这里以Stdio为例用于CLI集成 async with StdioServerTransport() as transport: await server.run(transport) if __name__ __main__: asyncio.run(main())这个流程展示了模板如何将你的业务工具与MCP协议桥接起来。MyContext类是你的“工具管家”负责工具的收集和基于会话的过滤。server.py中的装饰器函数则是标准的MCP请求处理器。4. 权限控制与安全增强实践MCP协议本身不强制规定权限模型这给了模板和开发者极大的灵活性。mcp-context-template通常会在上下文管理层预留钩子就像上面的get_tools_for_session。我们可以在此基础上进行强化。4.1 基于角色的工具访问控制RBAC我们可以设计一个更完善的权限系统。在config/permissions.yaml中定义角色与工具的映射roles: viewer: tools: - query_sales_summary - get_report analyst: inherits: viewer tools: - execute_data_pipeline - update_dataset admin: inherits: analyst tools: - manage_users - shutdown_server然后在上下文管理中读取这个配置import yaml from pathlib import Path class RBACContext: def __init__(self, config_path: Path): with open(config_path) as f: self.config yaml.safe_load(f) self.all_tools self._load_all_tools() # 加载所有工具对象 async def get_tools_for_session(self, session_metadata: dict) - list: user_role session_metadata.get(role, viewer) role_config self.config[roles].get(user_role, {}) allowed_tool_names set(role_config.get(tools, [])) # 处理继承 if inherits in role_config: parent_tools self.config[roles][role_config[inherits]].get(tools, []) allowed_tool_names.update(parent_tools) return [t for t in self.all_tools if t.name in allowed_tool_names]4.2 工具级参数校验与操作审计除了谁能调用工具我们还需要关心工具被调用时做了什么。可以在工具调用处理器中加入审计日志和更细粒度的参数校验。import logging logger logging.getLogger(__name__) server.call_tool() async def handle_call_tool(name: str, arguments: dict, session_metadata: dict) - dict: user_id session_metadata.get(user_id, anonymous) # 1. 权限检查同上 available_tools await context.get_tools_for_session(session_metadata) tool_map {t.name: t for t in available_tools} if name not in tool_map: logger.warning(fUnauthorized tool access attempt: user{user_id}, tool{name}) raise PermissionError(fTool {name} is not authorized.) tool tool_map[name] # 2. 审计日志记录谁在什么时候调用了什么工具参数是什么注意过滤敏感参数 safe_args arguments.copy() safe_args.pop(password, None) # 示例过滤密码字段 logger.info(fTool invoked: user{user_id}, tool{name}, args{safe_args}) # 3. 额外的参数业务校验例如查询时间范围不能超过365天 if name query_sales_summary: from datetime import date, timedelta start date.fromisoformat(arguments[start_date]) end date.fromisoformat(arguments[end_date]) if (end - start).days 365: raise ValueError(查询时间范围不能超过365天。) # 4. 执行工具 try: result await tool(**arguments) logger.info(fTool succeeded: user{user_id}, tool{name}) return result.dict() if hasattr(result, dict) else result except Exception as e: logger.error(fTool execution failed: user{user_id}, tool{name}, error{str(e)}) raise注意事项审计日志非常重要尤其是在生产环境。但要注意日志中不能记录敏感信息如完整的SQL语句、API密钥、个人身份信息。建议对参数进行清洗或只记录工具名和部分非敏感元数据。5. 客户端集成与调试技巧Server搭建好后我们需要一个MCP Client来测试它。目前许多AI应用框架和平台正在逐步支持MCP Client。5.1 使用MCP Inspector进行调试最直接的调试方法是使用官方或社区的MCP Inspector工具。它是一个独立的桌面应用或CLI工具可以连接到你的MCP Server列出所有可用的工具和资源并手动发起调用观察请求和响应。# 假设你的Server通过stdio启动 npx modelcontextprotocol/inspector python server.py运行后Inspector会打开一个图形界面或提供命令行交互让你可以浏览工具、填写参数并执行是开发和调试阶段不可或缺的利器。5.2 集成到Claude Desktop或Cursor等AI应用一些先进的AI应用已经开始支持MCP。例如在Claude Desktop中你可以通过编辑配置文件来添加自定义的MCP Server。找到Claude Desktop的配置文件夹。编辑claude_desktop_config.json添加你的Server配置{ mcpServers: { my-data-assistant: { command: python, args: [/absolute/path/to/your/server.py], env: { DATABASE_URL: your_db_connection_string } } } }重启Claude Desktop。之后Claude AI在对话中就能“看到”并使用你定义的query_sales_summary等工具了你可以直接说“帮我分析一下上季度的北美电子产品销售情况”。5.3 在自定义AI应用中集成如果你在构建自己的AI应用可以使用MCP Client SDK来连接你的Server。import asyncio from mcp import Client, StdioClientTransport async def main(): # 启动Server进程并创建Client传输层 transport await StdioClientTransport.create([python, server.py]) client Client(transport) await client.initialize() # 列出可用工具 tools await client.list_tools() print(Available tools:, [t.name for t in tools.tools]) # 调用工具 result await client.call_tool( namequery_sales_summary, arguments{ start_date: 2024-01-01, end_date: 2024-03-31, region: North America } ) print(Query result:, result) await client.finalize() asyncio.run(main())6. 性能优化与生产部署考量当工具数量和复杂度增长后性能和维护性成为关键。6.1 工具懒加载与缓存不是所有工具都需要在Server启动时就全部加载进内存。对于一些初始化成本高、使用频率低的工具可以采用懒加载。class LazyTool: def __init__(self, name, loader_func): self.name name self._loader_func loader_func self._tool_instance None property async def tool(self): if self._tool_instance is None: self._tool_instance await self._loader_func() return self._tool_instance async def __call__(self, **kwargs): inst await self.tool return await inst(**kwargs) # 在上下文中注册懒加载工具 async def load_heavy_ml_tool(): # 模拟加载一个大型机器学习模型 import torch model torch.load(large_model.pt) return model.predict context.tools.append(LazyTool(namepredict_with_ml, loader_funcload_heavy_ml_tool))同时对于查询类工具的结果如果参数相同可以考虑在短时间内缓存结果避免重复查询数据库。from functools import lru_cache import asyncio # 注意缓存需要根据业务场景设置合理的过期时间和大小 lru_cache(maxsize128) async def cached_query_sales_summary(start_date_str: str, end_date_str: str, region: str None): # 实际的数据库查询逻辑 pass # 在工具函数中调用缓存版本 async def query_sales_summary_tool_cached(query: SalesQuery): key (query.start_date.isoformat(), query.end_date.isoformat(), query.region) return await cached_query_sales_summary(*key)6.2 错误处理与健壮性生产环境的工具必须健壮。模板应鼓励全面的错误处理。输入验证充分利用Pydantic模型的验证能力。对于更复杂的业务规则在工具函数开头进行校验。超时控制为可能长时间运行的工具设置超时。import asyncio async def call_tool_with_timeout(tool_func, args, timeout30): try: return await asyncio.wait_for(tool_func(**args), timeouttimeout) except asyncio.TimeoutError: logger.error(fTool {tool_func.__name__} timed out after {timeout}s) raise TimeoutError(Tool execution timed out.)优雅降级当依赖的外部服务如某个API不可用时工具应返回有意义的错误信息或尝试使用备用方案而不是让整个Server崩溃。6.3 监控与可观测性除了审计日志还需要添加性能指标监控。可以集成像Prometheus这样的监控系统。from prometheus_client import Counter, Histogram TOOL_CALL_COUNT Counter(mcp_tool_calls_total, Total tool calls, [tool_name, status]) TOOL_CALL_DURATION Histogram(mcp_tool_call_duration_seconds, Tool call duration, [tool_name]) server.call_tool() async def handle_call_tool_with_metrics(name: str, arguments: dict, session_metadata: dict) - dict: with TOOL_CALL_DURATION.labels(tool_namename).time(): try: result await call_tool_impl(name, arguments, session_metadata) # 原有的调用逻辑 TOOL_CALL_COUNT.labels(tool_namename, statussuccess).inc() return result except Exception as e: TOOL_CALL_COUNT.labels(tool_namename, statuserror).inc() raise这样你就能清晰地看到每个工具的调用频率、成功率和延迟分布便于发现性能瓶颈和异常。7. 常见问题排查与模板定制7.1 工具未在客户端显示检查点1Server是否正常启动并输出了初始化日志确保没有端口冲突或依赖缺失。检查点2list_tools方法是否正确返回了工具列表在Inspector中查看或添加调试日志。检查点3权限过滤是否过于严格检查get_tools_for_session逻辑确保当前会话的元数据能通过过滤。临时可以注释掉权限逻辑进行测试。检查点4工具描述格式是否正确MCP对工具描述的schema有要求确保name、description、inputSchema字段完整且格式正确。使用Pydantic模型能自动生成合规的JSON Schema。7.2 工具调用失败返回参数错误检查点1客户端传入的参数格式是否与schema匹配在Inspector中仔细对照工具定义的输入模型。日期字段是否是字符串格式枚举值是否正确检查点2Pydantic模型验证是否失败在工具函数内部或调用前手动用模型验证一下输入SalesQuery(**arguments)。检查点3工具函数本身是否有未处理的异常查看Server端的错误日志。可能是数据库连接失败、网络超时等。7.3 性能瓶颈分析使用监控指标如上节所述通过指标定位是哪个工具慢。数据库查询分析对于数据库工具检查SQL是否使用了索引。可以在工具函数中记录或输出执行的SQL语句生产环境需谨慎用EXPLAIN分析。并发问题如果工具涉及共享状态如全局缓存、文件写入需考虑线程安全或异步锁。asyncio.Lock可以用于保护异步环境下的临界区。7.4 如何扩展模板添加新类型的工具模板是起点不是终点。添加新工具遵循固定模式在tools/目录下创建新模块如external_api_tools.py。实现具体的工具函数用Tool装饰器装饰并定义好输入输出模型。在上下文管理类中注册将新工具添加到MyContext.tools列表中。考虑权限更新权限配置如permissions.yaml决定哪些角色可以访问新工具。编写测试为新的工具函数编写单元测试模拟API调用。例如添加一个调用天气预报API的工具# tools/weather_tools.py import httpx from pydantic import BaseModel from mcp import Tool class WeatherQuery(BaseModel): city: str country_code: str US Tool(nameget_weather, description获取指定城市的当前天气, input_modelWeatherQuery) async def get_weather_tool(query: WeatherQuery) - str: async with httpx.AsyncClient() as client: # 使用一个模拟的天气API resp await client.get(fhttps://api.weather.example.com?city{query.city}country{query.country_code}) resp.raise_for_status() data resp.json() return fThe weather in {query.city} is {data[condition]} with a temperature of {data[temp_c]}°C.将这个工具导入并注册到上下文你的AI助手立刻就拥有了查询天气的能力。这种模块化的扩展方式正是mcp-context-template带来的最大便利。
