06 - MCP 协议完全指南

🔌 Model Context Protocol：打通 AI 与任何系统的桥梁

📋 本章目标

✅ 理解 MCP 的设计理念和架构
✅ 掌握 MCP Server 的开发
✅ 学会常用 MCP 集成
✅ 了解 MCP 最佳实践和安全性

预计阅读时间: 40分钟

🤔 什么是 MCP?

Model Context Protocol (MCP)

MCP 是 Anthropic 开发的开放协议，用于连接 AI 模型与外部系统。

简单理解:
MCP = AI 的"USB 接口"

就像 USB 让你的电脑连接任何设备（鼠标、键盘、打印机...）,
MCP 让 Claude 连接任何系统（数据库、API、工具...）

为什么需要 MCP?

问题：AI 的局限性

传统 AI:
┌─────────────┐
│   Claude    │  只能：
│             │  - 读取你提供的文本
│   (孤岛)    │  - 生成回复
└─────────────┘  
     ❌ 无法访问外部数据
     ❌ 无法执行操作
     ❌ 无法持久化状态

解决：MCP 连接一切

Claude + MCP:
┌─────────────┐
│   Claude    │
└──────┬──────┘
       │ MCP
       ├──────→ 数据库 (读写数据)
       ├──────→ API (调用服务)
       ├──────→ 文件系统 (管理文件)
       ├──────→ 浏览器 (自动化测试)
       └──────→ 任何你想连接的系统!

MCP vs 传统方案

特性	MCP	传统 API	传统插件
标准化	✅ 统一协议	❌ 各自标准	❌ 平台特定
双向通信	✅ 是	🟡 有限	✅ 是
动态能力	✅ 运行时发现	❌ 需预定义	🟡 部分
安全性	✅ 权限系统	🟡 API Key	🟡 沙盒
易扩展	✅ 插拔式	❌ 需集成	🟡 中等

🏗️ MCP 架构

核心组件

┌──────────────────────────────────────────────┐
│                 Claude (Client)               │
│  - 理解用户意图                               │
│  - 规划任务                                   │
│  - 调用 MCP Server                            │
└───────────────────┬──────────────────────────┘
                    │ MCP Protocol
                    ↓
┌──────────────────────────────────────────────┐
│              MCP Transport Layer              │
│  - stdio (标准输入输出)                        │
│  - HTTP/WebSocket                             │
│  - 序列化/反序列化                             │
└───────────────────┬──────────────────────────┘
                    │
        ┌───────────┴───────────┬───────────┐
        ↓                       ↓           ↓
┌──────────────┐     ┌──────────────┐  ┌───────────┐
│ MCP Server 1 │     │ MCP Server 2 │  │  Server N │
│  (Database)  │     │    (API)     │  │   (...)   │
└──────────────┘     └──────────────┘  └───────────┘
        │                       │           │
        ↓                       ↓           ↓
┌──────────────┐     ┌──────────────┐  ┌───────────┐
│  PostgreSQL  │     │  GitHub API  │  │  Notion   │
└──────────────┘     └──────────────┘  └───────────┘

MCP 消息流程

1. 用户: "查询数据库中最近的用户"
   ↓
2. Claude 分析: 需要数据库查询能力
   ↓
3. Claude 发现: PostgreSQL MCP Server
   ↓
4. Claude → MCP Server:
   {
     "method": "tools/call",
     "params": {
       "name": "query",
       "arguments": {
         "sql": "SELECT * FROM users ORDER BY created_at DESC LIMIT 10"
       }
     }
   }
   ↓
5. MCP Server 执行查询
   ↓
6. MCP Server → Claude:
   {
     "result": [
       { "id": 1, "name": "Alice", ... },
       { "id": 2, "name": "Bob", ... }
     ]
   }
   ↓
7. Claude → 用户: "找到了10个最近的用户：..."

🛠️ 创建 MCP Server

基础 MCP Server 结构

// server.ts

import { Server } from '@modelcontextprotocol/sdk/server/index.js'
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js'

// 1. 创建 Server 实例
const server = new Server(
  {
    name: 'my-awesome-server',
    version: '1.0.0',
  },
  {
    capabilities: {
      tools: {},  // 提供 tools
      resources: {},  // 提供 resources
    },
  }
)

// 2. 注册 Tools
server.setRequestHandler('tools/list', async () => {
  return {
    tools: [
      {
        name: 'get_weather',
        description: 'Get current weather for a location',
        inputSchema: {
          type: 'object',
          properties: {
            location: {
              type: 'string',
              description: 'City name'
            }
          },
          required: ['location']
        }
      }
    ]
  }
})

// 3. 处理 Tool 调用
server.setRequestHandler('tools/call', async (request) => {
  if (request.params.name === 'get_weather') {
    const location = request.params.arguments.location
    const weather = await fetchWeather(location)
    
    return {
      content: [
        {
          type: 'text',
          text: JSON.stringify(weather, null, 2)
        }
      ]
    }
  }
  
  throw new Error('Tool not found')
})

// 4. 启动 Server
const transport = new StdioServerTransport()
await server.connect(transport)

完整示例：数据库 MCP Server

// database-server.ts

import { Server } from '@modelcontextprotocol/sdk/server/index.js'
import { Pool } from 'pg'

const pool = new Pool({
  connectionString: process.env.DATABASE_URL
})

const server = new Server(
  { name: 'postgres-server', version: '1.0.0' },
  { capabilities: { tools: {} } }
)

// 列出可用 Tools
server.setRequestHandler('tools/list', async () => {
  return {
    tools: [
      {
        name: 'query',
        description: 'Execute a SQL query',
        inputSchema: {
          type: 'object',
          properties: {
            sql: {
              type: 'string',
              description: 'SQL query to execute'
            },
            params: {
              type: 'array',
              description: 'Query parameters',
              items: { type: 'string' }
            }
          },
          required: ['sql']
        }
      },
      {
        name: 'list_tables',
        description: 'List all tables in the database',
        inputSchema: { type: 'object', properties: {} }
      },
      {
        name: 'describe_table',
        description: 'Get table schema',
        inputSchema: {
          type: 'object',
          properties: {
            table: { type: 'string', description: 'Table name' }
          },
          required: ['table']
        }
      }
    ]
  }
})

// 处理 Tool 调用
server.setRequestHandler('tools/call', async (request) => {
  const { name, arguments: args } = request.params
  
  switch (name) {
    case 'query': {
      const { sql, params = [] } = args
      const result = await pool.query(sql, params)
      
      return {
        content: [
          {
            type: 'text',
            text: JSON.stringify({
              rows: result.rows,
              rowCount: result.rowCount
            }, null, 2)
          }
        ]
      }
    }
    
    case 'list_tables': {
      const result = await pool.query(`
        SELECT tablename 
        FROM pg_tables 
        WHERE schemaname = 'public'
      `)
      
      return {
        content: [
          {
            type: 'text',
            text: JSON.stringify(result.rows, null, 2)
          }
        ]
      }
    }
    
    case 'describe_table': {
      const { table } = args
      const result = await pool.query(`
        SELECT column_name, data_type, is_nullable
        FROM information_schema.columns
        WHERE table_name = $1
      `, [table])
      
      return {
        content: [
          {
            type: 'text',
            text: JSON.stringify(result.rows, null, 2)
          }
        ]
      }
    }
    
    default:
      throw new Error(`Unknown tool: ${name}`)
  }
})

// 启动
const transport = new StdioServerTransport()
await server.connect(transport)

配置 MCP Server

// .cursor/mcp.json 或 ~/Library/Application Support/Cursor/mcp.json

{
  "mcpServers": {
    "postgres": {
      "command": "node",
      "args": ["/path/to/database-server.js"],
      "env": {
        "DATABASE_URL": "postgresql://user:pass@localhost:5432/mydb"
      }
    },
    "weather": {
      "command": "node",
      "args": ["/path/to/weather-server.js"],
      "env": {
        "WEATHER_API_KEY": "your-api-key"
      }
    }
  }
}

🔌 常用 MCP 集成

1. 数据库集成

PostgreSQL MCP Server

// 安装
npm install @modelcontextprotocol/server-postgres

// 配置
{
  "postgres": {
    "command": "npx",
    "args": ["-y", "@modelcontextprotocol/server-postgres", "postgresql://..."]
  }
}

// 使用
Prompt: "查询 users 表中年龄>25的用户"
Claude 会: 自动生成SQL并执行

MongoDB MCP Server

{
  "mongodb": {
    "command": "node",
    "args": ["mongodb-server.js"],
    "env": {
      "MONGO_URI": "mongodb://localhost:27017/mydb"
    }
  }
}

// 使用
Prompt: "统计每个类别的文章数量"
Claude 会: 使用聚合管道查询

2. API 集成

GitHub MCP Server

// 安装官方 GitHub MCP
npm install @modelcontextprotocol/server-github

// 配置
{
  "github": {
    "command": "npx",
    "args": ["-y", "@modelcontextprotocol/server-github"],
    "env": {
      "GITHUB_TOKEN": "ghp_..."
    }
  }
}

// 使用示例
Prompt: "查看我的仓库最近的issues"
Claude: [调用 GitHub API 获取 issues]

Prompt: "为issue #123 添加标签 'bug'"
Claude: [调用 API 添加标签]

自定义 API MCP

// notion-server.ts

server.setRequestHandler('tools/list', async () => {
  return {
    tools: [
      {
        name: 'create_page',
        description: 'Create a new Notion page',
        inputSchema: {
          type: 'object',
          properties: {
            database_id: { type: 'string' },
            title: { type: 'string' },
            content: { type: 'string' }
          }
        }
      },
      {
        name: 'query_database',
        description: 'Query a Notion database',
        inputSchema: {
          type: 'object',
          properties: {
            database_id: { type: 'string' },
            filter: { type: 'object' }
          }
        }
      }
    ]
  }
})

3. 文件系统集成

// filesystem-server.ts

const tools = [
  {
    name: 'read_directory',
    description: 'List files in a directory',
    inputSchema: {
      type: 'object',
      properties: {
        path: { type: 'string' }
      }
    }
  },
  {
    name: 'search_files',
    description: 'Search files by pattern',
    inputSchema: {
      type: 'object',
      properties: {
        pattern: { type: 'string' },
        path: { type: 'string' }
      }
    }
  },
  {
    name: 'watch_directory',
    description: 'Watch directory for changes',
    inputSchema: {
      type: 'object',
      properties: {
        path: { type: 'string' }
      }
    }
  }
]

4. 浏览器自动化

// browser-server.ts (基于 Playwright)

import { chromium } from 'playwright'

server.setRequestHandler('tools/call', async (request) => {
  const browser = await chromium.launch()
  const page = await browser.newPage()
  
  const { name, arguments: args } = request.params
  
  switch (name) {
    case 'navigate':
      await page.goto(args.url)
      break
      
    case 'screenshot':
      const screenshot = await page.screenshot()
      return {
        content: [
          {
            type: 'image',
            data: screenshot.toString('base64')
          }
        ]
      }
      
    case 'fill_form':
      await page.fill(args.selector, args.value)
      break
      
    case 'click':
      await page.click(args.selector)
      break
      
    case 'extract_data':
      const data = await page.evaluate(() => {
        // 提取页面数据
        return document.body.innerText
      })
      return {
        content: [{ type: 'text', text: data }]
      }
  }
  
  await browser.close()
})

🔐 MCP 安全性

权限系统

// 定义权限级别
const PERMISSIONS = {
  READ: 'read',
  WRITE: 'write',
  ADMIN: 'admin'
}

// 在 MCP Server 中验证权限
server.setRequestHandler('tools/call', async (request, { auth }) => {
  const tool = request.params.name
  
  // 检查权限
  if (tool === 'delete_all' && auth.level !== PERMISSIONS.ADMIN) {
    throw new Error('Insufficient permissions')
  }
  
  // 执行操作
  return await executeTool(tool, request.params.arguments)
})

输入验证

import { z } from 'zod'

// 定义输入 Schema
const QuerySchema = z.object({
  sql: z.string()
    .regex(/^SELECT/i, 'Only SELECT queries allowed')
    .max(1000, 'Query too long'),
  params: z.array(z.string()).optional()
})

// 验证输入
server.setRequestHandler('tools/call', async (request) => {
  if (request.params.name === 'query') {
    try {
      const validated = QuerySchema.parse(request.params.arguments)
      // 执行验证后的查询
      return await executeQuery(validated)
    } catch (error) {
      throw new Error(`Invalid input: ${error.message}`)
    }
  }
})

速率限制

import { RateLimiter } from 'limiter'

const limiter = new RateLimiter({
  tokensPerInterval: 10,
  interval: 'minute'
})

server.setRequestHandler('tools/call', async (request) => {
  // 检查速率限制
  const allowed = await limiter.removeTokens(1)
  
  if (!allowed) {
    throw new Error('Rate limit exceeded')
  }
  
  // 执行操作
  return await handleRequest(request)
})

敏感数据保护

// 过滤敏感信息
function sanitizeResult(data: any) {
  if (typeof data === 'object') {
    const sanitized = { ...data }
    
    // 移除敏感字段
    delete sanitized.password
    delete sanitized.secret_key
    delete sanitized.api_token
    
    // 递归处理嵌套对象
    for (const key in sanitized) {
      if (typeof sanitized[key] === 'object') {
        sanitized[key] = sanitizeResult(sanitized[key])
      }
    }
    
    return sanitized
  }
  
  return data
}

// 在返回结果前过滤
server.setRequestHandler('tools/call', async (request) => {
  const result = await executeQuery(request)
  return {
    content: [
      {
        type: 'text',
        text: JSON.stringify(sanitizeResult(result))
      }
    ]
  }
})

💡 MCP 最佳实践

1. 设计原则

单一职责

// ❌ 不好的设计
{
  "all-in-one-server": {
    // 一个 Server 做所有事情
  }
}

// ✅ 好的设计
{
  "database": { ... },  // 只处理数据库
  "api": { ... },       // 只处理API调用
  "files": { ... }      // 只处理文件操作
}

清晰的命名

// ❌ 模糊的命名
{
  name: 'do_something',
  description: 'Does stuff'
}

// ✅ 清晰的命名
{
  name: 'query_users_by_email',
  description: 'Query users table filtered by email address. Returns user objects with id, name, email fields.'
}

2. 错误处理

server.setRequestHandler('tools/call', async (request) => {
  try {
    return await executeOperation(request)
  } catch (error) {
    // 详细的错误信息
    return {
      isError: true,
      content: [
        {
          type: 'text',
          text: JSON.stringify({
            error: error.message,
            code: error.code,
            suggestion: 'Please check your input parameters',
            documentation: 'https://docs.example.com/api'
          })
        }
      ]
    }
  }
})

3. 性能优化

连接池

// 使用连接池而不是每次创建新连接
const pool = new Pool({
  max: 20,
  idleTimeoutMillis: 30000,
  connectionTimeoutMillis: 2000,
})

// 复用连接
async function query(sql: string) {
  const client = await pool.connect()
  try {
    return await client.query(sql)
  } finally {
    client.release()
  }
}

缓存

import { LRUCache } from 'lru-cache'

const cache = new LRUCache({
  max: 100,
  ttl: 1000 * 60 * 5  // 5分钟
})

server.setRequestHandler('tools/call', async (request) => {
  const cacheKey = JSON.stringify(request.params)
  
  // 检查缓存
  if (cache.has(cacheKey)) {
    return cache.get(cacheKey)
  }
  
  // 执行操作
  const result = await executeOperation(request)
  
  // 存入缓存
  cache.set(cacheKey, result)
  
  return result
})

4. 日志和监控

import winston from 'winston'

const logger = winston.createLogger({
  level: 'info',
  format: winston.format.json(),
  transports: [
    new winston.transports.File({ filename: 'mcp-server.log' })
  ]
})

server.setRequestHandler('tools/call', async (request) => {
  logger.info('Tool called', {
    tool: request.params.name,
    timestamp: new Date().toISOString()
  })
  
  try {
    const result = await executeOperation(request)
    
    logger.info('Tool succeeded', {
      tool: request.params.name,
      duration: Date.now() - startTime
    })
    
    return result
  } catch (error) {
    logger.error('Tool failed', {
      tool: request.params.name,
      error: error.message
    })
    
    throw error
  }
})

🎯 实战案例

案例1: 全栈开发工作流

用户: "创建一个博客系统"

Claude 通过 MCP:
1. Database MCP → 创建数据库表
2. Filesystem MCP → 生成后端代码
3. Filesystem MCP → 生成前端代码
4. Git MCP → 初始化仓库
5. Terminal MCP → 安装依赖
6. Terminal MCP → 启动开发服务器
7. Browser MCP → 打开浏览器测试

一条命令，完整的开发环境就绪！

案例2: 数据分析流水线

用户: "分析用户行为数据并生成报告"

Claude 通过 MCP:
1. Database MCP → 从 PostgreSQL 提取数据
2. Python MCP → 运行数据分析脚本
3. Visualization MCP → 生成图表
4. Notion MCP → 创建报告页面
5. Slack MCP → 通知团队

自动化数据分析流程！

案例3: CI/CD 自动化

用户: "部署应用到生产环境"

Claude 通过 MCP:
1. Git MCP → 检查分支状态
2. Terminal MCP → 运行测试
3. Docker MCP → 构建镜像
4. Kubernetes MCP → 部署到集群
5. Monitoring MCP → 设置监控
6. Slack MCP → 发送部署通知

一键部署！

📚 MCP 生态

官方 MCP Servers

@modelcontextprotocol/server-postgres - PostgreSQL
@modelcontextprotocol/server-sqlite - SQLite
@modelcontextprotocol/server-github - GitHub API
@modelcontextprotocol/server-slack - Slack API
@modelcontextprotocol/server-google-drive - Google Drive

社区 MCP Servers

mcp-server-notion - Notion integration
mcp-server-jira - Jira/Confluence
mcp-server-aws - AWS Services
mcp-server-docker - Docker operations
mcp-server-k8s - Kubernetes

✅ 检查清单

完成本章后，你应该能够：

[ ] 理解 MCP 的架构和工作原理
[ ] 创建基本的 MCP Server
[ ] 集成常用的 MCP Servers
[ ] 实现安全的权限控制
[ ] 优化 MCP Server 性能
[ ] 设计复杂的自动化工作流

🎯 实战练习

练习1: 创建天气 MCP Server

// 要求:
- 调用天气 API
- 提供多个查询方式（城市、坐标）
- 缓存结果
- 错误处理

练习2: 数据库查询优化

// 要求:
- 实现查询缓存
- 添加连接池
- 记录慢查询日志
- 实现查询超时

练习3: 多 MCP 协作

// 场景: 自动化内容发布
1. 从 Notion 获取草稿
2. 调用 AI 生成摘要
3. 上传图片到 CDN
4. 发布到博客系统
5. 分享到社交媒体

📚 延伸阅读

🔌 MCP 是 Claude 连接世界的桥梁

掌握 MCP，你就能让 Claude 访问任何系统！

"The power of AI lies not in isolation, but in connection."