Skip to content

第26天:MCP模块总结与项目

学习目标

  • 回顾MCP模块的学习内容
  • 掌握MCP开发的最佳实践
  • 完成文件操作工具项目
  • 了解未来学习方向

核心内容

1. MCP模块回顾

1.1 学习内容总览

MCP模块8天学习内容

天数主题核心内容
第19天MCP协议深度解析协议原理、核心概念、架构设计、通信流程、消息格式
第20天MCP Server开发基础Server架构、资源定义、工具定义、请求处理、响应格式化
第21天MCP工具开发工具设计、参数验证、错误处理、工具注册、工具执行
第22天MCP资源管理资源定义、访问控制、权限管理、状态管理、版本控制
第23天MCP Client开发Client架构、连接管理、工具发现、工具调用、响应处理
第24天MCP与Claude Desktop集成注册流程、安全配置、测试集成、高级功能、最佳实践
第25天MCP生态与工具生态系统、主流工具、框架与库、市场分析、工具选择
第26天MCP模块总结与项目模块回顾、最佳实践、文件操作工具项目、未来学习方向

1.2 技术栈回顾

MCP模块技术栈

类别技术用途
协议技术MCP ProtocolModel Context Protocol
通信协议HTTP/HTTPS标准通信协议
远程调用JSON-RPC远程过程调用
实时通信WebSocket实时双向通信
后端框架FastAPI高性能Web框架
前端框架Claude DesktopAI模型集成
编程语言Python主要开发语言
数据格式JSON数据交换格式
认证授权API Key, JWT安全认证
工具库requests, json, os功能实现

1.3 能力提升

通过MCP模块学习,你应该具备以下能力

  1. 协议理解

    • 理解MCP协议的核心概念和工作原理
    • 掌握MCP消息格式和通信流程
    • 能够分析和调试MCP通信问题
  2. Server开发

    • 能够使用FastAPI开发MCP Server
    • 能够定义和实现MCP工具
    • 能够处理MCP请求和格式化响应
    • 能够实现资源管理和访问控制
  3. Client开发

    • 能够开发MCP Client
    • 能够与MCP Server建立连接
    • 能够发现和调用MCP工具
    • 能够处理MCP响应和错误
  4. 集成应用

    • 能够将MCP工具集成到Claude Desktop
    • 能够配置MCP Server的安全设置
    • 能够测试和验证MCP集成
  5. 生态应用

    • 了解MCP生态系统的组成和发展
    • 能够选择和使用适合的MCP工具
    • 能够评估MCP工具的质量和安全性

2. MCP开发最佳实践

2.1 代码组织最佳实践

代码组织原则

  1. 模块化设计

    • 将功能划分为独立的模块
    • 每个模块负责特定的功能
    • 模块之间通过清晰的接口通信
  2. 目录结构

    mcp-server/
    ├── app/
    │   ├── api/            # API端点
    │   ├── tools/          # 工具实现
    │   ├── resources/      # 资源管理
    │   ├── models/         # 数据模型
    │   ├── schemas/        # 请求/响应模式
    │   ├── auth/           # 认证授权
    │   ├── utils/          # 工具函数
    │   └── config/         # 配置管理
    ├── main.py             # 应用入口
    ├── requirements.txt    # 依赖管理
    ├── .env                # 环境变量
    └── README.md           # 项目文档
  3. 代码风格

    • 遵循PEP 8(Python)或相应语言的代码风格指南
    • 使用一致的命名约定
    • 编写清晰、简洁的代码
    • 添加适当的注释
  4. 依赖管理

    • 使用虚拟环境隔离依赖
    • 通过requirements.txt或类似文件管理依赖
    • 锁定依赖版本,确保可重现性
    • 定期更新依赖,修复安全漏洞

2.2 安全最佳实践

安全原则

  1. 认证与授权

    • 始终使用认证机制(API Key、JWT等)
    • 实现基于角色的访问控制
    • 定期轮换认证凭证
    • 加密存储敏感信息
  2. 输入验证

    • 验证所有用户输入
    • 使用Pydantic等库进行参数验证
    • 检查文件路径,防止路径遍历攻击
    • 限制输入大小,防止DoS攻击
  3. 输出编码

    • 对输出进行适当的编码
    • 防止XSS攻击
    • 确保敏感信息不被泄露
  4. 网络安全

    • 在生产环境中使用HTTPS
    • 配置适当的CORS策略
    • 限制允许的HTTP方法
    • 实现请求频率限制
  5. 日志与监控

    • 记录所有重要操作和错误
    • 监控系统健康状态
    • 实现异常检测
    • 定期审计日志
  6. 错误处理

    • 优雅处理错误
    • 不向用户暴露详细的错误信息
    • 记录详细的错误日志用于调试
    • 实现错误恢复机制

2.3 性能最佳实践

性能优化原则

  1. 代码优化

    • 使用异步操作处理并发请求
    • 优化循环和条件判断
    • 减少不必要的计算和I/O操作
    • 使用适当的数据结构
  2. 缓存策略

    • 缓存工具列表和频繁使用的结果
    • 使用内存缓存或Redis等缓存系统
    • 实现合理的缓存过期策略
    • 缓存计算密集型操作的结果
  3. 网络优化

    • 使用HTTP/2或HTTP/3
    • 实现连接池,复用TCP连接
    • 减少请求大小,压缩响应
    • 使用CDN分发静态内容
  4. 数据库优化

    • 使用索引加速查询
    • 优化SQL查询
    • 实现数据库连接池
    • 定期清理和优化数据库
  5. 资源管理

    • 合理分配内存和CPU资源
    • 限制并发连接数
    • 监控资源使用情况
    • 实现资源配额和限制
  6. 部署优化

    • 使用容器化部署
    • 实现负载均衡
    • 配置自动扩展
    • 优化启动时间

2.4 可靠性最佳实践

可靠性原则

  1. 错误处理

    • 处理所有可能的异常
    • 实现重试机制
    • 优雅降级,当部分功能不可用时提供替代方案
    • 实现断路器模式,防止级联失败
  2. 监控与告警

    • 监控系统健康状态
    • 设置合理的告警阈值
    • 实现多渠道告警(邮件、短信、Slack等)
    • 定期检查监控配置
  3. 备份与恢复

    • 定期备份数据
    • 测试备份恢复流程
    • 实现灾难恢复计划
    • 存储备份在多个位置
  4. 配置管理

    • 使用环境变量管理配置
    • 实现配置版本控制
    • 区分开发、测试和生产环境的配置
    • 定期审查配置变更
  5. 文档与知识管理

    • 编写详细的API文档
    • 记录系统架构和设计决策
    • 维护操作手册和故障处理指南
    • 定期更新文档
  6. 测试策略

    • 实现单元测试
    • 编写集成测试
    • 执行负载测试
    • 进行安全测试

3. 实战:文件操作工具项目

3.1 项目概述

项目名称:MCP文件操作工具集

项目目标

  • 开发完整的MCP文件操作工具集
  • 实现文件的读取、写入、修改、删除等操作
  • 提供安全、可靠、高效的文件操作能力
  • 集成到Claude Desktop中使用

项目技术栈

  • 后端:Python 3.8+, FastAPI
  • 通信:HTTP/HTTPS, JSON-RPC
  • 认证:API Key
  • 安全:输入验证、路径安全检查
  • 部署:本地或云服务器

3.2 项目结构

项目目录结构

mcp-file-tools/
├── app/
│   ├── api/
│   │   ├── __init__.py
│   │   ├── mcp.py          # MCP端点
│   │   └── health.py       # 健康检查
│   ├── tools/
│   │   ├── __init__.py
│   │   ├── base.py         # 工具基类
│   │   ├── file_read.py    # 文件读取工具
│   │   ├── file_write.py   # 文件写入工具
│   │   ├── file_append.py  # 文件追加工具
│   │   ├── file_delete.py  # 文件删除工具
│   │   ├── file_copy.py    # 文件复制工具
│   │   ├── file_move.py    # 文件移动工具
│   │   ├── file_list.py    # 文件列表工具
│   │   └── file_stat.py    # 文件状态工具
│   ├── schemas/
│   │   ├── __init__.py
│   │   ├── mcp.py          # MCP请求/响应模式
│   │   └── file.py         # 文件工具参数模式
│   ├── auth/
│   │   ├── __init__.py
│   │   └── api_key.py      # API Key认证
│   ├── utils/
│   │   ├── __init__.py
│   │   ├── security.py     # 安全工具
│   │   └── file.py         # 文件操作工具
│   └── config/
│       ├── __init__.py
│       └── settings.py     # 配置管理
├── main.py                 # 应用入口
├── requirements.txt        # 依赖管理
├── .env                    # 环境变量
├── Dockerfile              # Docker配置
├── docker-compose.yml      # Docker Compose配置
└── README.md               # 项目文档

3.3 核心实现

配置管理

python
# app/config/settings.py
from pydantic_settings import BaseSettings
from typing import List

class Settings(BaseSettings):
    """应用配置"""
    # 服务器配置
    HOST: str = "0.0.0.0"
    PORT: int = 8000
    DEBUG: bool = True
    
    # 认证配置
    API_KEY: str = "your-secret-api-key"
    
    # 安全配置
    ALLOWED_DIRS: List[str] = ["/tmp", "./data"]
    MAX_FILE_SIZE: int = 10 * 1024 * 1024  # 10MB
    MAX_REQUEST_SIZE: int = 100 * 1024 * 1024  # 100MB
    
    # 缓存配置
    CACHE_TTL: int = 3600  # 1小时
    
    # 日志配置
    LOG_LEVEL: str = "INFO"
    
    class Config:
        env_file = ".env"
        case_sensitive = True

settings = Settings()

工具基类

python
# app/tools/base.py
from abc import ABC, abstractmethod
from typing import Dict, Any, List, Optional

class MCPTool(ABC):
    """MCP工具基类"""
    
    name: str
    description: str
    parameters: List[Dict[str, Any]]
    
    def __init__(self):
        self.name = self.__class__.__name__.replace("Tool", "").lower()
        self.description = ""
        self.parameters = []
    
    def add_parameter(self, name: str, type: str, description: str, 
                     required: bool = False, default: Any = None):
        """添加参数"""
        param = {
            "name": name,
            "type": type,
            "description": description,
            "required": required
        }
        if default is not None:
            param["default"] = default
        self.parameters.append(param)
    
    @abstractmethod
    def execute(self, **kwargs) -> Dict[str, Any]:
        """执行工具"""
        pass
    
    def get_tool_info(self) -> Dict[str, Any]:
        """获取工具信息"""
        return {
            "name": self.name,
            "description": self.description,
            "parameters": self.parameters
        }

文件读取工具

python
# app/tools/file_read.py
from app.tools.base import MCPTool
from app.utils.security import validate_file_path
from app.config.settings import settings
import os

class FileReadTool(MCPTool):
    """文件读取工具"""
    
    def __init__(self):
        super().__init__()
        self.name = "file.read"
        self.description = "读取文件内容"
        self.add_parameter("file_path", "string", "文件路径", required=True)
        self.add_parameter("encoding", "string", "文件编码", default="utf-8")
    
    def execute(self, file_path: str, encoding: str = "utf-8") -> Dict[str, Any]:
        """执行文件读取"""
        try:
            # 验证文件路径
            if not validate_file_path(file_path, settings.ALLOWED_DIRS):
                return {
                    "status": "error",
                    "error": "File path not allowed"
                }
            
            # 检查文件大小
            if os.path.getsize(file_path) > settings.MAX_FILE_SIZE:
                return {
                    "status": "error",
                    "error": f"File size exceeds limit ({settings.MAX_FILE_SIZE} bytes)"
                }
            
            # 读取文件
            with open(file_path, "r", encoding=encoding) as f:
                content = f.read()
            
            return {
                "status": "success",
                "output": content
            }
        except FileNotFoundError:
            return {
                "status": "error",
                "error": "File not found"
            }
        except PermissionError:
            return {
                "status": "error",
                "error": "Permission denied"
            }
        except Exception as e:
            return {
                "status": "error",
                "error": str(e)
            }

文件写入工具

python
# app/tools/file_write.py
from app.tools.base import MCPTool
from app.utils.security import validate_file_path
from app.config.settings import settings
import os

class FileWriteTool(MCPTool):
    """文件写入工具"""
    
    def __init__(self):
        super().__init__()
        self.name = "file.write"
        self.description = "写入文件内容"
        self.add_parameter("file_path", "string", "文件路径", required=True)
        self.add_parameter("content", "string", "文件内容", required=True)
        self.add_parameter("encoding", "string", "文件编码", default="utf-8")
    
    def execute(self, file_path: str, content: str, encoding: str = "utf-8") -> Dict[str, Any]:
        """执行文件写入"""
        try:
            # 验证文件路径
            if not validate_file_path(file_path, settings.ALLOWED_DIRS):
                return {
                    "status": "error",
                    "error": "File path not allowed"
                }
            
            # 检查内容大小
            if len(content) > settings.MAX_FILE_SIZE:
                return {
                    "status": "error",
                    "error": f"Content size exceeds limit ({settings.MAX_FILE_SIZE} bytes)"
                }
            
            # 确保目录存在
            os.makedirs(os.path.dirname(file_path), exist_ok=True)
            
            # 写入文件
            with open(file_path, "w", encoding=encoding) as f:
                f.write(content)
            
            return {
                "status": "success",
                "output": f"File written successfully: {file_path}"
            }
        except PermissionError:
            return {
                "status": "error",
                "error": "Permission denied"
            }
        except Exception as e:
            return {
                "status": "error",
                "error": str(e)
            }

安全工具

python
# app/utils/security.py
import os
from typing import List

def validate_file_path(file_path: str, allowed_dirs: List[str]) -> bool:
    """验证文件路径"""
    # 防止路径遍历
    if ".." in file_path:
        return False
    
    # 规范化路径
    normalized_path = os.path.normpath(file_path)
    
    # 检查是否在允许的目录中
    for allowed_dir in allowed_dirs:
        allowed_dir_norm = os.path.normpath(allowed_dir)
        if normalized_path.startswith(allowed_dir_norm):
            return True
    
    return False

def sanitize_filename(filename: str) -> str:
    """清理文件名"""
    # 移除危险字符
    dangerous_chars = '<>"|?*'
    for char in dangerous_chars:
        filename = filename.replace(char, '')
    
    # 移除路径分隔符
    filename = filename.replace('/', '').replace('\\', '')
    
    return filename

def is_safe_content(content: str) -> bool:
    """检查内容是否安全"""
    # 这里可以添加内容安全检查逻辑
    # 例如检查是否包含恶意代码、敏感信息等
    return True

MCP端点

python
# app/api/mcp.py
from fastapi import APIRouter, Depends, HTTPException
from app.schemas.mcp import MCPRequest, MCPResponse
from app.auth.api_key import get_api_key
from app.tools import get_tool_registry

router = APIRouter()

@router.post("/mcp", response_model=MCPResponse)
async def handle_mcp(request: MCPRequest, api_key: str = Depends(get_api_key)):
    """处理MCP请求"""
    tool_registry = get_tool_registry()
    method = request.method
    params = request.params
    
    if method == "mcp.list_tools":
        # 返回工具列表
        tools = [tool.get_tool_info() for tool in tool_registry.values()]
        return MCPResponse(
            jsonrpc="2.0",
            result={"tools": tools},
            id=request.id
        )
    elif method == "mcp.call_tool":
        # 处理工具调用
        tool_name = params.get("tool_name")
        arguments = params.get("arguments", {})
        
        if tool_name not in tool_registry:
            return MCPResponse(
                jsonrpc="2.0",
                result={
                    "status": "error",
                    "tool_name": tool_name,
                    "error": "Tool not found"
                },
                id=request.id
            )
        
        # 执行工具
        tool = tool_registry[tool_name]
        result = tool.execute(**arguments)
        
        return MCPResponse(
            jsonrpc="2.0",
            result={
                **result,
                "tool_name": tool_name
            },
            id=request.id
        )
    else:
        return MCPResponse(
            jsonrpc="2.0",
            result={
                "status": "error",
                "error": "Method not found"
            },
            id=request.id
        )

工具注册

python
# app/tools/__init__.py
from app.tools.file_read import FileReadTool
from app.tools.file_write import FileWriteTool
from app.tools.file_append import FileAppendTool
from app.tools.file_delete import FileDeleteTool
from app.tools.file_copy import FileCopyTool
from app.tools.file_move import FileMoveTool
from app.tools.file_list import FileListTool
from app.tools.file_stat import FileStatTool
from typing import Dict

_tool_registry: Dict[str, any] = {}

def get_tool_registry() -> Dict[str, any]:
    """获取工具注册表"""
    global _tool_registry
    
    if not _tool_registry:
        # 注册工具
        tools = [
            FileReadTool(),
            FileWriteTool(),
            FileAppendTool(),
            FileDeleteTool(),
            FileCopyTool(),
            FileMoveTool(),
            FileListTool(),
            FileStatTool()
        ]
        
        for tool in tools:
            _tool_registry[tool.name] = tool
    
    return _tool_registry

应用入口

python
# main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api import mcp, health
from app.config.settings import settings
import logging

# 配置日志
logging.basicConfig(
    level=getattr(logging, settings.LOG_LEVEL),
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# 创建FastAPI应用
app = FastAPI(
    title="MCP File Tools",
    version="1.0.0",
    description="MCP File Operations Tools Server"
)

# 配置CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 在生产环境中应该设置具体的域名
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 注册路由
app.include_router(health.router)
app.include_router(mcp.router)

# 启动应用
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(
        "main:app",
        host=settings.HOST,
        port=settings.PORT,
        reload=settings.DEBUG
    )

3.4 部署与运行

依赖管理

txt
# requirements.txt
fastapi>=0.100.0
uvicorn>=0.20.0
pydantic>=2.0.0
pydantic-settings>=2.0.0
python-dotenv>=1.0.0

环境变量

bash
# .env
API_KEY=your-secret-api-key
ALLOWED_DIRS=/tmp,./data
MAX_FILE_SIZE=10485760
DEBUG=True

Docker配置

dockerfile
# Dockerfile
FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["python", "main.py"]

Docker Compose配置

yaml
# docker-compose.yml
version: "3.8"

services:
  mcp-file-tools:
    build: .
    ports:
      - "8000:8000"
    volumes:
      - ./data:/app/data
      - /tmp:/tmp
    environment:
      - API_KEY=your-secret-api-key
      - ALLOWED_DIRS=/tmp,/app/data
      - MAX_FILE_SIZE=10485760
      - DEBUG=False
    restart: unless-stopped

运行项目

bash
# 本地运行
pip install -r requirements.txt
python main.py

# 使用Docker运行
docker-compose up -d

# 检查运行状态
docker-compose ps

# 查看日志
docker-compose logs

测试项目

bash
# 测试工具列表
curl -X POST http://localhost:8000/mcp \
  -H "Content-Type: application/json" \
  -H "X-API-Key: your-secret-api-key" \
  -d '{"jsonrpc": "2.0", "method": "mcp.list_tools", "params": {}, "id": 1}'

# 测试文件写入
curl -X POST http://localhost:8000/mcp \
  -H "Content-Type: application/json" \
  -H "X-API-Key: your-secret-api-key" \
  -d '{"jsonrpc": "2.0", "method": "mcp.call_tool", "params": {"tool_name": "file.write", "arguments": {"file_path": "/tmp/test.txt", "content": "Hello, MCP!"}}, "id": 2}'

# 测试文件读取
curl -X POST http://localhost:8000/mcp \
  -H "Content-Type: application/json" \
  -H "X-API-Key: your-secret-api-key" \
  -d '{"jsonrpc": "2.0", "method": "mcp.call_tool", "params": {"tool_name": "file.read", "arguments": {"file_path": "/tmp/test.txt"}}, "id": 3}'

3.5 集成到Claude Desktop

步骤1:启动MCP File Tools Server

bash
docker-compose up -d

步骤2:在Claude Desktop中注册工具

  • 打开Claude Desktop
  • 进入工具注册界面
  • 配置以下信息:
  • 测试连接并保存配置

步骤3:测试文件工具

在Claude Desktop中测试文件工具:

请帮我在/tmp目录中创建一个名为test.txt的文件,内容为"Hello from Claude!"

请帮我读取/tmp/test.txt文件的内容

请帮我列出/tmp目录中的文件

请帮我删除/tmp/test.txt文件

4. 未来学习方向

4.1 进阶MCP开发

进阶学习内容

  1. 高级工具开发

    • 开发复杂的多步骤工具
    • 实现工具链和工具组合
    • 开发AI辅助工具
  2. MCP Server高级特性

    • 实现负载均衡
    • 开发高可用架构
    • 实现分布式MCP Server
  3. MCP Client高级特性

    • 开发智能Client
    • 实现Client缓存和预加载
    • 开发多平台Client
  4. 安全性进阶

    • 实现OAuth2认证
    • 开发细粒度权限控制
    • 实现审计日志系统
  5. 性能优化进阶

    • 实现异步工具执行
    • 开发高性能文件系统
    • 优化网络传输

4.2 相关技术学习

相关技术

  1. AI技术

    • 大型语言模型原理
    • prompt工程
    • 微调技术
    • 多模态AI
  2. Web技术

    • 现代前端框架(React、Vue、Angular)
    • WebSocket和实时通信
    • GraphQL和API设计
    • 微服务架构
  3. 云原生技术

    • Docker和容器编排
    • Kubernetes
    • 云服务(AWS、Azure、GCP)
    • Serverless架构
  4. DevOps技术

    • CI/CD流水线
    • 自动化测试
    • 监控和告警
    • 基础设施即代码
  5. 数据技术

    • 数据库设计和优化
    • 大数据处理
    • 数据湖和数据仓库
    • 数据可视化

4.3 职业发展方向

MCP相关职业方向

  1. AI工具开发者

    • 开发和维护MCP工具
    • 优化工具性能和安全性
    • 与AI模型团队合作
  2. MCP架构师

    • 设计MCP系统架构
    • 制定MCP开发规范
    • 评估和选择MCP技术
  3. AI集成工程师

    • 将AI模型集成到现有系统
    • 开发AI助手和聊天机器人
    • 优化AI系统性能
  4. DevOps工程师

    • 部署和管理MCP服务
    • 实现自动化运维
    • 监控和故障处理
  5. 解决方案架构师

    • 设计基于MCP的解决方案
    • 与客户沟通需求
    • 提供技术咨询

5. 总结

本课程是MCP模块的最后一天,我们回顾了整个MCP模块的学习内容,总结了MCP开发的最佳实践,并完成了一个完整的文件操作工具项目。通过这个项目,你应该能够:

  • 理解MCP项目的完整开发流程
  • 掌握MCP工具的设计和实现
  • 学会部署和运行MCP服务
  • 能够将MCP工具集成到Claude Desktop

MCP技术作为AI模型与工具交互的重要协议,具有广阔的应用前景。随着AI技术的不断发展,MCP的重要性将日益凸显。通过本模块的学习,你已经掌握了MCP开发的核心技能,为未来的AI应用开发打下了坚实的基础。

在未来的学习和工作中,你可以:

  • 继续深入学习MCP的高级特性
  • 探索MCP与其他技术的集成
  • 开发创新的MCP工具和应用
  • 参与MCP生态系统的建设

MCP模块的学习虽然结束了,但AI技术的发展永无止境。希望你能够保持学习的热情,不断探索AI技术的新领域,为AI技术的发展做出贡献。


课后作业

  1. 实践题

    • 完成文件操作工具项目的开发和部署
    • 测试所有文件工具的功能
    • 将工具集成到Claude Desktop中使用
    • 编写项目文档和使用指南
  2. 思考题

    • 如何进一步提高文件操作工具的安全性?
    • 如何优化文件操作工具的性能?
    • 如何扩展文件操作工具集,添加更多功能?
    • 如何将文件操作工具部署到生产环境?
  3. 总结题

    • 总结MCP模块的学习收获
    • 分析MCP技术的优势和局限性
    • 展望MCP技术的未来发展
    • 制定个人MCP技术学习计划

架构师AI杜公众号二维码

扫描二维码关注"架构师AI杜"公众号,获取更多技术内容和最新动态