在实际开发中，我们经常需要将外部工具、数据源或服务集成到应用中。传统方式往往需要为每个外部资源编写特定的适配器代码，这不仅增加了开发复杂度，也降低了系统的可维护性。MCP（Model Context Protocol）原语提供了一种标准化的方式来定义和使用这些外部资源，让开发者能够以统一的方式管理各种工具和资源。
本文将深入探讨 MCP 原语中 Tool 和 Resource 的使用方式，通过实际案例展示如何定义、配置和调用这些原语，并分析在生产环境中需要注意的关键问题。
1. 理解 MCP 原语的核心概念
1.1 MCP 协议的基本作用
MCP 协议定义了一套标准接口，允许 AI 模型与外部工具和资源进行安全、可控的交互。它解决了传统集成方式中的几个核心痛点：
- 标准化接口：不同工具和资源可以通过统一的协议进行访问，无需为每个工具编写特定的集成代码
- 安全控制：通过权限机制控制模型对资源的访问范围，防止越权操作
- 可扩展性：新的工具和资源可以快速接入现有系统，不影响已有功能
在实际项目中，MCP 协议通常用于构建智能助手、自动化流程和数据分析系统，让 AI 模型能够安全地使用外部 API、数据库和工具链。
1.2 Tool 与 Resource 的区别与联系
Tool 和 Resource 是 MCP 协议中的两个核心原语，它们有不同的职责和使用场景：
Tool（工具）代表可执行的操作，通常具有明确的输入和输出。例如：
- 调用外部 API 接口
- 执行数据库查询
- 运行系统命令
- 处理文件操作
Resource（资源）代表可访问的数据源或服务实例，通常用于提供上下文信息。例如：
- 数据库连接配置
- API 访问凭证
- 文件路径或 URL
- 配置参数集合
两者之间的关系是：Resource 为 Tool 提供运行所需的上下文和环境，Tool 利用 Resource 来执行具体操作。一个典型的例子是，数据库连接（Resource）为 SQL 查询工具（Tool）提供数据访问能力。
2. 环境准备与项目配置
2.1 开发环境要求
在开始使用 MCP 原语前，需要确保开发环境满足以下要求：
| 环境组件 | 版本要求 | 说明 |
|---|---|---|
| Python | 3.8+ | MCP 协议的主流实现语言 |
| Node.js | 16+ | 可选，用于前端集成 |
| Docker | 20.10+ | 用于容器化部署 |
| Git | 2.25+ | 版本管理 |
验证环境是否就绪：
# 检查 Python 版本 python --version # 检查 Node.js 版本 node --version # 检查 Docker 状态 docker --version docker ps2.2 创建 MCP 项目结构
标准的 MCP 项目应该包含以下目录结构：
mcp-project/ ├── src/ │ ├── tools/ # Tool 实现 │ │ ├── __init__.py │ │ ├── api_tools.py │ │ └── data_tools.py │ ├── resources/ # Resource 定义 │ │ ├── __init__.py │ │ ├── database.py │ │ └── api_client.py │ └── server.py # MCP 服务器入口 ├── config/ │ ├── development.yaml │ └── production.yaml ├── tests/ # 测试用例 ├── requirements.txt # Python 依赖 └── README.md初始化项目依赖：
# 创建虚拟环境 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装核心依赖 pip install mcp-protocol pydantic fastapi uvicorn pip install requests sqlalchemy python-dotenv # 保存依赖列表 pip freeze > requirements.txt3. Tool 原语的实现与使用
3.1 定义基本的 Tool 接口
Tool 原语需要实现特定的接口规范。以下是一个基础的 Tool 类定义：
from typing import Any, Dict, Optional from pydantic import BaseModel, Field from mcp import Tool, ToolResult class SearchToolInput(BaseModel): query: str = Field(..., description="搜索关键词") max_results: int = Field(10, description="最大返回结果数") class SearchTool(Tool): name = "search_tool" description = "执行网络搜索并返回结果" input_schema = SearchToolInput async def execute(self, input_data: Dict[str, Any]) -> ToolResult: query = input_data.get("query", "") max_results = input_data.get("max_results", 10) # 模拟搜索逻辑 results = await self._perform_search(query, max_results) return ToolResult( content=results, is_error=False ) async def _perform_search(self, query: str, max_results: int) -> List[Dict]: # 实际项目中这里会调用搜索引擎 API return [ {"title": f"结果 {i}", "url": f"https://example.com/result{i}", "snippet": f"关于 {query} 的内容"} for i in range(min(max_results, 5)) ]3.2 实现数据操作 Tool
在实际业务中，数据库操作是最常见的 Tool 使用场景：
from sqlalchemy import create_engine, text from sqlalchemy.orm import sessionmaker class DatabaseQueryInput(BaseModel): sql: str = Field(..., description="要执行的 SQL 查询") parameters: Optional[Dict] = Field(None, description="查询参数") class DatabaseQueryTool(Tool): name = "db_query" description = "执行安全的数据库查询" input_schema = DatabaseQueryInput def __init__(self, database_url: str): self.engine = create_engine(database_url) self.Session = sessionmaker(bind=self.engine) async def execute(self, input_data: Dict[str, Any]) -> ToolResult: sql = input_data.get("sql", "") parameters = input_data.get("parameters", {}) # 安全性检查：禁止写操作 if any(keyword in sql.upper() for keyword in ['INSERT', 'UPDATE', 'DELETE', 'DROP']): return ToolResult( content="错误：该工具只支持 SELECT 查询", is_error=True ) try: with self.Session() as session: result = session.execute(text(sql), parameters) columns = list(result.keys()) rows = [dict(zip(columns, row)) for row in result.fetchall()] return ToolResult( content={"columns": columns, "rows": rows}, is_error=False ) except Exception as e: return ToolResult( content=f"数据库查询错误: {str(e)}", is_error=True )3.3 Tool 的注册与调用
在 MCP 服务器中注册和管理 Tool：
from mcp import MCPServer from typing import List class MyMCPServer(MCPServer): def __init__(self): super().__init__() self._tools = {} def register_tool(self, tool: Tool): """注册工具到服务器""" self._tools[tool.name] = tool def get_available_tools(self) -> List[Dict]: """返回可用工具列表""" return [ { "name": tool.name, "description": tool.description, "input_schema": tool.input_schema.schema() } for tool in self._tools.values() ] async def call_tool(self, tool_name: str, arguments: Dict) -> Dict: """调用指定工具""" if tool_name not in self._tools: raise ValueError(f"工具不存在: {tool_name}") tool = self._tools[tool_name] return await tool.execute(arguments) # 使用示例 server = MyMCPServer() server.register_tool(SearchTool()) server.register_tool(DatabaseQueryTool("sqlite:///example.db")) # 获取工具列表 tools = server.get_available_tools() print("可用工具:", tools) # 调用工具 result = await server.call_tool("search_tool", {"query": "MCP 协议", "max_results": 3}) print("搜索结果:", result)4. Resource 原语的定义与管理
4.1 设计 Resource 数据结构
Resource 需要清晰定义其属性和访问规则：
from typing import Dict, Any, Optional from pydantic import BaseModel, validator import os class DatabaseResourceConfig(BaseModel): """数据库资源配置""" url: str pool_size: int = 5 max_overflow: int = 10 timeout: int = 30 @validator('url') def validate_url(cls, v): if not v.startswith(('postgresql://', 'mysql://', 'sqlite://')): raise ValueError('不支持的数据库类型') return v class ApiResourceConfig(BaseModel): """API 资源配置""" base_url: str api_key: Optional[str] = None timeout: int = 30 retry_times: int = 3 @validator('base_url') def validate_base_url(cls, v): if not v.startswith(('http://', 'https://')): raise ValueError('URL 必须以 http:// 或 https:// 开头') return v class FileResourceConfig(BaseModel): """文件资源配置""" path: str mode: str = 'r' encoding: str = 'utf-8' @validator('path') def validate_path(cls, v): if not os.path.exists(v): raise ValueError(f'文件路径不存在: {v}') return v4.2 实现 Resource 管理器
Resource 管理器负责资源的生命周期管理：
from contextlib import asynccontextmanager from typing import AsyncIterator, Dict, Any class ResourceManager: def __init__(self): self._resources: Dict[str, Any] = {} self._configs: Dict[str, Any] = {} def register_resource(self, name: str, config: BaseModel): """注册资源配置""" self._configs[name] = config @asynccontextmanager async def get_resource(self, name: str) -> AsyncIterator[Any]: """获取资源实例（上下文管理器）""" if name not in self._configs: raise ValueError(f"资源未注册: {name}") config = self._configs[name] # 根据配置类型创建资源实例 if isinstance(config, DatabaseResourceConfig): resource = await self._create_database_connection(config) elif isinstance(config, ApiResourceConfig): resource = await self._create_api_client(config) elif isinstance(config, FileResourceConfig): resource = await self._open_file(config) else: raise ValueError(f"不支持的资源类型: {type(config)}") try: yield resource finally: await self._close_resource(resource) async def _create_database_connection(self, config: DatabaseResourceConfig): """创建数据库连接""" from sqlalchemy import create_engine engine = create_engine( config.url, pool_size=config.pool_size, max_overflow=config.max_overflow, pool_timeout=config.timeout ) return engine.connect() async def _create_api_client(self, config: ApiResourceConfig): """创建 API 客户端""" import aiohttp session = aiohttp.ClientSession( base_url=config.base_url, timeout=aiohttp.ClientTimeout(total=config.timeout) ) if config.api_key: session.headers.update({'Authorization': f'Bearer {config.api_key}'}) return session async def _open_file(self, config: FileResourceConfig): """打开文件""" return open(config.path, config.mode, encoding=config.encoding) async def _close_resource(self, resource): """关闭资源""" if hasattr(resource, 'close'): if hasattr(resource.close, '__await__'): await resource.close() else: resource.close() # 使用示例 resource_manager = ResourceManager() # 注册资源 resource_manager.register_resource( "main_db", DatabaseResourceConfig(url="sqlite:///example.db") ) resource_manager.register_resource( "weather_api", ApiResourceConfig(base_url="https://api.weather.com", api_key="your_api_key") ) # 使用资源 async with resource_manager.get_resource("main_db") as db_conn: result = db_conn.execute("SELECT * FROM users LIMIT 5") print(result.fetchall())4.3 Resource 的安全访问控制
在生产环境中，需要对 Resource 的访问进行严格管控：
from enum import Enum from typing import Set class PermissionLevel(Enum): READ_ONLY = "read_only" READ_WRITE = "read_write" ADMIN = "admin" class SecureResourceManager(ResourceManager): def __init__(self): super().__init__() self._permissions: Dict[str, Set[PermissionLevel]] = {} def set_permission(self, resource_name: str, levels: Set[PermissionLevel]): """设置资源访问权限""" self._permissions[resource_name] = levels async def get_resource(self, name: str, user_permission: PermissionLevel) -> AsyncIterator[Any]: """带权限检查的资源获取""" if name not in self._permissions: raise PermissionError(f"资源访问未授权: {name}") allowed_levels = self._permissions[name] if user_permission not in allowed_levels: raise PermissionError(f"权限不足: 需要 {allowed_levels}, 当前 {user_permission}") async with super().get_resource(name) as resource: # 根据权限级别包装资源 if user_permission == PermissionLevel.READ_ONLY: resource = self._wrap_readonly(resource) yield resource def _wrap_readonly(self, resource): """包装资源为只读模式""" # 实际项目中会根据资源类型实现具体的只读包装 # 例如数据库连接可以包装为只允许 SELECT 的代理 return resource # 权限配置示例 secure_manager = SecureResourceManager() secure_manager.register_resource("user_db", DatabaseResourceConfig(url="sqlite:///users.db")) secure_manager.set_permission("user_db", {PermissionLevel.READ_ONLY, PermissionLevel.ADMIN}) # 不同权限级别的访问 try: async with secure_manager.get_resource("user_db", PermissionLevel.READ_ONLY) as db: # 只读操作 pass except PermissionError as e: print(f"访问被拒绝: {e}")5. Tool 与 Resource 的协同工作
5.1 依赖注入模式
Tool 可以通过依赖注入的方式使用 Resource：
from typing import List class ResourceAwareTool(Tool): """支持资源依赖的工具基类""" def __init__(self, resource_manager: ResourceManager): self.resource_manager = resource_manager self.required_resources: List[str] = [] class DatabaseReportTool(ResourceAwareTool): def __init__(self, resource_manager: ResourceManager): super().__init__(resource_manager) self.name = "db_report" self.description = "生成数据库报表" self.required_resources = ["main_db", "analytics_db"] self.input_schema = ReportInput async def execute(self, input_data: Dict[str, Any]) -> ToolResult: # 检查所需资源是否可用 for resource_name in self.required_resources: if resource_name not in self.resource_manager._configs: return ToolResult( content=f"所需资源不可用: {resource_name}", is_error=True ) try: # 使用多个资源执行复杂操作 async with self.resource_manager.get_resource("main_db") as main_db: async with self.resource_manager.get_resource("analytics_db") as analytics_db: report = await self._generate_report(main_db, analytics_db, input_data) return ToolResult(content=report, is_error=False) except Exception as e: return ToolResult(content=f"报表生成失败: {str(e)}", is_error=True)5.2 配置驱动的工具注册
通过配置文件管理 Tool 和 Resource 的注册关系：
# config/tools.yaml tools: search_tool: class: "src.tools.api_tools.SearchTool" resources: ["search_api"] db_query_tool: class: "src.tools.data_tools.DatabaseQueryTool" resources: ["main_db"] report_tool: class: "src.tools.ReportTool" resources: ["main_db", "analytics_db"] # config/resources.yaml resources: main_db: type: "database" url: "postgresql://user:pass@localhost/main" pool_size: 5 analytics_db: type: "database" url: "postgresql://user:pass@localhost/analytics" pool_size: 3 search_api: type: "api" base_url: "https://api.search.com" api_key: "${SEARCH_API_KEY}"配置加载器实现：
import yaml from typing import Dict, Any class ConfigLoader: @staticmethod def load_tools_config(file_path: str) -> Dict[str, Any]: with open(file_path, 'r', encoding='utf-8') as f: config = yaml.safe_load(f) return config.get('tools', {}) @staticmethod def load_resources_config(file_path: str) -> Dict[str, Any]: with open(file_path, 'r', encoding='utf-8') as f: config = yaml.safe_load(f) return config.get('resources', {}) class MCPApplication: def __init__(self, config_dir: str = "config"): self.config_dir = config_dir self.resource_manager = SecureResourceManager() self.server = MyMCPServer() def setup(self): """根据配置文件初始化应用""" # 加载资源配置 resources_config = ConfigLoader.load_resources_config( f"{self.config_dir}/resources.yaml" ) self._setup_resources(resources_config) # 加载工具配置 tools_config = ConfigLoader.load_tools_config( f"{self.config_dir}/tools.yaml" ) self._setup_tools(tools_config) def _setup_resources(self, config: Dict[str, Any]): """根据配置注册资源""" for name, resource_config in config.items(): if resource_config['type'] == 'database': config_obj = DatabaseResourceConfig(**resource_config) elif resource_config['type'] == 'api': config_obj = ApiResourceConfig(**resource_config) else: continue self.resource_manager.register_resource(name, config_obj) def _setup_tools(self, config: Dict[str, Any]): """根据配置注册工具""" for name, tool_config in config.items(): class_path = tool_config['class'] # 动态导入工具类 module_name, class_name = class_path.rsplit('.', 1) module = __import__(module_name, fromlist=[class_name]) tool_class = getattr(module, class_name) # 创建工具实例 tool_instance = tool_class(self.resource_manager) self.server.register_tool(tool_instance) # 应用启动 app = MCPApplication() app.setup()6. 生产环境部署与监控
6.1 容器化部署配置
使用 Docker 进行标准化部署：
# Dockerfile FROM python:3.9-slim WORKDIR /app # 安装系统依赖 RUN apt-get update && apt-get install -y \ gcc \ && rm -rf /var/lib/apt/lists/* # 复制依赖文件 COPY requirements.txt . # 安装 Python 依赖 RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY src/ src/ COPY config/ config/ # 创建非 root 用户 RUN useradd --create-home --shell /bin/bash mcpuser USER mcpuser # 健康检查 HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \ CMD curl -f http://localhost:8000/health || exit 1 # 启动应用 CMD ["python", "src/server.py"]对应的 Docker Compose 配置：
# docker-compose.yml version: '3.8' services: mcp-server: build: . ports: - "8000:8000" environment: - DATABASE_URL=postgresql://user:pass@db:5432/mcp - LOG_LEVEL=INFO depends_on: - db volumes: - ./logs:/app/logs restart: unless-stopped db: image: postgres:13 environment: - POSTGRES_DB=mcp - POSTGRES_USER=user - POSTGRES_PASSWORD=pass volumes: - postgres_data:/var/lib/postgresql/data restart: unless-stopped volumes: postgres_data:6.2 监控与日志配置
完善的监控体系对于生产环境至关重要：
import logging import json from datetime import datetime from typing import Dict, Any class MonitoringMiddleware: def __init__(self, app): self.app = app self.logger = logging.getLogger('mcp.monitoring') # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('logs/mcp_server.log'), logging.StreamHandler() ] ) async def log_tool_usage(self, tool_name: str, user: str, duration: float, success: bool): """记录工具使用情况""" log_entry = { "timestamp": datetime.utcnow().isoformat(), "tool": tool_name, "user": user, "duration_ms": round(duration * 1000, 2), "success": success, "type": "tool_usage" } self.logger.info(json.dumps(log_entry)) async def log_resource_access(self, resource_name: str, operation: str, user: str): """记录资源访问情况""" log_entry = { "timestamp": datetime.utcnow().isoformat(), "resource": resource_name, "operation": operation, "user": user, "type": "resource_access" } self.logger.info(json.dumps(log_entry)) class HealthCheckHandler: async def check_health(self) -> Dict[str, Any]: """健康检查端点""" checks = { "database": await self._check_database(), "external_apis": await self._check_external_apis(), "disk_space": await self._check_disk_space(), "memory_usage": await self._check_memory_usage() } overall_status = "healthy" if all(check["status"] == "healthy" for check in checks.values()) else "unhealthy" return { "status": overall_status, "timestamp": datetime.utcnow().isoformat(), "checks": checks } async def _check_database(self) -> Dict[str, Any]: """检查数据库连接""" try: # 实际项目中这里会测试真实数据库连接 return {"status": "healthy", "details": "Database connection OK"} except Exception as e: return {"status": "unhealthy", "details": str(e)}6.3 性能优化建议
针对高并发场景的优化策略：
from concurrent.futures import ThreadPoolExecutor import asyncio from typing import List class OptimizedResourceManager(ResourceManager): def __init__(self, max_workers: int = 10): super().__init__() self.thread_pool = ThreadPoolExecutor(max_workers=max_workers) self.connection_pools: Dict[str, Any] = {} async def get_resource(self, name: str) -> AsyncIterator[Any]: """带连接池的资源获取""" if name not in self._configs: raise ValueError(f"资源未注册: {name}") # 使用连接池管理数据库连接 if isinstance(self._configs[name], DatabaseResourceConfig): if name not in self.connection_pools: self.connection_pools[name] = await self._create_connection_pool(name) async with self.connection_pools[name].acquire() as connection: yield connection else: async with super().get_resource(name) as resource: yield resource async def _create_connection_pool(self, resource_name: str): """创建数据库连接池""" config = self._configs[resource_name] # 使用 asyncpg 或其他异步数据库驱动 import asyncpg return await asyncpg.create_pool(config.url) class ToolExecutionOptimizer: def __init__(self, server: MyMCPServer): self.server = server self.cache = {} async def execute_tool_with_cache(self, tool_name: str, arguments: Dict, cache_key: str = None, ttl: int = 300): """带缓存的工具执行""" if cache_key is None: cache_key = self._generate_cache_key(tool_name, arguments) # 检查缓存 if cache_key in self.cache: cached_result = self.cache[cache_key] if datetime.utcnow().timestamp() - cached_result['timestamp'] < ttl: return cached_result['result'] # 执行工具 result = await self.server.call_tool(tool_name, arguments) # 缓存结果 self.cache[cache_key] = { 'result': result, 'timestamp': datetime.utcnow().timestamp() } return result def _generate_cache_key(self, tool_name: str, arguments: Dict) -> str: """生成缓存键""" import hashlib key_data = f"{tool_name}:{json.dumps(arguments, sort_keys=True)}" return hashlib.md5(key_data.encode()).hexdigest()7. 常见问题排查与解决方案
7.1 工具执行失败排查
工具执行过程中可能遇到的各种问题及解决方法：
| 问题现象 | 可能原因 | 检查步骤 | 解决方案 |
|---|---|---|---|
| 工具调用返回权限错误 | 资源权限配置不当 | 检查用户权限级别和资源访问规则 | 调整权限配置或使用更高权限级别 |
| 数据库连接超时 | 网络问题或连接池耗尽 | 检查数据库状态和连接数限制 | 增加连接池大小或优化查询性能 |
| API 调用返回 4xx/5xx | 参数错误或服务不可用 | 验证输入参数和服务状态 | 修正参数或实现重试机制 |
| 内存使用持续增长 | 资源未正确释放或内存泄漏 | 检查资源清理逻辑和对象引用 | 确保使用上下文管理器，定期重启服务 |
具体的排查代码示例：
class DiagnosticTool(Tool): """诊断工具，用于排查系统问题""" name = "diagnostic" description = "系统诊断和问题排查" async def execute(self, input_data: Dict[str, Any]) -> ToolResult: check_type = input_data.get("check_type", "all") results = {} if check_type in ["all", "resources"]: results["resources"] = await self._check_resources() if check_type in ["all", "tools"]: results["tools"] = await self._check_tools() if check_type in ["all", "performance"]: results["performance"] = await self._check_performance() return ToolResult(content=results, is_error=False) async def _check_resources(self) -> Dict[str, Any]: """检查资源状态""" checks = {} # 实现具体的资源检查逻辑 return checks async def _check_tools(self) -> Dict[str, Any]: """检查工具状态""" checks = {} # 实现具体的工具检查逻辑 return checks7.2 资源管理问题处理
资源管理中的常见问题及应对策略：
class ResourceHealthMonitor: def __init__(self, resource_manager: ResourceManager): self.resource_manager = resource_manager self.health_status = {} async def monitor_resources(self): """定期监控资源健康状态""" while True: for resource_name in self.resource_manager._configs.keys(): status = await self._check_resource_health(resource_name) self.health_status[resource_name] = status if status != "healthy": await self._handle_unhealthy_resource(resource_name, status) await asyncio.sleep(60) # 每分钟检查一次 async def _check_resource_health(self, resource_name: str) -> str: """检查单个资源健康状态""" try: async with self.resource_manager.get_resource(resource_name) as resource: # 根据资源类型执行健康检查 if hasattr(resource, 'execute'): # 数据库连接检查 await resource.execute("SELECT 1") elif hasattr(resource, 'get'): # API 客户端检查 async with resource.get("/health") as response: if response.status == 200: return "healthy" return "healthy" except Exception as e: return f"unhealthy: {str(e)}" async def _handle_unhealthy_resource(self, resource_name: str, status: str): """处理不健康的资源""" # 记录告警 logging.warning(f"资源 {resource_name} 状态异常: {status}") # 尝试恢复或切换备用资源 if "database" in resource_name: await self._try_failover_database(resource_name)7.3 配置验证与错误预防
在系统启动时进行配置验证，预防运行时错误：
class ConfigurationValidator: def __init__(self, config_dir: str): self.config_dir = config_dir async def validate_all_configs(self) -> List[str]: """验证所有配置文件""" errors = [] # 验证资源配置 resources_config = ConfigLoader.load_resources_config( f"{self.config_dir}/resources.yaml" ) errors.extend(await self._validate_resources_config(resources_config)) # 验证工具配置 tools_config = ConfigLoader.load_tools_config( f"{self.config_dir}/tools.yaml" ) errors.extend(await self._validate_tools_config(tools_config)) return errors async def _validate_resources_config(self, config: Dict[str, Any]) -> List[str]: """验证资源配置""" errors = [] for name, resource_config in config.items(): try: # 测试资源连接 if resource_config['type'] == 'database': await self._test_database_connection(resource_config) elif resource_config['type'] == 'api': await self._test_api_connection(resource_config) except Exception as e: errors.append(f"资源 {name} 配置错误: {str(e)}") return errors async def _test_database_connection(self, config: Dict[str, Any]): """测试数据库连接""" # 实现数据库连接测试 pass通过以上完整的实现方案，可以构建出健壮、可维护的 MCP 原语系统。在实际项目中，还需要根据具体业务需求调整工具和资源的定义，并建立相应的监控和告警机制。
