1. MCP协议核心原理

1.1 为什么需要MCP

传统AI工具调用的困境:
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

2023-2024: 各自为战
  AI助手 → 浏览器: 专用API
  AI助手 → 数据库: 专用SDK
  AI助手 → GitHub: 专用CLI
  AI助手 → 文件系统: 专用工具
  
  问题:
  ❌ 每集成一个工具要写一套适配代码
  ❌ 工具之间无法互通
  ❌ 无法动态发现工具
  ❌ 每次换AI模型要重写工具层

2025-2026: MCP统一协议 ⭐
  
  ┌─────────────┐
  │  AI Model   │
  └──────┬──────┘
         │ JSON-RPC 2.0
  ┌──────▼──────┐
  │ MCP Protocol│ ← 统一接口
  └──┬──┬──┬──┬─┘
     │  │  │  │
  ┌──▼─┐┌▼──┐┌▼──┐
  │浏览 ││数据││Git│
  │器   ││库  ││Hub│
  └──┬──┘└┬──┘└┬──┘
     │     │     │
     ▼     ▼     ▼
  Chrome  PG   API

  优势:
  ✅ 一次实现，任何AI模型可用
  ✅ 工具动态发现
  ✅ 类型安全，协议约束
  ✅ 生态复用
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

1.2 协议架构

MCP协议架构:
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

┌─────────────────────────────────────────┐
│             Host (Claude/OpenClaw)       │
│         AI模型 + MCP客户端                │
└────────────────────┬────────────────────┘
                     │ stdio / HTTP(SSE)
┌────────────────────▼────────────────────┐
│          MCP Server                       │
│    (Python/TypeScript/其他语言)            │
│                                          │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐  │
│  │Resources│ │  Tools  │ │ Prompts │  │
│  │(数据读取)│ │(操作执行)│ │(模板生成)│  │
│  └────┬────┘ └────┬────┘ └────┬────┘  │
│       └───────────┼───────────┘         │
│                   ▼                       │
│         ┌───────────────┐                │
│         │ Tool Executor │                │
│         │  (实际操作)   │                │
│         └───────┬───────┘                │
└─────────────────┼───────────────────────┘
                  │
        ┌─────────┴─────────┐
        ▼         ▼         ▼
      文件      数据库     网络
      系统      PostgreSQL  API

JSON-RPC 2.0通信格式:
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

// 请求
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "postgresql_query",
    "arguments": {
      "sql": "SELECT * FROM users WHERE id = 1"
    }
  }
}

// 响应
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "{\"id\": 1, \"name\": \"张三\", \"email\": \"zhangsan@example.com\"}"
      }
    ],
    "isError": false
  }
}
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

2. MCP Server开发实战

2.1 Python MCP Server

"""
Python MCP Server开发
使用 mcp 官方SDK
"""

from mcp.server import Server
from mcp.types import Tool, TextContent
from mcp.server.stdio import stdio_server
import asyncio
import json
from dataclasses import dataclass
from typing import Any
import httpx

# ===== 项目结构 =====
"""
mcp-server/
├── src/
│   ├── __init__.py
│   ├── server.py           # 主服务器
│   ├── tools/
│   │   ├── __init__.py
│   │   ├── web_tools.py    # Web相关工具
│   │   ├── db_tools.py     # 数据库工具
│   │   └── file_tools.py    # 文件工具
│   ├── resources/
│   │   └── __init__.py
│   └── prompts/
│       └── __init__.py
├── pyproject.toml
└── README.md
"""

# ===== 基础Server骨架 =====

# src/server.py
from mcp.server import Server
from mcp.types import (
    Tool, TextContent, Resource, Prompt,
    ListToolsResult, CallToolResult
)
from mcp.server.stdio import stdio_server
import asyncio

# 创建Server实例
app = Server("my-mcp-server")

# ===== 工具定义 =====

@app.list_tools()
async def list_tools() -> ListToolsResult:
    """列出所有可用工具"""
    return ListToolsResult(tools=[
        Tool(
            name="fetch_webpage",
            description="获取网页内容，支持提取正文",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "网页URL"
                    },
                    "extract_text": {
                        "type": "boolean",
                        "description": "是否只提取文本",
                        "default": True
                    }
                },
                "required": ["url"]
            }
        ),
        Tool(
            name="search_code",
            description="在代码仓库中搜索代码",
            inputSchema={
                "type": "object",
                "properties": {
                    "repo": {
                        "type": "string",
                        "description": "仓库名，格式: owner/repo"
                    },
                    "query": {
                        "type": "string",
                        "description": "搜索关键词"
                    },
                    "language": {
                        "type": "string",
                        "description": "编程语言过滤"
                    }
                },
                "required": ["repo", "query"]
            }
        ),
        Tool(
            name="run_sql",
            description="执行SQL查询",
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {
                        "type": "string",
                        "description": "SQL查询语句"
                    },
                    "limit": {
                        "type": "integer",
                        "description": "最大返回行数",
                        "default": 100
                    }
                },
                "required": ["sql"]
            }
        ),
        Tool(
            name="file_glob",
            description="按模式搜索文件",
            inputSchema={
                "type": "object",
                "properties": {
                    "pattern": {
                        "type": "string",
                        "description": "Glob模式，如: **/*.py"
                    },
                    "root": {
                        "type": "string",
                        "description": "搜索根目录"
                    }
                },
                "required": ["pattern"]
            }
        )
    ])


@app.call_tool()
async def call_tool(
    name: str,
    arguments: dict
) -> CallToolResult:
    """执行工具调用"""
    
    # 路由到具体工具
    tools = {
        "fetch_webpage": fetch_webpage,
        "search_code": search_code,
        "run_sql": run_sql,
        "file_glob": file_glob,
    }
    
    if name not in tools:
        return CallToolResult(
            content=[TextContent(type="text", text=f"Unknown tool: {name}")],
            isError=True
        )
    
    try:
        result = await tools[name](**arguments)
        return CallToolResult(
            content=[TextContent(type="text", text=json.dumps(result, ensure_ascii=False, indent=2))],
            isError=False
        )
    except Exception as e:
        return CallToolResult(
            content=[TextContent(type="text", text=f"Error: {str(e)}")],
            isError=True
        )


# ===== 工具实现 =====

async def fetch_webpage(url: str, extract_text: bool = True) -> dict:
    """获取网页内容"""
    async with httpx.AsyncClient(timeout=30.0) as client:
        response = await client.get(url)
        response.raise_for_status()
        
        if extract_text:
            # 简单文本提取（实际可用trafilatura等库）
            content = response.text
            # 移除脚本和样式
            import re
            content = re.sub(r'<script[^>]*>.*?</script>', '', content, flags=re.DOTALL)
            content = re.sub(r'<style[^>]*>.*?</style>', '', content, flags=re.DOTALL)
            content = re.sub(r'<[^>]+>', '', content)
            content = re.sub(r'\s+', ' ', content).strip()
            return {"url": url, "text": content[:5000]}
        else:
            return {"url": url, "html": response.text[:10000]}


async def search_code(repo: str, query: str, language: str = None) -> dict:
    """搜索GitHub代码"""
    # 实际使用GitHub API
    api_url = f"https://api.github.com/search/code"
    params = {"q": f"{query}+repo:{repo}"}
    if language:
        params["q"] += f"+language:{language}"
    
    # 注意：需要GitHub Token
    headers = {
        "Accept": "application/vnd.github.v3+json",
        # "Authorization": f"token {GITHUB_TOKEN}"
    }
    
    async with httpx.AsyncClient() as client:
        response = await client.get(api_url, params=params, headers=headers)
        response.raise_for_status()
        data = response.json()
        
        return {
            "total": data.get("total_count", 0),
            "items": [
                {"name": item["name"], "path": item["path"], "url": item["html_url"]}
                for item in data.get("items", [])[:10]
            ]
        }


async def run_sql(sql: str, limit: int = 100) -> dict:
    """执行SQL查询（示例，需要配置数据库）"""
    # 实际使用asyncpg/aiomysql
    import asyncpg
    
    conn = await asyncpg.connect(
        host="localhost",
        port=5432,
        user="postgres",
        password="password",
        database="mydb"
    )
    
    try:
        # 安全检查：只允许SELECT
        sql_upper = sql.strip().upper()
        if not sql_upper.startswith("SELECT"):
            raise ValueError("Only SELECT queries are allowed")
        
        rows = await conn.fetch(sql + f" LIMIT {limit}")
        columns = list(rows[0].keys()) if rows else []
        
        return {
            "columns": columns,
            "rows": [dict(row) for row in rows],
            "count": len(rows)
        }
    finally:
        await conn.close()


async def file_glob(pattern: str, root: str = ".") -> dict:
    """文件搜索"""
    import glob
    import os
    
    # 安全检查：限制搜索范围
    root = os.path.abspath(root)
    matches = glob.glob(os.path.join(root, pattern), recursive=True)
    
    # 过滤：只返回文件，不返回目录
    files = [m for m in matches if os.path.isfile(m)]
    
    return {
        "pattern": pattern,
        "root": root,
        "files": files[:100]  # 限制返回数量
    }


# ===== 主入口 =====

async def main():
    """启动MCP Server"""
    async with stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

if __name__ == "__main__":
    asyncio.run(main())

2.2 TypeScript MCP Server

/**
 * TypeScript MCP Server
 * 使用 @modelcontextprotocol/sdk
 */

import { Server } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
  Tool,
} from "@modelcontextprotocol/sdk/types.js";
import { z } from "zod";
import type { Request, Response } from "express";

// ===== 工具Schema定义 =====

const FetchWebpageSchema = z.object({
  url: z.string().url(),
  extract_text: z.boolean().default(true),
});

const SearchCodeSchema = z.object({
  repo: z.string().regex(/^[\w-]+\/[\w-]+$/),
  query: z.string(),
  language: z.string().optional(),
});

const RunSQLSchema = z.object({
  sql: z.string(),
  limit: z.number().default(100),
});

const FileGlobSchema = z.object({
  pattern: z.string(),
  root: z.string().default("."),
});

// ===== 创建Server =====

const server = new Server(
  {
    name: "my-mcp-server",
    version: "1.0.0",
  },
  {
    capabilities: {
      tools: {},
    },
  }
);

// ===== 注册工具列表 =====

server.setRequestHandler(ListToolsRequestSchema, async () => {
  return {
    tools: [
      {
        name: "fetch_webpage",
        description: "获取网页内容，支持提取纯文本",
        inputSchema: {
          type: "object",
          properties: {
            url: {
              type: "string",
              description: "网页URL",
            },
            extract_text: {
              type: "boolean",
              description: "是否提取纯文本",
              default: true,
            },
          },
          required: ["url"],
        },
      } as Tool,
      {
        name: "search_code",
        description: "在GitHub仓库中搜索代码",
        inputSchema: {
          type: "object",
          properties: {
            repo: {
              type: "string",
              description: "仓库名，格式: owner/repo",
            },
            query: {
              type: "string",
              description: "搜索关键词",
            },
            language: {
              type: "string",
              description: "语言过滤",
            },
          },
          required: ["repo", "query"],
        },
      } as Tool,
      {
        name: "run_sql",
        description: "执行SQL查询（仅SELECT）",
        inputSchema: {
          type: "object",
          properties: {
            sql: {
              type: "string",
              description: "SQL查询语句",
            },
            limit: {
              type: "number",
              description: "最大返回行数",
              default: 100,
            },
          },
          required: ["sql"],
        },
      } as Tool,
    ],
  };
});

// ===== 工具执行处理 =====

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;

  try {
    let result: any;

    switch (name) {
      case "fetch_webpage":
        result = await handleFetchWebpage(FetchWebpageSchema.parse(args));
        break;
      case "search_code":
        result = await handleSearchCode(SearchCodeSchema.parse(args));
        break;
      case "run_sql":
        result = await handleRunSQL(RunSQLSchema.parse(args));
        break;
      default:
        throw new Error(`Unknown tool: ${name}`);
    }

    return {
      content: [
        {
          type: "text",
          text: JSON.stringify(result, null, 2),
        },
      ],
    };
  } catch (error) {
    return {
      content: [
        {
          type: "text",
          text: `Error: ${error instanceof Error ? error.message : String(error)}`,
        },
      ],
      isError: true,
    };
  }
});

// ===== 工具实现 =====

async function handleFetchWebpage(args: z.infer<typeof FetchWebpageSchema>) {
  const response = await fetch(args.url);
  const text = await response.text();

  if (args.extract_text) {
    // 简单文本提取
    const cleaned = text
      .replace(/<script[^>]*>.*?<\/script>/gs, "")
      .replace(/<style[^>]*>.*?<\/style>/gs, "")
      .replace(/<[^>]+>/g, " ")
      .replace(/\s+/g, " ")
      .trim();

    return { url: args.url, text: cleaned.slice(0, 5000) };
  }

  return { url: args.url, html: text.slice(0, 10000) };
}

async function handleSearchCode(args: z.infer<typeof SearchCodeSchema>) {
  const query = encodeURIComponent(args.query);
  const url = `https://api.github.com/search/code?q=${query}+repo:${args.repo}` +
    (args.language ? `+language:${args.language}` : "");

  const response = await fetch(url, {
    headers: { Accept: "application/vnd.github.v3+json" },
  });

  const data = await response.json();

  return {
    total: data.total_count,
    items: (data.items || []).slice(0, 10).map((item: any) => ({
      name: item.name,
      path: item.path,
      url: item.html_url,
    })),
  };
}

async function handleRunSQL(args: z.infer<typeof RunSQLSchema>) {
  // 实际使用 pg 或 knex
  const sql = args.sql.trim().toUpperCase();
  if (!sql.startsWith("SELECT")) {
    throw new Error("Only SELECT queries are allowed");
  }

  // 示例实现
  return {
    message: "SQL execution requires database configuration",
    sql: args.sql,
    limit: args.limit,
  };
}

// ===== 启动 =====

import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";

const transport = new StdioServerTransport();
await server.connect(transport);
console.error("MCP Server running on stdio");

3. 主流MCP工具集成

3.1 PostgreSQL MCP Server

# Claude Desktop配置 ~/.config/claude-desktop.json
# Windows: %APPDATA%\Claude\claude_desktop_config.json

{
  "mcpServers": {
    "postgresql": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres"],
      "env": {
        "PG_HOST": "localhost",
        "PG_PORT": "5432",
        "PG_USER": "postgres",
        "PG_PASSWORD": "${PG_PASSWORD}",
        "PG_DATABASE": "production"
      }
    }
  }
}

-- PostgreSQL MCP使用示例

-- 查看所有表
SELECT table_name FROM information_schema.tables 
WHERE table_schema = 'public';

-- 分析慢查询
EXPLAIN ANALYZE 
SELECT * FROM orders 
WHERE user_id = 123 
AND created_at > NOW() - INTERVAL '30 days';

-- 索引建议
SELECT 
    schemaname,
    tablename,
    seq_scan,
    idx_scan,
    pg_size_pretty(pg_relation_size(schemaname || '.' || tablename))
FROM pg_stat_user_tables
WHERE seq_scan > idx_scan * 10
ORDER BY seq_scan DESC;

3.2 Chrome DevTools MCP

# Chrome DevTools MCP配置
mcpServers:
  chrome-devtools:
    command: "npx"
    args: ["-y", "@anthropic-ai/mcp-chrome-devtools"]

// Chrome DevTools MCP工具

// 工具列表:
// 1. navigate - 导航到URL
// 2. screenshot - 截取页面截图
// 3. click - 点击元素
// 4. type - 输入文本
// 5. evaluate - 执行JavaScript
// 6. get_html - 获取页面HTML
// 7. find_elements - 查找DOM元素
// 8. get_cookies - 获取Cookie

// 使用示例: 自动登录测试
async function autoLogin(url, credentials) {
  // 1. 打开登录页
  await mcp_chrome_navigate({ url: `${url}/login` });
  
  // 2. 截图确认
  await mcp_chrome_screenshot();
  
  // 3. 输入账号
  await mcp_chrome_type({ 
    selector: 'input[name="email"]', 
    text: credentials.email 
  });
  
  // 4. 输入密码
  await mcp_chrome_type({
    selector: 'input[name="password"]',
    text: credentials.password
  });
  
  // 5. 点击登录
  await mcp_chrome_click({ selector: 'button[type="submit"]' });
  
  // 6. 等待跳转
  await new Promise(r => setTimeout(r, 2000));
  
  // 7. 获取结果页面
  const html = await mcp_chrome_get_html();
  
  return html.includes("Dashboard") ? "登录成功" : "登录失败";
}

3.3 文件系统MCP

# 文件系统MCP
mcpServers:
  filesystem:
    command: "npx"
    args: ["-y", "@modelcontextprotocol/server-filesystem"]
    env:
      # 限制访问目录
      ALLOWED_DIRECTORIES: "/path/to/project,/tmp"

# 文件系统MCP工具

# 工具列表:
# 1. read_file - 读取文件内容
# 2. write_file - 写入文件
# 3. list_directory - 列出目录
# 4. search_files - 搜索文件
# 5. get_file_info - 获取文件信息

# 安全配置:
ALLOWED_DIRECTORIES = ["/project/src", "/project/tests"]

def check_path(path: str) -> bool:
    """安全检查：防止路径遍历"""
    import os
    abs_path = os.path.abspath(path)
    
    for allowed in ALLOWED_DIRECTORIES:
        if abs_path.startswith(os.path.abspath(allowed)):
            return True
    return False

# 使用示例
async def refactor_code(file_path: str, changes: list):
    """代码重构"""
    if not check_path(file_path):
        raise PermissionError("Path not allowed")
    
    # 读取原文件
    content = await read_file(file_path)
    
    # 应用修改
    new_content = apply_changes(content, changes)
    
    # 写入
    await write_file(file_path, new_content)
    
    return {"status": "success", "file": file_path}

4. MCP生态工具矩阵

2026年主流MCP工具生态:
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

开发工具:
• Chrome DevTools MCP  ⭐41K  — 浏览器自动化
• PostgreSQL MCP        ⭐28K  — 数据库操作
• Filesystem MCP        ⭐15K  — 文件系统
• GitHub MCP            ⭐12K  — GitHub API
• Slack MCP             ⭐8K   — 团队协作
• Notion MCP            ⭐7K   — 知识库

浏览器自动化:
• browser-use           ⭐97K  — AI驱动浏览器
• chrome-devtools-mcp   ⭐41K  — Chrome原生控制
• playwright-mcp        ⭐20K  — Playwright集成

数据处理:
• sql-mcp               ⭐8K   — 多数据库支持
• s3-mcp                ⭐5K   — 对象存储
• redis-mcp             ⭐4K   — 缓存操作

搜索与检索:
• tavily-mcp            ⭐6K   — 网络搜索
• brave-search-mcp       ⭐4K   — 隐私搜索
• wikipedia-mcp         ⭐3K   — 百科检索

AI模型集成:
• openai-mcp            ⭐10K  — OpenAI API
• anthropic-mcp          ⭐8K   — Claude API
• ollama-mcp             ⭐6K   — 本地LLM
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

5. 生产环境安全配置

5.1 权限与沙箱

# 安全配置文件 mcp-security.yaml

security:
  # 工具权限控制
  tool_permissions:
    # 默认全部禁止
    default: deny
    
    # 按工具授权
    allow:
      - fetch_webpage: ["GET"]           # 只允许GET请求
      - run_sql: ["SELECT"]               # 只允许SELECT
      - file_read: ["*.md", "*.json"]     # 只读特定文件
      - file_write: ["**/temp/**"]        # 只写临时目录
    
    deny:
      - run_sql: ["DELETE", "UPDATE", "DROP", "TRUNCATE"]
      - file_delete: ["*"]                # 禁止删除文件
      - exec_command: ["*"]               # 禁止执行命令
  
  # 网络访问限制
  network:
    allowed_domains:
      - "*.github.com"
      - "*.api.github.com"
      - "localhost"
      - "127.0.0.1"
    
    blocked_domains:
      - "*.internal"
      - "10.0.0.0/8"
      - "192.168.0.0/16"
  
  # 速率限制
  rate_limit:
    default: 60        # 每分钟60次
    fetch_webpage: 30  # 每分钟30次
    run_sql: 120       # 每分钟120次
  
  # 审计日志
  audit:
    enabled: true
    log_file: "/var/log/mcp-audit.jsonl"
    log_level: "info"
    log_fields:
      - timestamp
      - tool_name
      - arguments
      - result
      - duration_ms
      - user_id

5.2 安全中间件

# 安全检查中间件

from functools import wraps
from typing import Callable, Any
import re
import yaml

class MCPSecurity:
    """MCP安全检查器"""
    
    def __init__(self, config_path: str):
        with open(config_path) as f:
            self.config = yaml.safe_load(f)
    
    def check_tool_permission(self, tool_name: str, args: dict) -> bool:
        """检查工具权限"""
        permissions = self.config["security"]["tool_permissions"]
        
        # 检查是否在黑名单
        if "deny" in permissions:
            for pattern in permissions["deny"].get(tool_name, []):
                if self._match_pattern(pattern, args):
                    return False
        
        # 检查是否在白名单
        if "allow" in permissions:
            allowed = permissions["allow"].get(tool_name, [])
            if not any(self._match_pattern(p, args) for p in allowed):
                return False
        
        return True
    
    def check_network_access(self, url: str) -> bool:
        """检查网络访问权限"""
        from urllib.parse import urlparse
        domain = urlparse(url).netloc
        
        blocked = self.config["security"]["network"]["blocked_domains"]
        for pattern in blocked:
            if self._match_domain(domain, pattern):
                return False
        
        allowed = self.config["security"]["network"]["allowed_domains"]
        return any(self._match_domain(domain, p) for p in allowed)
    
    def _match_pattern(self, pattern: str, args: dict) -> bool:
        """通用模式匹配"""
        for value in args.values():
            if isinstance(value, str) and self._fnmatch(value, pattern):
                return True
        return False
    
    def _match_domain(self, domain: str, pattern: str) -> bool:
        """域名匹配"""
        import fnmatch
        return fnmatch.fnmatch(domain, pattern)


def secure_tool(func: Callable) -> Callable:
    """工具安全装饰器"""
    @wraps(func)
    async def wrapper(*args, **kwargs):
        security = MCPSecurity("mcp-security.yaml")
        
        # 获取工具名
        tool_name = func.__name__.replace("handle_", "")
        
        # 检查权限
        if not security.check_tool_permission(tool_name, kwargs):
            raise PermissionError(f"Tool {tool_name} not permitted")
        
        return await func(*args, **kwargs)
    
    return wrapper


@secure_tool
async def handle_run_sql(sql: str, limit: int = 100) -> dict:
    """安全的SQL执行"""
    # ... 实现
    pass

5.3 审计日志

# 审计日志记录器

import json
import time
from datetime import datetime
from typing import Any, Optional
from contextlib import asynccontextmanager

class AuditLogger:
    """MCP审计日志"""
    
    def __init__(self, log_file: str):
        self.log_file = log_file
    
    def log(
        self,
        tool_name: str,
        arguments: dict,
        result: Any,
        duration_ms: float,
        user_id: str = "anonymous",
        error: Optional[str] = None
    ):
        """记录审计日志"""
        entry = {
            "timestamp": datetime.now().isoformat(),
            "tool_name": tool_name,
            "arguments": self._sanitize_args(arguments),
            "result_size": len(str(result)) if result else 0,
            "duration_ms": round(duration_ms, 2),
            "user_id": user_id,
            "error": error,
        }
        
        with open(self.log_file, "a") as f:
            f.write(json.dumps(entry, ensure_ascii=False) + "\n")
    
    def _sanitize_args(self, args: dict) -> dict:
        """敏感信息脱敏"""
        sensitive_keys = ["password", "token", "secret", "api_key"]
        sanitized = {}
        
        for key, value in args.items():
            if any(s in key.lower() for s in sensitive_keys):
                sanitized[key] = "***REDACTED***"
            else:
                sanitized[key] = value
        
        return sanitized


@asynccontextmanager
async def audit_context(tool_name: str, args: dict):
    """审计上下文管理器"""
    logger = AuditLogger("/var/log/mcp-audit.jsonl")
    start = time.perf_counter()
    error = None
    result = None
    
    try:
        yield
    except Exception as e:
        error = str(e)
        raise
    finally:
        duration = (time.perf_counter() - start) * 1000
        logger.log(tool_name, args, result, duration, error=error)

6. 总结

MCP开发要点

MCP协议开发核心:
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

1. Server开发
   • Python: mcp SDK + asyncio
   • TypeScript: @modelcontextprotocol/sdk
   • 核心: list_tools + call_tool

2. 工具设计
   • 单一职责: 每个工具做一件事
   • 幂等性: 同一输入总是相同输出
   • 错误处理: 清晰的错误消息
   • 文档: 完整的Schema和描述

3. 安全配置
   • 白名单: 最小权限原则
   • 输入验证: Zod/Pydantic Schema
   • 审计日志: 记录所有操作
   • 速率限制: 防止滥用

4. 生态选择
   • 浏览器: chrome-devtools-mcp
   • 数据库: postgresql-mcp
   • 文件: filesystem-mcp
   • 搜索: tavily-mcp
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━