如果你正在构建AI应用可能会遇到这样的困境大模型虽然能力强大但总是一问三不知——不是模型本身不够聪明而是它无法访问你需要的特定数据源。无论是公司内部的数据库、第三方API还是本地文件系统这些数据孤岛让AI应用的实际价值大打折扣。这正是MCPModel Context Protocol要解决的核心问题。作为2024年AI工程化领域的重要突破MCP不是另一个复杂的框架而是一个标准化的通信协议让大模型能够安全、可控地访问外部工具和数据源。本文基于吴恩达团队的最新实践将带你从零构建完整的MCP程序。不同于简单的概念介绍我们将重点放在实际可落地的工程方案上涵盖协议设计、服务实现、客户端集成等关键环节。1. MCP协议的核心价值与适用场景1.1 为什么需要MCP协议传统AI应用开发中模型与外部工具的集成往往需要定制化开发。每个数据源都需要专门的适配器每个工具都要编写特定的调用逻辑。这种模式存在三个主要问题开发效率低下每次接入新工具都要重新设计接口安全风险难以控制模型直接访问敏感数据缺乏有效监管可移植性差为特定模型开发的工具链难以复用MCP通过标准化协议解决了这些问题。它定义了模型与工具之间的通用通信规范使得工具开发与模型选择完全解耦。1.2 MCP的典型应用场景在实际项目中MCP特别适合以下场景企业知识库问答让模型安全访问内部文档库、数据库自动化工作流集成日历、邮件、项目管理工具代码助手增强连接Git仓库、CI/CD系统、测试平台数据分析应用对接业务数据库、可视化工具2. MCP协议架构深度解析2.1 核心组件与通信流程MCP协议基于JSON-RPC 2.0规范包含三个核心组件MCP客户端承载大模型的应用或框架MCP服务器提供工具和能力的外部服务传输层支持Stdio、SSE、HTTP等多种通信方式典型的交互流程如下客户端启动时连接一个或多个MCP服务器服务器向客户端注册可用的工具Tools和资源Resources客户端根据模型需求调用相应工具服务器执行工具并返回结构化结果2.2 工具Tools与资源Resources的区别这是MCP中容易混淆但至关重要的概念工具Tools代表可执行的操作如查询数据库、发送邮件、执行命令。特点是需要明确的输入参数执行后产生输出结果可能改变系统状态资源Resources代表可读取的数据源如文件内容、数据库记录、API数据。特点是提供只读访问通过URI标识和定位内容可能随时间变化3. 环境准备与开发工具链3.1 基础环境要求开始MCP开发前需要准备以下环境# 检查Python版本推荐3.9 python --version # Python 3.9.6 # 安装核心依赖 pip install mcp python-dotenv openai3.2 开发工具推荐代码编辑器VS Code with Python扩展调试工具MCP DevTools Chrome扩展API测试Postman或curl环境管理pyenv或conda4. 构建第一个MCP服务器文件系统访问示例4.1 项目结构设计创建标准的MCP项目目录结构file-mcp-server/ ├── src/ │ └── file_server/ │ ├── __init__.py │ ├── server.py │ └── tools/ │ ├── __init__.py │ ├── file_tools.py │ └── search_tools.py ├── tests/ ├── requirements.txt └── README.md4.2 实现基础文件操作工具创建src/file_server/tools/file_tools.pyimport os import json from typing import Dict, Any, List from mcp import Tool, TextContent class FileReadTool(Tool): def __init__(self, base_path: str .): self.base_path os.path.abspath(base_path) property def name(self) - str: return read_file property def description(self) - str: return 读取指定路径的文本文件内容 property def input_schema(self) - Dict[str, Any]: return { type: object, properties: { file_path: { type: string, description: 相对于基础目录的文件路径 } }, required: [file_path] } def execute(self, file_path: str) - List[TextContent]: full_path os.path.join(self.base_path, file_path) # 安全检查防止路径遍历攻击 if not os.path.abspath(full_path).startswith(self.base_path): raise ValueError(访问路径超出允许范围) if not os.path.exists(full_path): raise FileNotFoundError(f文件不存在: {file_path}) if not os.path.isfile(full_path): raise ValueError(f路径不是文件: {file_path}) with open(full_path, r, encodingutf-8) as f: content f.read() return [TextContent(typetext, textcontent)] class FileListTool(Tool): def __init__(self, base_path: str .): self.base_path os.path.abspath(base_path) property def name(self) - str: return list_files property def description(self) - str: return 列出指定目录下的文件和子目录 property def input_schema(self) - Dict[str, Any]: return { type: object, properties: { directory_path: { type: string, description: 相对于基础目录的路径默认为当前目录, default: . } } } def execute(self, directory_path: str .) - List[TextContent]: full_path os.path.join(self.base_path, directory_path) # 路径安全检查 if not os.path.abspath(full_path).startswith(self.base_path): raise ValueError(访问路径超出允许范围) if not os.path.exists(full_path): raise FileNotFoundError(f目录不存在: {directory_path}) if not os.path.isdir(full_path): raise ValueError(f路径不是目录: {directory_path}) items os.listdir(full_path) result { directory: directory_path, files: [], directories: [] } for item in items: item_path os.path.join(full_path, item) if os.path.isfile(item_path): result[files].append(item) else: result[directories].append(item) return [TextContent( typetext, textjson.dumps(result, ensure_asciiFalse, indent2) )]4.3 实现文件搜索工具创建src/file_server/tools/search_tools.pyimport os import re from typing import Dict, Any, List from mcp import Tool, TextContent class FileSearchTool(Tool): def __init__(self, base_path: str .): self.base_path os.path.abspath(base_path) property def name(self) - str: return search_files property def description(self) - str: return 在文件中搜索指定文本内容 property def input_schema(self) - Dict[str, Any]: return { type: object, properties: { search_pattern: { type: string, description: 要搜索的文本或正则表达式 }, file_pattern: { type: string, description: 文件匹配模式如 *.py, default: * }, max_results: { type: integer, description: 最大返回结果数, default: 10 } }, required: [search_pattern] } def execute(self, search_pattern: str, file_pattern: str *, max_results: int 10) - List[TextContent]: import fnmatch results [] pattern re.compile(search_pattern, re.IGNORECASE) for root, dirs, files in os.walk(self.base_path): # 跳过隐藏目录 dirs[:] [d for d in dirs if not d.startswith(.)] for file in files: if not fnmatch.fnmatch(file, file_pattern): continue file_path os.path.join(root, file) try: with open(file_path, r, encodingutf-8) as f: content f.read() matches list(pattern.finditer(content)) if matches: relative_path os.path.relpath(file_path, self.base_path) results.append({ file: relative_path, match_count: len(matches), sample_matches: [m.group() for m in matches[:3]] }) if len(results) max_results: break except (UnicodeDecodeError, PermissionError): continue if len(results) max_results: break return [TextContent( typetext, textjson.dumps(results, ensure_asciiFalse, indent2) )]4.4 集成MCP服务器主程序创建src/file_server/server.pyimport asyncio import os from mcp import MCPServer from mcp.server.stdio import stdio_server from .tools.file_tools import FileReadTool, FileListTool from .tools.search_tools import FileSearchTool class FileMCPServer: def __init__(self, base_path: str .): self.base_path os.path.abspath(base_path) self.server MCPServer(file-mcp-server) # 注册工具 self.server.register_tool(FileReadTool(base_path)) self.server.register_tool(FileListTool(base_path)) self.server.register_tool(FileSearchTool(base_path)) async def run(self): 启动MCP服务器 async with stdio_server() as (read_stream, write_stream): await self.server.run( read_stream, write_stream, self.server.create_initialization_options() ) async def main(): # 从环境变量获取基础路径默认为当前目录 base_path os.getenv(MCP_BASE_PATH, .) server FileMCPServer(base_path) await server.run() if __name__ __main__: asyncio.run(main())5. 客户端集成与模型调用实战5.1 创建Python客户端示例import asyncio import os from mcp.client import create_session from mcp.client.stdio import stdio_client from openai import OpenAI class MCPClient: def __init__(self, server_path: str, model: str gpt-4): self.server_path server_path self.model model self.client OpenAI() async def query_with_tools(self, prompt: str) - str: 使用MCP工具增强的模型查询 # 第一步连接到MCP服务器并获取可用工具 async with stdio_client(self.server_path) as (read, write): async with create_session(read, write) as session: # 初始化连接 init_result await session.initialize() tools init_result.available_tools # 构建工具描述供模型使用 tool_descriptions [] for tool in tools: tool_descriptions.append({ type: function, function: { name: tool.name, description: tool.description, parameters: tool.input_schema } }) # 调用OpenAI API传入工具信息 response self.client.chat.completions.create( modelself.model, messages[{role: user, content: prompt}], toolstool_descriptions, tool_choiceauto ) message response.choices[0].message # 如果模型选择使用工具 if message.tool_calls: tool_results [] for tool_call in message.tool_calls: tool_name tool_call.function.name arguments json.loads(tool_call.function.arguments) # 通过MCP会话执行工具调用 result await session.call_tool( tool_name, arguments ) tool_results.append({ tool: tool_name, result: result.content }) # 将工具结果返回给模型进行最终回答 second_response self.client.chat.completions.create( modelself.model, messages[ {role: user, content: prompt}, message, { role: tool, content: json.dumps(tool_results), tool_call_id: tool_call.id } ] ) return second_response.choices[0].message.content else: return message.content # 使用示例 async def main(): client MCPClient( server_pathpython src/file_server/server.py, modelgpt-4 ) result await client.query_with_tools( 请帮我查找项目中有多少Python文件包含了def main这个模式 ) print(result) if __name__ __main__: asyncio.run(main())5.2 环境配置与安全设置创建.env配置文件# MCP服务器配置 MCP_BASE_PATH./workspace MCP_SERVER_LOG_LEVELINFO # OpenAI配置 OPENAI_API_KEYyour_api_key_here OPENAI_BASE_URLhttps://api.openai.com/v1 # 安全配置 MAX_FILE_SIZE10485760 # 10MB ALLOWED_EXTENSIONS.py,.txt,.md,.json,.yaml,.yml创建安全配置检查工具import os from pathlib import Path class SecurityValidator: def __init__(self, base_path: str, max_size: int 10*1024*1024): self.base_path Path(base_path).resolve() self.max_size max_size self.allowed_extensions {.py, .txt, .md, .json, .yaml, .yml} def validate_path(self, user_path: str) - Path: 验证路径安全性 try: full_path (self.base_path / user_path).resolve() # 检查是否在允许的基础路径内 if not str(full_path).startswith(str(self.base_path)): raise SecurityError(路径遍历攻击检测) # 检查文件扩展名 if full_path.is_file(): suffix full_path.suffix.lower() if suffix not in self.allowed_extensions: raise SecurityError(f不允许的文件类型: {suffix}) return full_path except Exception as e: raise SecurityError(f路径验证失败: {str(e)}) def validate_file_size(self, file_path: Path) - bool: 验证文件大小 if file_path.is_file(): size file_path.stat().st_size if size self.max_size: raise SecurityError(f文件过大: {size}字节) return True class SecurityError(Exception): 安全异常 pass6. 高级特性资源管理与实时更新6.1 实现资源提供器除了工具调用MCP还支持资源管理让模型能够订阅数据源的实时更新from mcp import Resource, ResourceTemplate from typing import List, Dict, Any import asyncio class DatabaseResource(Resource): def __init__(self, connection_string: str): self.connection_string connection_string property def uri(self) - str: return fdb://{hash(self.connection_string)}/schema property def name(self) - str: return database_schema property def description(self) - str: return 数据库表结构信息 async def read(self) - str: 读取当前数据库模式 # 模拟数据库连接和模式读取 import json schema { tables: [ { name: users, columns: [id, name, email, created_at] }, { name: orders, columns: [id, user_id, amount, status] } ] } return json.dumps(schema, indent2) class ResourceManager: def __init__(self): self.resources: Dict[str, Resource] {} def register_resource(self, resource: Resource): self.resources[resource.uri] resource async def get_resource(self, uri: str) - str: if uri not in self.resources: raise ValueError(f资源不存在: {uri}) return await self.resources[uri].read()6.2 实现资源订阅机制from typing import AsyncIterator import asyncio class ResourceSubscription: def __init__(self, resource: Resource, interval: int 30): self.resource resource self.interval interval self.subscribers set() async def notify_subscribers(self, content: str): 通知所有订阅者资源更新 for subscriber in self.subscribers: await subscriber(content) async def watch(self) - AsyncIterator[str]: 监视资源变化 last_content None while True: current_content await self.resource.read() if current_content ! last_content: last_content current_content await self.notify_subscribers(current_content) yield current_content await asyncio.sleep(self.interval)7. 生产环境部署与监控7.1 Docker容器化部署创建DockerfileFROM python:3.9-slim WORKDIR /app # 安装系统依赖 RUN apt-get update apt-get install -y \ gcc \ rm -rf /var/lib/apt/lists/* # 复制依赖文件 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY src/ ./src/ # 创建非root用户 RUN useradd -m -u 1000 mcpuser USER mcpuser # 设置环境变量 ENV PYTHONPATH/app/src ENV MCP_BASE_PATH/app/workspace # 创建数据目录 RUN mkdir -p /app/workspace CMD [python, -m, src.file_server.server]创建docker-compose.ymlversion: 3.8 services: mcp-file-server: build: . ports: - 8000:8000 volumes: - ./workspace:/app/workspace - ./logs:/app/logs environment: - MCP_BASE_PATH/app/workspace - MCP_SERVER_LOG_LEVELINFO restart: unless-stopped mcp-monitor: image: prom/prometheus:latest ports: - 9090:9090 volumes: - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml depends_on: - mcp-file-server7.2 监控与日志配置创建日志配置logging_config.pyimport logging import logging.config import json LOGGING_CONFIG { version: 1, disable_existing_loggers: False, formatters: { detailed: { format: %(asctime)s - %(name)s - %(levelname)s - %(message)s }, json: { format: %(asctime)s %(name)s %(levelname)s %(message)s, class: pythonjsonlogger.jsonlogger.JsonFormatter } }, handlers: { file: { class: logging.handlers.RotatingFileHandler, filename: /app/logs/mcp_server.log, maxBytes: 10485760, # 10MB backupCount: 5, formatter: detailed }, console: { class: logging.StreamHandler, formatter: detailed } }, loggers: { mcp: { handlers: [file, console], level: INFO, propagate: False } } } def setup_logging(): logging.config.dictConfig(LOGGING_CONFIG)8. 性能优化与最佳实践8.1 工具调用优化策略from functools import lru_cache import asyncio from concurrent.futures import ThreadPoolExecutor class ToolExecutor: def __init__(self, max_workers: int 4): self.thread_pool ThreadPoolExecutor(max_workersmax_workers) self.cache {} lru_cache(maxsize1000) async def execute_tool(self, tool_name: str, arguments: str) - str: 带缓存的工具执行 cache_key f{tool_name}:{arguments} if cache_key in self.cache: return self.cache[cache_key] # 在实际线程池中执行阻塞操作 loop asyncio.get_event_loop() result await loop.run_in_executor( self.thread_pool, self._execute_sync_tool, tool_name, arguments ) self.cache[cache_key] result return result def _execute_sync_tool(self, tool_name: str, arguments: str) - str: 同步执行工具在线程池中运行 # 实际工具执行逻辑 time.sleep(0.1) # 模拟耗时操作 return fResult for {tool_name} with {arguments}8.2 错误处理与重试机制import asyncio from typing import Type, Tuple from tenacity import retry, stop_after_attempt, wait_exponential class ResilientMCPClient: def __init__(self, max_retries: int 3): self.max_retries max_retries retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10) ) async def call_tool_with_retry(self, tool_name: str, arguments: dict): 带重试的工具调用 try: # 工具调用逻辑 return await self._call_tool(tool_name, arguments) except Exception as e: if self._is_retryable_error(e): raise # 触发重试 else: raise # 非重试性错误直接抛出 def _is_retryable_error(self, error: Exception) - bool: 判断错误是否可重试 retryable_errors ( ConnectionError, TimeoutError, asyncio.TimeoutError ) return isinstance(error, retryable_errors)9. 实际项目集成案例9.1 与LangChain集成示例from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from mcp import LangChainToolAdapter class MCPLangChainIntegration: def __init__(self, mcp_server_path: str, model: str gpt-4): self.mcp_tools self._load_mcp_tools(mcp_server_path) self.llm ChatOpenAI(modelmodel) self.agent self._create_agent() def _load_mcp_tools(self, server_path: str) - list: 将MCP工具转换为LangChain工具格式 tools [] # 模拟工具加载 tools.append(LangChainToolAdapter( nameread_file, description读取文件内容, funcself._read_file_wrapper )) return tools def _create_agent(self) - AgentExecutor: 创建LangChain代理 prompt ChatPromptTemplate.from_messages([ (system, 你是一个有帮助的AI助手可以使用工具来回答问题。), (placeholder, {chat_history}), (human, {input}), (placeholder, {agent_scratchpad}) ]) agent create_tool_calling_agent( self.llm, self.mcp_tools, prompt ) return AgentExecutor( agentagent, toolsself.mcp_tools, verboseTrue ) async def run_query(self, query: str) - str: 执行查询 result await self.agent.ainvoke({input: query}) return result[output]通过以上完整的MCP程序构建教程你不仅能够理解协议的核心概念更重要的是掌握了从开发到部署的全流程实践。MCP的真正价值在于它为大模型应用提供了一种标准化、安全可控的外部能力集成方案这在AI工程化日益重要的今天具有重大意义。建议在实际项目中从小规模开始先实现核心的文件操作工具再逐步扩展数据库访问、API集成等复杂功能。记得始终把安全性放在首位特别是在处理用户输入和文件路径时要做好充分的验证和限制。
MCP协议详解：构建AI应用的外部数据访问标准化方案
如果你正在构建AI应用可能会遇到这样的困境大模型虽然能力强大但总是一问三不知——不是模型本身不够聪明而是它无法访问你需要的特定数据源。无论是公司内部的数据库、第三方API还是本地文件系统这些数据孤岛让AI应用的实际价值大打折扣。这正是MCPModel Context Protocol要解决的核心问题。作为2024年AI工程化领域的重要突破MCP不是另一个复杂的框架而是一个标准化的通信协议让大模型能够安全、可控地访问外部工具和数据源。本文基于吴恩达团队的最新实践将带你从零构建完整的MCP程序。不同于简单的概念介绍我们将重点放在实际可落地的工程方案上涵盖协议设计、服务实现、客户端集成等关键环节。1. MCP协议的核心价值与适用场景1.1 为什么需要MCP协议传统AI应用开发中模型与外部工具的集成往往需要定制化开发。每个数据源都需要专门的适配器每个工具都要编写特定的调用逻辑。这种模式存在三个主要问题开发效率低下每次接入新工具都要重新设计接口安全风险难以控制模型直接访问敏感数据缺乏有效监管可移植性差为特定模型开发的工具链难以复用MCP通过标准化协议解决了这些问题。它定义了模型与工具之间的通用通信规范使得工具开发与模型选择完全解耦。1.2 MCP的典型应用场景在实际项目中MCP特别适合以下场景企业知识库问答让模型安全访问内部文档库、数据库自动化工作流集成日历、邮件、项目管理工具代码助手增强连接Git仓库、CI/CD系统、测试平台数据分析应用对接业务数据库、可视化工具2. MCP协议架构深度解析2.1 核心组件与通信流程MCP协议基于JSON-RPC 2.0规范包含三个核心组件MCP客户端承载大模型的应用或框架MCP服务器提供工具和能力的外部服务传输层支持Stdio、SSE、HTTP等多种通信方式典型的交互流程如下客户端启动时连接一个或多个MCP服务器服务器向客户端注册可用的工具Tools和资源Resources客户端根据模型需求调用相应工具服务器执行工具并返回结构化结果2.2 工具Tools与资源Resources的区别这是MCP中容易混淆但至关重要的概念工具Tools代表可执行的操作如查询数据库、发送邮件、执行命令。特点是需要明确的输入参数执行后产生输出结果可能改变系统状态资源Resources代表可读取的数据源如文件内容、数据库记录、API数据。特点是提供只读访问通过URI标识和定位内容可能随时间变化3. 环境准备与开发工具链3.1 基础环境要求开始MCP开发前需要准备以下环境# 检查Python版本推荐3.9 python --version # Python 3.9.6 # 安装核心依赖 pip install mcp python-dotenv openai3.2 开发工具推荐代码编辑器VS Code with Python扩展调试工具MCP DevTools Chrome扩展API测试Postman或curl环境管理pyenv或conda4. 构建第一个MCP服务器文件系统访问示例4.1 项目结构设计创建标准的MCP项目目录结构file-mcp-server/ ├── src/ │ └── file_server/ │ ├── __init__.py │ ├── server.py │ └── tools/ │ ├── __init__.py │ ├── file_tools.py │ └── search_tools.py ├── tests/ ├── requirements.txt └── README.md4.2 实现基础文件操作工具创建src/file_server/tools/file_tools.pyimport os import json from typing import Dict, Any, List from mcp import Tool, TextContent class FileReadTool(Tool): def __init__(self, base_path: str .): self.base_path os.path.abspath(base_path) property def name(self) - str: return read_file property def description(self) - str: return 读取指定路径的文本文件内容 property def input_schema(self) - Dict[str, Any]: return { type: object, properties: { file_path: { type: string, description: 相对于基础目录的文件路径 } }, required: [file_path] } def execute(self, file_path: str) - List[TextContent]: full_path os.path.join(self.base_path, file_path) # 安全检查防止路径遍历攻击 if not os.path.abspath(full_path).startswith(self.base_path): raise ValueError(访问路径超出允许范围) if not os.path.exists(full_path): raise FileNotFoundError(f文件不存在: {file_path}) if not os.path.isfile(full_path): raise ValueError(f路径不是文件: {file_path}) with open(full_path, r, encodingutf-8) as f: content f.read() return [TextContent(typetext, textcontent)] class FileListTool(Tool): def __init__(self, base_path: str .): self.base_path os.path.abspath(base_path) property def name(self) - str: return list_files property def description(self) - str: return 列出指定目录下的文件和子目录 property def input_schema(self) - Dict[str, Any]: return { type: object, properties: { directory_path: { type: string, description: 相对于基础目录的路径默认为当前目录, default: . } } } def execute(self, directory_path: str .) - List[TextContent]: full_path os.path.join(self.base_path, directory_path) # 路径安全检查 if not os.path.abspath(full_path).startswith(self.base_path): raise ValueError(访问路径超出允许范围) if not os.path.exists(full_path): raise FileNotFoundError(f目录不存在: {directory_path}) if not os.path.isdir(full_path): raise ValueError(f路径不是目录: {directory_path}) items os.listdir(full_path) result { directory: directory_path, files: [], directories: [] } for item in items: item_path os.path.join(full_path, item) if os.path.isfile(item_path): result[files].append(item) else: result[directories].append(item) return [TextContent( typetext, textjson.dumps(result, ensure_asciiFalse, indent2) )]4.3 实现文件搜索工具创建src/file_server/tools/search_tools.pyimport os import re from typing import Dict, Any, List from mcp import Tool, TextContent class FileSearchTool(Tool): def __init__(self, base_path: str .): self.base_path os.path.abspath(base_path) property def name(self) - str: return search_files property def description(self) - str: return 在文件中搜索指定文本内容 property def input_schema(self) - Dict[str, Any]: return { type: object, properties: { search_pattern: { type: string, description: 要搜索的文本或正则表达式 }, file_pattern: { type: string, description: 文件匹配模式如 *.py, default: * }, max_results: { type: integer, description: 最大返回结果数, default: 10 } }, required: [search_pattern] } def execute(self, search_pattern: str, file_pattern: str *, max_results: int 10) - List[TextContent]: import fnmatch results [] pattern re.compile(search_pattern, re.IGNORECASE) for root, dirs, files in os.walk(self.base_path): # 跳过隐藏目录 dirs[:] [d for d in dirs if not d.startswith(.)] for file in files: if not fnmatch.fnmatch(file, file_pattern): continue file_path os.path.join(root, file) try: with open(file_path, r, encodingutf-8) as f: content f.read() matches list(pattern.finditer(content)) if matches: relative_path os.path.relpath(file_path, self.base_path) results.append({ file: relative_path, match_count: len(matches), sample_matches: [m.group() for m in matches[:3]] }) if len(results) max_results: break except (UnicodeDecodeError, PermissionError): continue if len(results) max_results: break return [TextContent( typetext, textjson.dumps(results, ensure_asciiFalse, indent2) )]4.4 集成MCP服务器主程序创建src/file_server/server.pyimport asyncio import os from mcp import MCPServer from mcp.server.stdio import stdio_server from .tools.file_tools import FileReadTool, FileListTool from .tools.search_tools import FileSearchTool class FileMCPServer: def __init__(self, base_path: str .): self.base_path os.path.abspath(base_path) self.server MCPServer(file-mcp-server) # 注册工具 self.server.register_tool(FileReadTool(base_path)) self.server.register_tool(FileListTool(base_path)) self.server.register_tool(FileSearchTool(base_path)) async def run(self): 启动MCP服务器 async with stdio_server() as (read_stream, write_stream): await self.server.run( read_stream, write_stream, self.server.create_initialization_options() ) async def main(): # 从环境变量获取基础路径默认为当前目录 base_path os.getenv(MCP_BASE_PATH, .) server FileMCPServer(base_path) await server.run() if __name__ __main__: asyncio.run(main())5. 客户端集成与模型调用实战5.1 创建Python客户端示例import asyncio import os from mcp.client import create_session from mcp.client.stdio import stdio_client from openai import OpenAI class MCPClient: def __init__(self, server_path: str, model: str gpt-4): self.server_path server_path self.model model self.client OpenAI() async def query_with_tools(self, prompt: str) - str: 使用MCP工具增强的模型查询 # 第一步连接到MCP服务器并获取可用工具 async with stdio_client(self.server_path) as (read, write): async with create_session(read, write) as session: # 初始化连接 init_result await session.initialize() tools init_result.available_tools # 构建工具描述供模型使用 tool_descriptions [] for tool in tools: tool_descriptions.append({ type: function, function: { name: tool.name, description: tool.description, parameters: tool.input_schema } }) # 调用OpenAI API传入工具信息 response self.client.chat.completions.create( modelself.model, messages[{role: user, content: prompt}], toolstool_descriptions, tool_choiceauto ) message response.choices[0].message # 如果模型选择使用工具 if message.tool_calls: tool_results [] for tool_call in message.tool_calls: tool_name tool_call.function.name arguments json.loads(tool_call.function.arguments) # 通过MCP会话执行工具调用 result await session.call_tool( tool_name, arguments ) tool_results.append({ tool: tool_name, result: result.content }) # 将工具结果返回给模型进行最终回答 second_response self.client.chat.completions.create( modelself.model, messages[ {role: user, content: prompt}, message, { role: tool, content: json.dumps(tool_results), tool_call_id: tool_call.id } ] ) return second_response.choices[0].message.content else: return message.content # 使用示例 async def main(): client MCPClient( server_pathpython src/file_server/server.py, modelgpt-4 ) result await client.query_with_tools( 请帮我查找项目中有多少Python文件包含了def main这个模式 ) print(result) if __name__ __main__: asyncio.run(main())5.2 环境配置与安全设置创建.env配置文件# MCP服务器配置 MCP_BASE_PATH./workspace MCP_SERVER_LOG_LEVELINFO # OpenAI配置 OPENAI_API_KEYyour_api_key_here OPENAI_BASE_URLhttps://api.openai.com/v1 # 安全配置 MAX_FILE_SIZE10485760 # 10MB ALLOWED_EXTENSIONS.py,.txt,.md,.json,.yaml,.yml创建安全配置检查工具import os from pathlib import Path class SecurityValidator: def __init__(self, base_path: str, max_size: int 10*1024*1024): self.base_path Path(base_path).resolve() self.max_size max_size self.allowed_extensions {.py, .txt, .md, .json, .yaml, .yml} def validate_path(self, user_path: str) - Path: 验证路径安全性 try: full_path (self.base_path / user_path).resolve() # 检查是否在允许的基础路径内 if not str(full_path).startswith(str(self.base_path)): raise SecurityError(路径遍历攻击检测) # 检查文件扩展名 if full_path.is_file(): suffix full_path.suffix.lower() if suffix not in self.allowed_extensions: raise SecurityError(f不允许的文件类型: {suffix}) return full_path except Exception as e: raise SecurityError(f路径验证失败: {str(e)}) def validate_file_size(self, file_path: Path) - bool: 验证文件大小 if file_path.is_file(): size file_path.stat().st_size if size self.max_size: raise SecurityError(f文件过大: {size}字节) return True class SecurityError(Exception): 安全异常 pass6. 高级特性资源管理与实时更新6.1 实现资源提供器除了工具调用MCP还支持资源管理让模型能够订阅数据源的实时更新from mcp import Resource, ResourceTemplate from typing import List, Dict, Any import asyncio class DatabaseResource(Resource): def __init__(self, connection_string: str): self.connection_string connection_string property def uri(self) - str: return fdb://{hash(self.connection_string)}/schema property def name(self) - str: return database_schema property def description(self) - str: return 数据库表结构信息 async def read(self) - str: 读取当前数据库模式 # 模拟数据库连接和模式读取 import json schema { tables: [ { name: users, columns: [id, name, email, created_at] }, { name: orders, columns: [id, user_id, amount, status] } ] } return json.dumps(schema, indent2) class ResourceManager: def __init__(self): self.resources: Dict[str, Resource] {} def register_resource(self, resource: Resource): self.resources[resource.uri] resource async def get_resource(self, uri: str) - str: if uri not in self.resources: raise ValueError(f资源不存在: {uri}) return await self.resources[uri].read()6.2 实现资源订阅机制from typing import AsyncIterator import asyncio class ResourceSubscription: def __init__(self, resource: Resource, interval: int 30): self.resource resource self.interval interval self.subscribers set() async def notify_subscribers(self, content: str): 通知所有订阅者资源更新 for subscriber in self.subscribers: await subscriber(content) async def watch(self) - AsyncIterator[str]: 监视资源变化 last_content None while True: current_content await self.resource.read() if current_content ! last_content: last_content current_content await self.notify_subscribers(current_content) yield current_content await asyncio.sleep(self.interval)7. 生产环境部署与监控7.1 Docker容器化部署创建DockerfileFROM python:3.9-slim WORKDIR /app # 安装系统依赖 RUN apt-get update apt-get install -y \ gcc \ rm -rf /var/lib/apt/lists/* # 复制依赖文件 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # 复制应用代码 COPY src/ ./src/ # 创建非root用户 RUN useradd -m -u 1000 mcpuser USER mcpuser # 设置环境变量 ENV PYTHONPATH/app/src ENV MCP_BASE_PATH/app/workspace # 创建数据目录 RUN mkdir -p /app/workspace CMD [python, -m, src.file_server.server]创建docker-compose.ymlversion: 3.8 services: mcp-file-server: build: . ports: - 8000:8000 volumes: - ./workspace:/app/workspace - ./logs:/app/logs environment: - MCP_BASE_PATH/app/workspace - MCP_SERVER_LOG_LEVELINFO restart: unless-stopped mcp-monitor: image: prom/prometheus:latest ports: - 9090:9090 volumes: - ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml depends_on: - mcp-file-server7.2 监控与日志配置创建日志配置logging_config.pyimport logging import logging.config import json LOGGING_CONFIG { version: 1, disable_existing_loggers: False, formatters: { detailed: { format: %(asctime)s - %(name)s - %(levelname)s - %(message)s }, json: { format: %(asctime)s %(name)s %(levelname)s %(message)s, class: pythonjsonlogger.jsonlogger.JsonFormatter } }, handlers: { file: { class: logging.handlers.RotatingFileHandler, filename: /app/logs/mcp_server.log, maxBytes: 10485760, # 10MB backupCount: 5, formatter: detailed }, console: { class: logging.StreamHandler, formatter: detailed } }, loggers: { mcp: { handlers: [file, console], level: INFO, propagate: False } } } def setup_logging(): logging.config.dictConfig(LOGGING_CONFIG)8. 性能优化与最佳实践8.1 工具调用优化策略from functools import lru_cache import asyncio from concurrent.futures import ThreadPoolExecutor class ToolExecutor: def __init__(self, max_workers: int 4): self.thread_pool ThreadPoolExecutor(max_workersmax_workers) self.cache {} lru_cache(maxsize1000) async def execute_tool(self, tool_name: str, arguments: str) - str: 带缓存的工具执行 cache_key f{tool_name}:{arguments} if cache_key in self.cache: return self.cache[cache_key] # 在实际线程池中执行阻塞操作 loop asyncio.get_event_loop() result await loop.run_in_executor( self.thread_pool, self._execute_sync_tool, tool_name, arguments ) self.cache[cache_key] result return result def _execute_sync_tool(self, tool_name: str, arguments: str) - str: 同步执行工具在线程池中运行 # 实际工具执行逻辑 time.sleep(0.1) # 模拟耗时操作 return fResult for {tool_name} with {arguments}8.2 错误处理与重试机制import asyncio from typing import Type, Tuple from tenacity import retry, stop_after_attempt, wait_exponential class ResilientMCPClient: def __init__(self, max_retries: int 3): self.max_retries max_retries retry( stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10) ) async def call_tool_with_retry(self, tool_name: str, arguments: dict): 带重试的工具调用 try: # 工具调用逻辑 return await self._call_tool(tool_name, arguments) except Exception as e: if self._is_retryable_error(e): raise # 触发重试 else: raise # 非重试性错误直接抛出 def _is_retryable_error(self, error: Exception) - bool: 判断错误是否可重试 retryable_errors ( ConnectionError, TimeoutError, asyncio.TimeoutError ) return isinstance(error, retryable_errors)9. 实际项目集成案例9.1 与LangChain集成示例from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from mcp import LangChainToolAdapter class MCPLangChainIntegration: def __init__(self, mcp_server_path: str, model: str gpt-4): self.mcp_tools self._load_mcp_tools(mcp_server_path) self.llm ChatOpenAI(modelmodel) self.agent self._create_agent() def _load_mcp_tools(self, server_path: str) - list: 将MCP工具转换为LangChain工具格式 tools [] # 模拟工具加载 tools.append(LangChainToolAdapter( nameread_file, description读取文件内容, funcself._read_file_wrapper )) return tools def _create_agent(self) - AgentExecutor: 创建LangChain代理 prompt ChatPromptTemplate.from_messages([ (system, 你是一个有帮助的AI助手可以使用工具来回答问题。), (placeholder, {chat_history}), (human, {input}), (placeholder, {agent_scratchpad}) ]) agent create_tool_calling_agent( self.llm, self.mcp_tools, prompt ) return AgentExecutor( agentagent, toolsself.mcp_tools, verboseTrue ) async def run_query(self, query: str) - str: 执行查询 result await self.agent.ainvoke({input: query}) return result[output]通过以上完整的MCP程序构建教程你不仅能够理解协议的核心概念更重要的是掌握了从开发到部署的全流程实践。MCP的真正价值在于它为大模型应用提供了一种标准化、安全可控的外部能力集成方案这在AI工程化日益重要的今天具有重大意义。建议在实际项目中从小规模开始先实现核心的文件操作工具再逐步扩展数据库访问、API集成等复杂功能。记得始终把安全性放在首位特别是在处理用户输入和文件路径时要做好充分的验证和限制。
