MCP 模型上下文协议完全教程

第一章 MCP 协议概述：为什么需要标准化 AI 工具协议

1.1 AI 工具集成的现状与困境

在过去两年里，大语言模型（LLM）的能力突飞猛进。然而，一个关键问题始终困扰着开发者：如何让 AI 模型高效、安全地连接外部世界？

当前的 AI 工具集成面临三大困境：

碎片化问题：每个 AI 应用都在发明自己的工具调用方式。ChatGPT 有 Plugins、Functions Calling；Claude 有 Tool Use；各种开源框架各有各的接口规范。开发者为每个平台重复开发适配层，维护成本极高。

上下文孤岛：AI 模型被困在对话窗口中，无法直接访问用户的文件系统、数据库、API 等外部资源。用户不得不手动复制粘贴上下文信息，效率低下。

安全隐患：缺乏标准化的权限控制和沙箱机制。每个工具集成都是一个潜在的安全风险点，没有统一的安全模型可以依赖。

1.2 MCP 是什么

MCP（Model Context Protocol，模型上下文协议）是由 Anthropic 于 2024 年 11 月发布的开放标准协议。它的核心目标是：为 AI 模型与外部工具、数据源之间建立一个统一的连接标准。

可以把 MCP 理解为 AI 世界的"USB-C 接口"——不管你用的是什么 AI 模型（Claude、GPT、Gemini、Llama），也不管你要连接什么工具（文件系统、数据库、GitHub、浏览器），都通过同一个标准协议进行通信。

MCP 的核心设计理念：

• 标准化：统一的协议规范，一次开发，处处可用
• 安全性：内置权限控制、沙箱隔离、用户确认机制
• 可扩展性：模块化架构，支持自定义扩展
• 开放性：开源协议，任何厂商都可以实现

1.3 MCP 与传统方案的对比

维度	传统 Function Calling	MCP
标准化	各平台各不相同	统一开放标准
可复用性	绑定特定平台	跨平台通用
安全模型	各自实现	统一安全框架
发现机制	静态配置	动态能力发现
传输方式	HTTP 为主	支持多种传输层
生态系统	碎片化	统一生态

1.4 MCP 的应用场景

MCP 的应用场景非常广泛：

• IDE 集成：让 AI 编程助手访问项目文件、Git 仓库、终端
• 数据分析：让 AI 直接查询数据库、分析 CSV、生成图表
• 企业自动化：让 AI 操作内部系统、审批流程、知识库
• 个人助手：让 AI 管理日历、邮件、笔记、待办事项
• 开发运维：让 AI 执行部署、监控、日志分析

---

第二章 MCP 架构详解

2.1 整体架构

MCP 采用经典的 Client-Server 架构：

┌─────────────────────────────────────────────────┐
│                 MCP Host                         │
│  (Claude Desktop / Cursor / VSCode / 自定义应用) │
│                                                  │
│  ┌─────────────┐  ┌─────────────┐               │
│  │ MCP Client  │  │ MCP Client  │    ...        │
│  └──────┬──────┘  └──────┬──────┘               │
└─────────┼────────────────┼───────────────────────┘
          │                │
    ┌─────▼─────┐    ┌─────▼─────┐
    │ MCP Server│    │ MCP Server│         ...
    │ (文件系统) │    │ (数据库)  │
    └───────────┘    └───────────┘

MCP Host：宿主应用程序，如 Claude Desktop、Cursor、VS Code 等。它负责管理 MCP Client 的生命周期。

MCP Client：协议客户端，由 Host 创建和管理。每个 Client 维持与一个 MCP Server 的一对一连接。

MCP Server：协议服务端，提供具体的工具、资源和提示词能力。一个 Server 可以同时被多个 Client 连接。

2.2 传输层

MCP 支持两种标准传输方式：

2.2.1 stdio 传输

基于标准输入/输出的传输方式，适用于本地进程间通信：

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/list",
  "params": {}
}

Client 通过启动 Server 进程，向其 stdin 写入 JSON-RPC 消息，从 stdout 读取响应。stderr 用于日志输出。

优点：简单、安全、无需网络 缺点：仅限本地、单 Client 连接

2.2.2 HTTP + SSE 传输

基于 HTTP 的远程传输方式，适用于网络部署：

Client ──POST──→ Server (消息发送)
Client ←─SSE──── Server (消息推送)

Client 通过 HTTP POST 发送请求，Server 通过 Server-Sent Events (SSE) 推送响应和通知。

优点：支持远程访问、多 Client 连接 缺点：需要网络、需要处理连接管理

2.3 消息格式

MCP 使用 JSON-RPC 2.0 作为消息格式：

请求（Request）

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "read_file",
    "arguments": {
      "path": "/home/user/document.txt"
    }
  }
}

响应（Response）

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "文件内容..."
      }
    ]
  }
}

错误响应（Error）

{
  "jsonrpc": "2.0",
  "id": 1,
  "error": {
    "code": -32602,
    "message": "Invalid params",
    "data": {
      "details": "Path does not exist"
    }
  }
}

通知（Notification）

{
  "jsonrpc": "2.0",
  "method": "notifications/resources/updated",
  "params": {
    "uri": "file:///home/user/document.txt"
  }
}

2.4 生命周期

MCP 连接的生命周期分为三个阶段：

1. 初始化阶段

// Client 发送初始化请求
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {
    "protocolVersion": "2024-11-05",
    "capabilities": {
      "roots": {
        "listChanged": true
      },
      "sampling": {}
    },
    "clientInfo": {
      "name": "MyClient",
      "version": "1.0.0"
    }
  }
}

// Server 响应
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "protocolVersion": "2024-11-05",
    "capabilities": {
      "tools": {
        "listChanged": true
      },
      "resources": {
        "subscribe": true,
        "listChanged": true
      },
      "prompts": {
        "listChanged": true
      },
      "logging": {}
    },
    "serverInfo": {
      "name": "FileSystemServer",
      "version": "1.0.0"
    }
  }
}

// Client 发送初始化完成通知
{
  "jsonrpc": "2.0",
  "method": "notifications/initialized"
}

2. 操作阶段

在此阶段，Client 和 Server 之间进行正常的请求-响应交互，包括工具调用、资源访问、提示词获取等。

3. 关闭阶段

任一方可以发送关闭请求，另一方响应后连接终止。

2.5 能力协商

在初始化阶段，Client 和 Server 通过能力协商确定双方支持的功能：

Server 能力：

{
  "capabilities": {
    "tools": {},           // 支持工具调用
    "resources": {
      "subscribe": true,   // 支持资源订阅
      "listChanged": true  // 支持资源列表变更通知
    },
    "prompts": {},         // 支持提示词
    "logging": {},         // 支持日志
    "completions": {}      // 支持自动补全
  }
}

Client 能力：

{
  "capabilities": {
    "roots": {
      "listChanged": true  // 支持根目录变更通知
    },
    "sampling": {}         // 支持采样请求
  }
}

---

第三章 MCP 核心能力

3.1 Resources（资源）

Resources 是 MCP 中最基础的能力，用于向 AI 模型暴露数据和内容。

3.1.1 资源列表

// 请求
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "resources/list",
  "params": {}
}

// 响应
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "resources": [
      {
        "uri": "file:///home/user/projects",
        "name": "项目目录",
        "description": "用户项目文件夹",
        "mimeType": "text/directory"
      },
      {
        "uri": "postgres://localhost/mydb",
        "name": "数据库",
        "description": "应用数据库",
        "mimeType": "application/sql"
      }
    ]
  }
}

3.1.2 资源读取

// 请求
{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "resources/read",
  "params": {
    "uri": "file:///home/user/projects/README.md"
  }
}

// 响应
{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "contents": [
      {
        "uri": "file:///home/user/projects/README.md",
        "mimeType": "text/markdown",
        "text": "# My Project\n\nThis is a sample project..."
      }
    ]
  }
}

3.1.3 资源模板

对于动态资源，使用 URI 模板：

{
  "resourceTemplates": [
    {
      "uriTemplate": "file:///{path}",
      "name": "文件",
      "description": "读取指定路径的文件",
      "mimeType": "text/plain"
    },
    {
      "uriTemplate": "postgres://{host}/{database}/tables/{table}",
      "name": "数据库表",
      "description": "查询数据库表数据",
      "mimeType": "application/json"
    }
  ]
}

3.1.4 资源订阅

Client 可以订阅资源变更通知：

// 订阅
{
  "jsonrpc": "2.0",
  "id": 3,
  "method": "resources/subscribe",
  "params": {
    "uri": "file:///home/user/projects"
  }
}

// Server 推送变更通知
{
  "jsonrpc": "2.0",
  "method": "notifications/resources/updated",
  "params": {
    "uri": "file:///home/user/projects/src/main.py"
  }
}

3.2 Tools（工具）

Tools 是 MCP 最强大的能力，允许 AI 模型执行具体操作。

3.2.1 工具列表

// 请求
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/list",
  "params": {}
}

// 响应
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "tools": [
      {
        "name": "read_file",
        "description": "读取文件内容",
        "inputSchema": {
          "type": "object",
          "properties": {
            "path": {
              "type": "string",
              "description": "文件路径"
            },
            "encoding": {
              "type": "string",
              "description": "文件编码",
              "default": "utf-8"
            }
          },
          "required": ["path"]
        }
      },
      {
        "name": "write_file",
        "description": "写入文件内容",
        "inputSchema": {
          "type": "object",
          "properties": {
            "path": {
              "type": "string",
              "description": "文件路径"
            },
            "content": {
              "type": "string",
              "description": "文件内容"
            }
          },
          "required": ["path", "content"]
        }
      },
      {
        "name": "execute_sql",
        "description": "执行 SQL 查询",
        "inputSchema": {
          "type": "object",
          "properties": {
            "query": {
              "type": "string",
              "description": "SQL 查询语句"
            },
            "database": {
              "type": "string",
              "description": "数据库名称",
              "default": "main"
            }
          },
          "required": ["query"]
        }
      }
    ]
  }
}

3.2.2 工具调用

// 请求
{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "tools/call",
  "params": {
    "name": "read_file",
    "arguments": {
      "path": "/home/user/document.txt"
    }
  }
}

// 响应（成功）
{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "文件内容..."
      }
    ],
    "isError": false
  }
}

// 响应（失败）
{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "错误：文件不存在"
      }
    ],
    "isError": true
  }
}

3.2.3 工具内容类型

工具可以返回多种内容类型：

{
  "content": [
    {
      "type": "text",
      "text": "文本内容"
    },
    {
      "type": "image",
      "data": "base64编码的图片数据",
      "mimeType": "image/png"
    },
    {
      "type": "resource",
      "resource": {
        "uri": "file:///path/to/resource",
        "text": "资源内容",
        "mimeType": "text/plain"
      }
    }
  ]
}

3.3 Prompts（提示词）

Prompts 允许 Server 提供预定义的提示词模板。

3.3.1 提示词列表

// 请求
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "prompts/list",
  "params": {}
}

// 响应
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "prompts": [
      {
        "name": "code_review",
        "description": "代码审查提示词",
        "arguments": [
          {
            "name": "language",
            "description": "编程语言",
            "required": true
          },
          {
            "name": "code",
            "description": "待审查的代码",
            "required": true
          }
        ]
      },
      {
        "name": "explain_code",
        "description": "代码解释提示词",
        "arguments": [
          {
            "name": "code",
            "description": "待解释的代码",
            "required": true
          }
        ]
      }
    ]
  }
}

3.3.2 获取提示词

// 请求
{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "prompts/get",
  "params": {
    "name": "code_review",
    "arguments": {
      "language": "python",
      "code": "def hello():\n    print('hello')"
    }
  }
}

// 响应
{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "description": "Python 代码审查",
    "messages": [
      {
        "role": "user",
        "content": {
          "type": "text",
          "text": "请审查以下 Python 代码：\n\n```python\ndef hello():\n    print('hello')\n```\n\n请从代码质量、性能、安全性、可维护性等方面给出建议。"
        }
      }
    ]
  }
}

3.4 Sampling（采样）

Sampling 允许 Server 请求 Client 进行 LLM 推理，实现"人在回路"的交互模式：

// Server 发送采样请求
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "sampling/createMessage",
  "params": {
    "messages": [
      {
        "role": "user",
        "content": {
          "type": "text",
          "text": "请分析这段代码的功能"
        }
      }
    ],
    "modelPreferences": {
      "hints": [
        { "name": "claude-3-5-sonnet" }
      ],
      "costPriority": 0.5,
      "speedPriority": 0.8,
      "intelligencePriority": 0.9
    },
    "systemPrompt": "你是一个代码分析专家",
    "maxTokens": 1000
  }
}

// Client 响应（可能经过用户确认）
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "model": "claude-3-5-sonnet-20241022",
    "role": "assistant",
    "content": {
      "type": "text",
      "text": "这段代码实现了一个简单的问候函数..."
    },
    "stopReason": "endTurn"
  }
}

3.5 Logging（日志）

Server 可以向 Client 发送日志消息：

{
  "jsonrpc": "2.0",
  "method": "notifications/message",
  "params": {
    "level": "info",
    "logger": "filesystem",
    "data": {
      "message": "文件读取成功",
      "path": "/home/user/document.txt",
      "size": 1024
    }
  }
}

日志级别：debug、info、warning、error、critical

---

第四章 MCP Server 开发

4.1 Python SDK 开发

4.1.1 环境准备

# 创建虚拟环境
python -m venv mcp-env
source mcp-env/bin/activate  # Linux/macOS
# mcp-env\Scripts\activate   # Windows

# 安装 MCP Python SDK
pip install mcp

# 或使用 uv（推荐）
pip install uv
uv pip install mcp

4.1.2 基础 Server 示例

# server.py - 最简单的 MCP Server
import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
    Tool,
    TextContent,
    CallToolResult,
    ListToolsResult,
)

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

# 定义工具列表
@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="greet",
            description="向指定的人打招呼",
            inputSchema={
                "type": "object",
                "properties": {
                    "name": {
                        "type": "string",
                        "description": "要打招呼的人的名字"
                    }
                },
                "required": ["name"]
            }
        ),
        Tool(
            name="calculate",
            description="执行简单的数学计算",
            inputSchema={
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "数学表达式，如 '2 + 3 * 4'"
                    }
                },
                "required": ["expression"]
            }
        )
    ]

# 实现工具调用
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "greet":
        greeting = f"你好，{arguments['name']}！欢迎使用 MCP！"
        return [TextContent(type="text", text=greeting)]
    
    elif name == "calculate":
        try:
            # 安全地计算表达式
            result = eval(arguments["expression"], {"__builtins__": {}}, {})
            return [TextContent(type="text", text=f"计算结果：{result}")]
        except Exception as e:
            return [TextContent(type="text", text=f"计算错误：{str(e)}")]
    
    else:
        raise ValueError(f"未知工具：{name}")

# 启动 Server
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options()
        )

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

4.1.3 文件系统 Server

# filesystem_server.py - 文件系统 MCP Server
import os
import json
import asyncio
from pathlib import Path
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
    Tool,
    Resource,
    TextContent,
    CallToolResult,
    ListToolsResult,
    ListResourcesResult,
    ReadResourceResult,
)

server = Server("filesystem-server")

# 配置允许访问的根目录
ALLOWED_ROOTS = [Path.home() / "documents", Path.home() / "projects"]

def is_path_allowed(path: Path) -> bool:
    """检查路径是否在允许的范围内"""
    try:
        resolved = path.resolve()
        return any(
            str(resolved).startswith(str(root.resolve()))
            for root in ALLOWED_ROOTS
        )
    except:
        return False

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="read_file",
            description="读取文件内容",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "文件路径"}
                },
                "required": ["path"]
            }
        ),
        Tool(
            name="write_file",
            description="写入文件内容",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "文件路径"},
                    "content": {"type": "string", "description": "文件内容"}
                },
                "required": ["path", "content"]
            }
        ),
        Tool(
            name="list_directory",
            description="列出目录内容",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "目录路径"},
                    "recursive": {"type": "boolean", "description": "是否递归列出", "default": False}
                },
                "required": ["path"]
            }
        ),
        Tool(
            name="search_files",
            description="搜索文件",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "搜索目录"},
                    "pattern": {"type": "string", "description": "文件名模式（支持通配符）"}
                },
                "required": ["path", "pattern"]
            }
        ),
        Tool(
            name="get_file_info",
            description="获取文件信息",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "文件路径"}
                },
                "required": ["path"]
            }
        )
    ]

@server.list_resources()
async def list_resources() -> list[Resource]:
    resources = []
    for root in ALLOWED_ROOTS:
        if root.exists():
            resources.append(
                Resource(
                    uri=f"file://{root}",
                    name=f"目录：{root.name}",
                    description=f"用户目录 {root}",
                    mimeType="text/directory"
                )
            )
    return resources

@server.read_resource()
async def read_resource(uri: str) -> str:
    path = Path(uri.replace("file://", ""))
    if not is_path_allowed(path):
        raise PermissionError(f"无权访问：{path}")
    
    if path.is_file():
        return path.read_text(encoding="utf-8")
    elif path.is_dir():
        items = []
        for item in sorted(path.iterdir()):
            prefix = "📁" if item.is_dir() else "📄"
            items.append(f"{prefix} {item.name}")
        return "\n".join(items)
    else:
        raise FileNotFoundError(f"路径不存在：{path}")

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    try:
        if name == "read_file":
            path = Path(arguments["path"])
            if not is_path_allowed(path):
                return [TextContent(type="text", text=f"错误：无权访问 {path}")]
            content = path.read_text(encoding="utf-8")
            return [TextContent(type="text", text=content)]
        
        elif name == "write_file":
            path = Path(arguments["path"])
            if not is_path_allowed(path):
                return [TextContent(type="text", text=f"错误：无权写入 {path}")]
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_text(arguments["content"], encoding="utf-8")
            return [TextContent(type="text", text=f"文件已写入：{path}")]
        
        elif name == "list_directory":
            path = Path(arguments["path"])
            if not is_path_allowed(path):
                return [TextContent(type="text", text=f"错误：无权访问 {path}")]
            
            recursive = arguments.get("recursive", False)
            items = []
            
            if recursive:
                for item in path.rglob("*"):
                    relative = item.relative_to(path)
                    prefix = "📁" if item.is_dir() else "📄"
                    items.append(f"{prefix} {relative}")
            else:
                for item in sorted(path.iterdir()):
                    prefix = "📁" if item.is_dir() else "📄"
                    items.append(f"{prefix} {item.name}")
            
            return [TextContent(type="text", text="\n".join(items) or "空目录")]
        
        elif name == "search_files":
            path = Path(arguments["path"])
            if not is_path_allowed(path):
                return [TextContent(type="text", text=f"错误：无权访问 {path}")]
            
            pattern = arguments["pattern"]
            matches = list(path.rglob(pattern))
            
            if matches:
                result = "\n".join(str(m) for m in matches[:50])
                return [TextContent(type="text", text=f"找到 {len(matches)} 个文件：\n{result}")]
            else:
                return [TextContent(type="text", text="未找到匹配的文件")]
        
        elif name == "get_file_info":
            path = Path(arguments["path"])
            if not is_path_allowed(path):
                return [TextContent(type="text", text=f"错误：无权访问 {path}")]
            
            stat = path.stat()
            info = {
                "name": path.name,
                "type": "directory" if path.is_dir() else "file",
                "size": stat.st_size,
                "created": stat.st_ctime,
                "modified": stat.st_mtime,
                "permissions": oct(stat.st_mode)[-3:]
            }
            return [TextContent(type="text", text=json.dumps(info, indent=2, ensure_ascii=False))]
        
        else:
            return [TextContent(type="text", text=f"未知工具：{name}")]
    
    except Exception as e:
        return [TextContent(type="text", text=f"错误：{str(e)}")]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options()
        )

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

4.1.4 数据库 Server

# database_server.py - 数据库 MCP Server
import sqlite3
import json
import asyncio
from pathlib import Path
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

server = Server("database-server")
DB_PATH = Path("data.db")

def get_db():
    conn = sqlite3.connect(str(DB_PATH))
    conn.row_factory = sqlite3.Row
    return conn

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="execute_query",
            description="执行 SQL 查询（SELECT）",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "SQL 查询语句"},
                    "params": {"type": "array", "description": "查询参数", "items": {}}
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="execute_command",
            description="执行 SQL 命令（INSERT/UPDATE/DELETE）",
            inputSchema={
                "type": "object",
                "properties": {
                    "command": {"type": "string", "description": "SQL 命令"},
                    "params": {"type": "array", "description": "命令参数", "items": {}}
                },
                "required": ["command"]
            }
        ),
        Tool(
            name="list_tables",
            description="列出所有表",
            inputSchema={"type": "object", "properties": {}}
        ),
        Tool(
            name="describe_table",
            description="查看表结构",
            inputSchema={
                "type": "object",
                "properties": {
                    "table": {"type": "string", "description": "表名"}
                },
                "required": ["table"]
            }
        ),
        Tool(
            name="create_table",
            description="创建表",
            inputSchema={
                "type": "object",
                "properties": {
                    "table": {"type": "string", "description": "表名"},
                    "schema": {"type": "string", "description": "表结构定义"}
                },
                "required": ["table", "schema"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    try:
        if name == "execute_query":
            conn = get_db()
            cursor = conn.execute(arguments["query"], arguments.get("params", []))
            rows = cursor.fetchall()
            result = [dict(row) for row in rows]
            conn.close()
            return [TextContent(
                type="text",
                text=json.dumps(result, indent=2, ensure_ascii=False, default=str)
            )]
        
        elif name == "execute_command":
            conn = get_db()
            cursor = conn.execute(arguments["command"], arguments.get("params", []))
            conn.commit()
            affected = cursor.rowcount
            conn.close()
            return [TextContent(type="text", text=f"命令执行成功，影响 {affected} 行")]
        
        elif name == "list_tables":
            conn = get_db()
            cursor = conn.execute(
                "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
            )
            tables = [row["name"] for row in cursor.fetchall()]
            conn.close()
            return [TextContent(type="text", text="\n".join(tables) or "数据库中没有表")]
        
        elif name == "describe_table":
            table = arguments["table"]
            conn = get_db()
            cursor = conn.execute(f"PRAGMA table_info({table})")
            columns = [dict(row) for row in cursor.fetchall()]
            conn.close()
            return [TextContent(
                type="text",
                text=json.dumps(columns, indent=2, ensure_ascii=False)
            )]
        
        elif name == "create_table":
            table = arguments["table"]
            schema = arguments["schema"]
            conn = get_db()
            conn.execute(f"CREATE TABLE IF NOT EXISTS {table} ({schema})")
            conn.commit()
            conn.close()
            return [TextContent(type="text", text=f"表 {table} 创建成功")]
        
        else:
            return [TextContent(type="text", text=f"未知工具：{name}")]
    
    except Exception as e:
        return [TextContent(type="text", text=f"错误：{str(e)}")]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options()
        )

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

4.2 Node.js SDK 开发

4.2.1 环境准备

# 创建项目
mkdir my-mcp-server && cd my-mcp-server
npm init -y

# 安装依赖
npm install @modelcontextprotocol/sdk zod

# TypeScript 支持（可选）
npm install -D typescript @types/node tsx

4.2.2 基础 Server 示例

// src/server.ts - Node.js MCP Server
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";
import { z } from "zod";

// 创建 Server 实例
const server = new Server(
  { name: "my-node-server", version: "1.0.0" },
  { capabilities: { tools: {} } }
);

// 定义工具输入模式
const GreetSchema = z.object({
  name: z.string().describe("要打招呼的人的名字"),
});

const CalculateSchema = z.object({
  expression: z.string().describe("数学表达式"),
});

// 列出工具
server.setRequestHandler(ListToolsRequestSchema, async () => {
  return {
    tools: [
      {
        name: "greet",
        description: "向指定的人打招呼",
        inputSchema: {
          type: "object",
          properties: {
            name: { type: "string", description: "要打招呼的人的名字" }
          },
          required: ["name"]
        }
      },
      {
        name: "calculate",
        description: "执行简单的数学计算",
        inputSchema: {
          type: "object",
          properties: {
            expression: { type: "string", description: "数学表达式" }
          },
          required: ["expression"]
        }
      },
      {
        name: "get_time",
        description: "获取当前时间",
        inputSchema: { type: "object", properties: {} }
      }
    ]
  };
});

// 处理工具调用
server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;
  
  switch (name) {
    case "greet": {
      const { name: personName } = GreetSchema.parse(args);
      return {
        content: [
          {
            type: "text",
            text: `你好，${personName}！欢迎使用 MCP Server！`
          }
        ]
      };
    }
    
    case "calculate": {
      const { expression } = CalculateSchema.parse(args);
      try {
        // 简单的数学计算（生产环境应使用安全的表达式解析器）
        const result = Function(`"use strict"; return (${expression})`)();
        return {
          content: [
            {
              type: "text",
              text: `${expression} = ${result}`
            }
          ]
        };
      } catch (error) {
        return {
          content: [
            {
              type: "text",
              text: `计算错误：${(error as Error).message}`
            }
          ],
          isError: true
        };
      }
    }
    
    case "get_time": {
      return {
        content: [
          {
            type: "text",
            text: `当前时间：${new Date().toLocaleString("zh-CN")}`
          }
        ]
      };
    }
    
    default:
      throw new Error(`未知工具：${name}`);
  }
});

// 启动 Server
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("MCP Server 已启动");
}

main().catch(console.error);

// tsconfig.json
{
  "compilerOptions": {
    "target": "ES2022",
    "module": "ES2022",
    "moduleResolution": "node16",
    "outDir": "./dist",
    "rootDir": "./src",
    "strict": true,
    "esModuleInterop": true,
    "skipLibCheck": true,
    "forceConsistentCasingInFileNames": true,
    "resolveJsonModule": true,
    "declaration": true,
    "declarationMap": true,
    "sourceMap": true
  },
  "include": ["src/**/*"],
  "exclude": ["node_modules", "dist"]
}

// package.json (需要添加)
{
  "type": "module",
  "scripts": {
    "build": "tsc",
    "start": "node dist/server.js",
    "dev": "tsx src/server.ts"
  }
}

4.2.3 Web 搜索 Server

// src/web-search-server.ts
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import {
  CallToolRequestSchema,
  ListToolsRequestSchema,
} from "@modelcontextprotocol/sdk/types.js";
import { z } from "zod";

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

// 搜索结果类型
interface SearchResult {
  title: string;
  url: string;
  snippet: string;
}

// 使用 DuckDuckGo 搜索（无需 API Key）
async function searchDuckDuckGo(query: string, limit: number = 5): Promise<SearchResult[]> {
  const url = `https://api.duckduckgo.com/?q=${encodeURIComponent(query)}&format=json&no_html=1`;
  
  const response = await fetch(url);
  const data = await response.json();
  
  const results: SearchResult[] = [];
  
  // 解析结果
  if (data.Abstract) {
    results.push({
      title: data.Heading || query,
      url: data.AbstractURL || "",
      snippet: data.Abstract
    });
  }
  
  if (data.RelatedTopics) {
    for (const topic of data.RelatedTopics.slice(0, limit - results.length)) {
      if (topic.Text && topic.FirstURL) {
        results.push({
          title: topic.Text.split(" - ")[0] || topic.Text.substring(0, 50),
          url: topic.FirstURL,
          snippet: topic.Text
        });
      }
    }
  }
  
  return results.slice(0, limit);
}

const SearchSchema = z.object({
  query: z.string().describe("搜索关键词"),
  limit: z.number().optional().default(5).describe("返回结果数量")
});

server.setRequestHandler(ListToolsRequestSchema, async () => {
  return {
    tools: [
      {
        name: "web_search",
        description: "搜索网页",
        inputSchema: {
          type: "object",
          properties: {
            query: { type: "string", description: "搜索关键词" },
            limit: { type: "number", description: "返回结果数量", default: 5 }
          },
          required: ["query"]
        }
      }
    ]
  };
});

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;
  
  if (name === "web_search") {
    const { query, limit } = SearchSchema.parse(args);
    const results = await searchDuckDuckGo(query, limit);
    
    const formatted = results.map((r, i) => 
      `${i + 1}. ${r.title}\n   ${r.url}\n   ${r.snippet}`
    ).join("\n\n");
    
    return {
      content: [
        {
          type: "text",
          text: results.length > 0 
            ? `搜索 "${query}" 的结果：\n\n${formatted}`
            : `未找到 "${query}" 的相关结果`
        }
      ]
    };
  }
  
  throw new Error(`未知工具：${name}`);
});

async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("Web Search MCP Server 已启动");
}

main().catch(console.error);

---

第五章 MCP Client 开发

5.1 Python Client

# client.py - MCP Client 示例
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    # 配置 Server 参数
    server_params = StdioServerParameters(
        command="python",
        args=["server.py"],
        env=None
    )
    
    # 连接到 Server
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 初始化
            await session.initialize()
            
            # 列出可用工具
            tools = await session.list_tools()
            print("可用工具：")
            for tool in tools.tools:
                print(f"  - {tool.name}: {tool.description}")
            
            # 调用工具
            result = await session.call_tool(
                "greet",
                arguments={"name": "MCP 用户"}
            )
            print(f"\n调用结果：{result.content[0].text}")
            
            # 列出资源
            resources = await session.list_resources()
            print(f"\n可用资源：{len(resources.resources)} 个")
            for resource in resources.resources:
                print(f"  - {resource.name}: {resource.uri}")
            
            # 读取资源
            if resources.resources:
                content = await session.read_resource(resources.resources[0].uri)
                print(f"\n资源内容：{content.contents[0].text[:200]}...")
            
            # 列出提示词
            prompts = await session.list_prompts()
            print(f"\n可用提示词：{len(prompts.prompts)} 个")
            for prompt in prompts.prompts:
                print(f"  - {prompt.name}: {prompt.description}")

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

5.2 Node.js Client

// src/client.ts
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";

async function main() {
  // 创建传输层
  const transport = new StdioClientTransport({
    command: "node",
    args: ["dist/server.js"]
  });
  
  // 创建 Client
  const client = new Client(
    { name: "my-client", version: "1.0.0" },
    { capabilities: {} }
  );
  
  // 连接到 Server
  await client.connect(transport);
  console.log("已连接到 MCP Server");
  
  // 列出工具
  const tools = await client.listTools();
  console.log("\n可用工具：");
  for (const tool of tools.tools) {
    console.log(`  - ${tool.name}: ${tool.description}`);
  }
  
  // 调用工具
  const greetResult = await client.callTool({
    name: "greet",
    arguments: { name: "MCP 用户" }
  });
  console.log(`\n调用结果：${greetResult.content[0].text}`);
  
  // 获取时间
  const timeResult = await client.callTool({
    name: "get_time",
    arguments: {}
  });
  console.log(`当前时间：${timeResult.content[0].text}`);
  
  // 计算
  const calcResult = await client.callTool({
    name: "calculate",
    arguments: { expression: "2 + 3 * 4" }
  });
  console.log(`计算结果：${calcResult.content[0].text}`);
  
  // 断开连接
  await client.close();
}

main().catch(console.error);

5.3 HTTP + SSE Client

# http_client.py - HTTP 传输的 Client
import asyncio
from mcp import ClientSession
from mcp.client.sse import sse_client

async def main():
    # 连接到远程 Server
    async with sse_client("http://localhost:8080/sse") as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # 正常使用
            tools = await session.list_tools()
            print("可用工具：")
            for tool in tools.tools:
                print(f"  - {tool.name}")

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

---

第六章 主流 MCP Server 实战

6.1 文件系统 Server

官方提供的文件系统 Server 是最常用的 MCP Server 之一：

# 安装
npx -y @modelcontextprotocol/server-filesystem /path/to/allowed/dir

# 配置到 Claude Desktop
# 编辑 ~/Library/Application Support/Claude/claude_desktop_config.json (macOS)
# 或 %APPDATA%\Claude\claude_desktop_config.json (Windows)

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-filesystem",
        "/Users/username/Documents",
        "/Users/username/Projects"
      ]
    }
  }
}

6.2 数据库 Server

PostgreSQL Server

{
  "mcpServers": {
    "postgres": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-postgres",
        "postgresql://user:password@localhost:5432/mydb"
      ]
    }
  }
}

SQLite Server

{
  "mcpServers": {
    "sqlite": {
      "command": "uvx",
      "args": [
        "mcp-server-sqlite",
        "--db-path",
        "/path/to/database.db"
      ]
    }
  }
}

6.3 GitHub Server

{
  "mcpServers": {
    "github": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-github"
      ],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxxxxxxxxxxx"
      }
    }
  }
}

GitHub Server 提供的能力：

• 搜索仓库、Issues、PR
• 创建/更新 Issue 和 PR
• 读取文件内容
• 管理分支
• 查看代码差异

6.4 浏览器 Server

{
  "mcpServers": {
    "browser": {
      "command": "npx",
      "args": [
        "-y",
        "@anthropic/puppeteer-mcp-server"
      ]
    }
  }
}

浏览器 Server 提供的能力：

• 打开网页
• 点击元素
• 填写表单
• 截图
• 获取页面内容

6.5 Puppeteer Server

{
  "mcpServers": {
    "puppeteer": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-puppeteer"
      ]
    }
  }
}

6.6 Google Maps Server

{
  "mcpServers": {
    "google-maps": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-google-maps"
      ],
      "env": {
        "GOOGLE_MAPS_API_KEY": "your-api-key"
      }
    }
  }
}

6.7 Brave Search Server

{
  "mcpServers": {
    "brave-search": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-brave-search"
      ],
      "env": {
        "BRAVE_API_KEY": "your-api-key"
      }
    }
  }
}

---

第七章 与 Claude Desktop 集成配置

7.1 配置文件位置

Claude Desktop 的 MCP 配置文件位置：

• macOS：~/Library/Application Support/Claude/claude_desktop_config.json
• Windows：%APPDATA%\Claude\claude_desktop_config.json

7.2 基础配置

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-filesystem",
        "/Users/username/Documents"
      ]
    }
  }
}

7.3 多 Server 配置

{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-filesystem",
        "/Users/username/Documents",
        "/Users/username/Projects"
      ]
    },
    "github": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-github"
      ],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxxxxxxxxxxx"
      }
    },
    "postgres": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-postgres",
        "postgresql://user:password@localhost:5432/mydb"
      ]
    },
    "brave-search": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-brave-search"
      ],
      "env": {
        "BRAVE_API_KEY": "your-api-key"
      }
    }
  }
}

7.4 自定义 Server 配置

{
  "mcpServers": {
    "my-custom-server": {
      "command": "python",
      "args": ["/path/to/my_server.py"],
      "env": {
        "DATABASE_URL": "sqlite:///data.db",
        "API_KEY": "your-api-key"
      }
    },
    "my-node-server": {
      "command": "node",
      "args": ["/path/to/dist/server.js"],
      "env": {
        "NODE_ENV": "production"
      }
    }
  }
}

7.5 使用 uv 管理 Python Server

{
  "mcpServers": {
    "my-python-server": {
      "command": "uv",
      "args": [
        "run",
        "--with", "mcp",
        "python",
        "/path/to/server.py"
      ]
    }
  }
}

7.6 验证配置

1. 重启 Claude Desktop
2. 在对话框中查看是否有工具图标（🔧）
3. 尝试调用工具："帮我列出 Documents 目录下的文件"
4. 查看 Claude Desktop 的日志文件排查问题：
   • macOS：~/Library/Logs/Claude/
   • Windows：%APPDATA%\Claude\logs\

---

第八章 与 Cursor/VSCode 集成配置

8.1 Cursor 集成

Cursor 编辑器原生支持 MCP：

// .cursor/mcp.json（项目级别）
{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-filesystem",
        "."
      ]
    },
    "github": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-github"
      ],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxxxxxxxxxxx"
      }
    }
  }
}

或者在 Cursor 设置中全局配置：

1. 打开 Cursor Settings
2. 找到 MCP Servers 配置
3. 添加 Server 配置

8.2 VSCode 集成

VSCode 通过 Continue 扩展支持 MCP：

// .continue/config.json
{
  "models": [
    {
      "title": "Claude",
      "provider": "anthropic",
      "model": "claude-3-5-sonnet-20241022",
      "apiKey": "your-api-key"
    }
  ],
  "mcpServers": [
    {
      "name": "filesystem",
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-filesystem",
        "."
      ]
    }
  ]
}

8.3 Windsurf 集成

// ~/.codeium/windsurf/mcp_config.json
{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-filesystem",
        "/path/to/workspace"
      ]
    }
  }
}

8.4 Cline 集成

Cline 是 VS Code 中另一个支持 MCP 的 AI 编程助手：

// 在 Cline 设置中添加 MCP Server
{
  "mcpServers": {
    "my-server": {
      "command": "python",
      "args": ["/path/to/server.py"]
    }
  }
}

---

第九章 自定义 MCP Server 开发最佳实践

9.1 设计原则

单一职责：每个 Server 专注于一个领域，如文件操作、数据库访问、API 调用等。

清晰的工具命名：使用动词_名词的命名方式，如 read_file、create_issue、search_web。

详细的描述：为每个工具提供清晰的描述，帮助 AI 模型理解何时使用该工具。

合理的输入验证：使用 JSON Schema 或 Zod 验证输入参数。

错误处理：返回有意义的错误信息，而不是让 Server 崩溃。

9.2 安全最佳实践

# security.py - 安全最佳实践示例
import os
from pathlib import Path
from typing import Optional

class SecurityManager:
    def __init__(self):
        self.allowed_roots = self._load_allowed_roots()
        self.blocked_commands = [
            "rm -rf", "sudo", "chmod", "chown",
            "wget", "curl", "nc", "ncat"
        ]
    
    def _load_allowed_roots(self) -> list[Path]:
        """加载允许访问的根目录"""
        roots_str = os.environ.get("MCP_ALLOWED_ROOTS", "")
        if roots_str:
            return [Path(r.strip()) for r in roots_str.split(",")]
        return [Path.home() / "documents"]
    
    def validate_path(self, path: str) -> Optional[str]:
        """验证路径安全性"""
        try:
            resolved = Path(path).resolve()
            
            # 检查是否在允许的范围内
            allowed = any(
                str(resolved).startswith(str(root.resolve()))
                for root in self.allowed_roots
            )
            
            if not allowed:
                return f"路径 {path} 不在允许的访问范围内"
            
            # 检查路径遍历攻击
            if ".." in path:
                return "路径中不允许包含 .."
            
            return None  # 验证通过
        
        except Exception as e:
            return f"路径验证失败：{str(e)}"
    
    def validate_command(self, command: str) -> Optional[str]:
        """验证命令安全性"""
        command_lower = command.lower()
        
        for blocked in self.blocked_commands:
            if blocked in command_lower:
                return f"命令中包含不允许的操作：{blocked}"
        
        return None
    
    def sanitize_input(self, text: str) -> str:
        """清理用户输入"""
        # 移除潜在的注入字符
        dangerous_chars = ["`", "$", "|", ";", "&", "(", ")"]
        for char in dangerous_chars:
            text = text.replace(char, "")
        return text

9.3 性能优化

# performance.py - 性能优化示例
import asyncio
from functools import lru_cache
from typing import Any

class CacheManager:
    def __init__(self, ttl: int = 300):
        self.cache: dict[str, tuple[Any, float]] = {}
        self.ttl = ttl
    
    def get(self, key: str) -> Any:
        """获取缓存"""
        if key in self.cache:
            value, timestamp = self.cache[key]
            if asyncio.get_event_loop().time() - timestamp < self.ttl:
                return value
            else:
                del self.cache[key]
        return None
    
    def set(self, key: str, value: Any):
        """设置缓存"""
        self.cache[key] = (value, asyncio.get_event_loop().time())
    
    def clear(self):
        """清除所有缓存"""
        self.cache.clear()

class RateLimiter:
    def __init__(self, max_requests: int = 100, window: int = 60):
        self.max_requests = max_requests
        self.window = window
        self.requests: dict[str, list[float]] = {}
    
    async def check(self, client_id: str) -> bool:
        """检查是否超过速率限制"""
        now = asyncio.get_event_loop().time()
        
        if client_id not in self.requests:
            self.requests[client_id] = []
        
        # 清理过期的请求记录
        self.requests[client_id] = [
            t for t in self.requests[client_id]
            if now - t < self.window
        ]
        
        # 检查是否超过限制
        if len(self.requests[client_id]) >= self.max_requests:
            return False
        
        self.requests[client_id].append(now)
        return True

9.4 日志与监控

# logging_config.py
import logging
import sys
from datetime import datetime

def setup_logging(level: str = "INFO"):
    """配置日志"""
    logging.basicConfig(
        level=getattr(logging, level),
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
        handlers=[
            logging.StreamHandler(sys.stderr),
            logging.FileHandler(f"mcp_server_{datetime.now().strftime('%Y%m%d')}.log")
        ]
    )

# 在工具调用中添加日志
import logging

logger = logging.getLogger(__name__)

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    logger.info(f"工具调用：{name}，参数：{arguments}")
    
    try:
        result = await handle_tool(name, arguments)
        logger.info(f"工具调用成功：{name}")
        return result
    except Exception as e:
        logger.error(f"工具调用失败：{name}，错误：{str(e)}")
        raise

9.5 测试策略

# test_server.py
import pytest
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

@pytest.fixture
async def client():
    """创建测试客户端"""
    server_params = StdioServerParameters(
        command="python",
        args=["server.py"]
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            yield session

@pytest.mark.asyncio
async def test_list_tools(client):
    """测试工具列表"""
    tools = await client.list_tools()
    assert len(tools.tools) > 0
    
    tool_names = [t.name for t in tools.tools]
    assert "greet" in tool_names
    assert "calculate" in tool_names

@pytest.mark.asyncio
async def test_greet(client):
    """测试打招呼工具"""
    result = await client.call_tool(
        "greet",
        arguments={"name": "测试用户"}
    )
    assert "你好" in result.content[0].text
    assert "测试用户" in result.content[0].text

@pytest.mark.asyncio
async def test_calculate(client):
    """测试计算工具"""
    result = await client.call_tool(
        "calculate",
        arguments={"expression": "2 + 3"}
    )
    assert "5" in result.content[0].text

@pytest.mark.asyncio
async def test_invalid_tool(client):
    """测试无效工具调用"""
    with pytest.raises(Exception):
        await client.call_tool(
            "nonexistent_tool",
            arguments={}
        )

---

第十章 MCP 生态系统与未来展望

10.1 当前生态系统

MCP 生态正在快速发展，目前已有大量官方和社区 Server：

官方 Server：

• @modelcontextprotocol/server-filesystem - 文件系统访问
• @modelcontextprotocol/server-github - GitHub 操作
• @modelcontextprotocol/server-gitlab - GitLab 操作
• @modelcontextprotocol/server-postgres - PostgreSQL 数据库
• @modelcontextprotocol/server-sqlite - SQLite 数据库
• @modelcontextprotocol/server-slack - Slack 消息
• @modelcontextprotocol/server-google-maps - Google Maps
• @modelcontextprotocol/server-brave-search - Brave 搜索
• @modelcontextprotocol/server-memory - 知识图谱记忆

社区 Server：

• Notion、Linear、Jira、Asana 等项目管理工具
• AWS、GCP、Azure 等云服务
• Docker、Kubernetes 等运维工具
• 各种数据库（MongoDB、Redis、Elasticsearch）

10.2 MCP 与 AI Agent 的关系

MCP 是构建 AI Agent 的重要基础设施：

┌─────────────────────────────────────────┐
│              AI Agent                    │
│  (具备推理、规划、执行能力的 AI 系统)     │
│                                          │
│  ┌──────────────────────────────────┐   │
│  │         MCP Client Layer          │   │
│  │  管理多个 MCP Server 连接         │   │
│  └────────────┬─────────────────────┘   │
│               │                          │
└───────────────┼──────────────────────────┘
                │
    ┌───────────┼───────────┐
    │           │           │
┌───▼───┐ ┌────▼───┐ ┌────▼───┐
│文件系统│ │数据库  │ │API服务 │
│Server │ │Server  │ │Server  │
└───────┘ └────────┘ └────────┘

MCP 为 Agent 提供了标准化的工具访问能力，使得 Agent 可以：

• 动态发现可用工具
• 安全地调用外部服务
• 获取实时的上下文信息
• 执行复杂的多步骤任务

10.3 MCP 的未来发展方向

协议扩展：

• 更丰富的权限控制模型
• 支持流式工具调用
• 增强的资源订阅机制
• 跨 Server 编排能力

生态系统：

• 更多官方 Server 支持
• Server 市场和发现机制
• 标准化的测试和认证框架
• 企业级部署方案

工具集成：

• 更多 IDE 原生支持
• 移动端 AI 助手集成
• 硬件设备（IoT）集成
• 多模态能力扩展

10.4 学习资源

官方资源：

• MCP 官方文档：https://modelcontextprotocol.io
• MCP GitHub 仓库：https://github.com/modelcontextprotocol
• MCP 规范：https://spec.modelcontextprotocol.io

社区资源：

• MCP Server 仓库列表
• Discord 社区
• 各语言 SDK 文档

---

第十一章 实战项目：构建企业内部 MCP Server 生态

11.1 项目背景

假设你是一家科技公司的技术负责人，需要为团队构建一套内部 MCP Server 生态，让 AI 编程助手能够安全地访问公司内部系统。

11.2 架构设计

┌─────────────────────────────────────────────────┐
│                 统一网关层                        │
│  (认证、授权、限流、日志)                         │
└───────────────────────┬─────────────────────────┘
                        │
    ┌───────────────────┼───────────────────┐
    │                   │                   │
┌───▼───────┐ ┌────────▼──────┐ ┌──────────▼──┐
│知识库Server│ │项目管理Server │ │监控Server   │
│(Confluence)│ │(Jira/Linear) │ │(Grafana)    │
└───────────┘ └───────────────┘ └─────────────┘
    │                   │                   │
┌───▼───────┐ ┌────────▼──────┐ ┌──────────▼──┐
│Git Server │ │数据库Server   │ │CI/CD Server │
│(GitLab)   │ │(PostgreSQL)   │ │(Jenkins)    │
└───────────┘ └───────────────┘ └─────────────┘

11.3 统一网关实现

# gateway.py - MCP 统一网关
import asyncio
import hashlib
import json
import time
from typing import Optional
from fastapi import FastAPI, HTTPException, Depends, Header
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import httpx

app = FastAPI(title="MCP Gateway")
security = HTTPBearer()

# 配置
API_KEYS = {
    "team-alpha": {
        "key_hash": hashlib.sha256(b"secret-key-alpha").hexdigest(),
        "allowed_servers": ["knowledge-base", "git-server", "project-mgmt"],
        "rate_limit": 100  # 每分钟
    },
    "team-beta": {
        "key_hash": hashlib.sha256(b"secret-key-beta").hexdigest(),
        "allowed_servers": ["database", "monitoring"],
        "rate_limit": 50
    }
}

# Server 注册表
SERVER_REGISTRY = {
    "knowledge-base": {
        "url": "http://localhost:8001",
        "description": "企业知识库",
        "tools": ["search_docs", "read_doc", "create_doc"]
    },
    "git-server": {
        "url": "http://localhost:8002",
        "description": "Git 仓库管理",
        "tools": ["list_repos", "read_file", "create_mr"]
    },
    "project-mgmt": {
        "url": "http://localhost:8003",
        "description": "项目管理",
        "tools": ["list_issues", "create_issue", "update_issue"]
    },
    "database": {
        "url": "http://localhost:8004",
        "description": "数据库访问",
        "tools": ["query", "list_tables", "describe_table"]
    },
    "monitoring": {
        "url": "http://localhost:8005",
        "description": "监控系统",
        "tools": ["get_metrics", "list_alerts", "get_logs"]
    }
}

# 速率限制器
rate_limiters: dict[str, list[float]] = {}

def check_rate_limit(team_id: str, limit: int) -> bool:
    now = time.time()
    if team_id not in rate_limiters:
        rate_limiters[team_id] = []
    
    # 清理过期记录
    rate_limiters[team_id] = [t for t in rate_limiters[team_id] if now - t < 60]
    
    if len(rate_limiters[team_id]) >= limit:
        return False
    
    rate_limiters[team_id].append(now)
    return True

async def verify_api_key(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    token_hash = hashlib.sha256(token.encode()).hexdigest()
    
    for team_id, config in API_KEYS.items():
        if config["key_hash"] == token_hash:
            return team_id, config
    
    raise HTTPException(status_code=401, detail="无效的 API Key")

@app.get("/servers")
async def list_servers(team_id: str = Depends(verify_api_key)):
    team_id, config = team_id
    allowed = config["allowed_servers"]
    
    servers = [
        {**info, "name": name}
        for name, info in SERVER_REGISTRY.items()
        if name in allowed
    ]
    
    return {"servers": servers}

@app.post("/servers/{server_name}/tools/{tool_name}")
async def call_tool(
    server_name: str,
    tool_name: str,
    arguments: dict,
    team_id: str = Depends(verify_api_key)
):
    team_id, config = team_id
    
    # 检查权限
    if server_name not in config["allowed_servers"]:
        raise HTTPException(status_code=403, detail=f"无权访问服务器：{server_name}")
    
    # 检查速率限制
    if not check_rate_limit(team_id, config["rate_limit"]):
        raise HTTPException(status_code=429, detail="请求过于频繁")
    
    # 检查服务器是否存在
    if server_name not in SERVER_REGISTRY:
        raise HTTPException(status_code=404, detail=f"服务器不存在：{server_name}")
    
    server_info = SERVER_REGISTRY[server_name]
    
    # 检查工具是否存在
    if tool_name not in server_info["tools"]:
        raise HTTPException(status_code=404, detail=f"工具不存在：{tool_name}")
    
    # 转发请求到目标 Server
    try:
        async with httpx.AsyncClient(timeout=30) as client:
            response = await client.post(
                f"{server_info['url']}/tools/{tool_name}",
                json={"arguments": arguments},
                headers={"X-Team-ID": team_id}
            )
            return response.json()
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"调用失败：{str(e)}")

@app.get("/health")
async def health():
    """健康检查"""
    server_status = {}
    async with httpx.AsyncClient(timeout=5) as client:
        for name, info in SERVER_REGISTRY.items():
            try:
                response = await client.get(f"{info['url']}/health")
                server_status[name] = "healthy" if response.status_code == 200 else "unhealthy"
            except:
                server_status[name] = "unreachable"
    
    return {
        "status": "healthy",
        "servers": server_status,
        "timestamp": time.time()
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

11.4 知识库 Server

# knowledge_base_server.py
import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import httpx

server = Server("knowledge-base-server")

# 配置
CONFLUENCE_URL = "https://confluence.company.com"
CONFLUENCE_TOKEN = "your-token"

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_docs",
            description="搜索企业知识库文档",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索关键词"},
                    "space": {"type": "string", "description": "知识空间（可选）"},
                    "limit": {"type": "number", "description": "返回数量", "default": 10}
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="read_doc",
            description="读取文档内容",
            inputSchema={
                "type": "object",
                "properties": {
                    "doc_id": {"type": "string", "description": "文档ID"}
                },
                "required": ["doc_id"]
            }
        ),
        Tool(
            name="list_spaces",
            description="列出知识空间",
            inputSchema={"type": "object", "properties": {}}
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    headers = {
        "Authorization": f"Bearer {CONFLUENCE_TOKEN}",
        "Content-Type": "application/json"
    }
    
    try:
        async with httpx.AsyncClient(timeout=30) as client:
            if name == "search_docs":
                query = arguments["query"]
                limit = arguments.get("limit", 10)
                
                response = await client.get(
                    f"{CONFLUENCE_URL}/rest/api/search",
                    params={"cql": f'text ~ "{query}"', "limit": limit},
                    headers=headers
                )
                data = response.json()
                
                results = []
                for item in data.get("results", []):
                    content = item.get("content", {})
                    results.append(
                        f"标题：{content.get('title', 'N/A')}\n"
                        f"ID：{content.get('id', 'N/A')}\n"
                        f"链接：{CONFLUENCE_URL}/pages/viewpage.action?pageId={content.get('id', '')}\n"
                    )
                
                return [TextContent(type="text", text="\n---\n".join(results) or "未找到相关文档")]
            
            elif name == "read_doc":
                doc_id = arguments["doc_id"]
                
                response = await client.get(
                    f"{CONFLUENCE_URL}/rest/api/content/{doc_id}",
                    params={"expand": "body.storage"},
                    headers=headers
                )
                data = response.json()
                
                title = data.get("title", "N/A")
                body = data.get("body", {}).get("storage", {}).get("value", "无内容")
                
                return [TextContent(type="text", text=f"# {title}\n\n{body}")]
            
            elif name == "list_spaces":
                response = await client.get(
                    f"{CONFLUENCE_URL}/rest/api/space",
                    headers=headers
                )
                data = response.json()
                
                spaces = []
                for space in data.get("results", []):
                    spaces.append(f"- {space.get('name', 'N/A')} ({space.get('key', 'N/A')})")
                
                return [TextContent(type="text", text="\n".join(spaces) or "无知识空间")]
            
            else:
                return [TextContent(type="text", text=f"未知工具：{name}")]
    
    except Exception as e:
        return [TextContent(type="text", text=f"错误：{str(e)}")]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options()
        )

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

11.5 部署配置

# docker-compose.yml
version: '3.8'

services:
  mcp-gateway:
    build: ./gateway
    ports:
      - "8000:8000"
    environment:
      - API_KEYS_FILE=/config/api_keys.json
    volumes:
      - ./config:/config
    depends_on:
      - knowledge-base
      - git-server
      - project-mgmt
      - database
      - monitoring
  
  knowledge-base:
    build: ./servers/knowledge-base
    environment:
      - CONFLUENCE_URL=${CONFLUENCE_URL}
      - CONFLUENCE_TOKEN=${CONFLUENCE_TOKEN}
    volumes:
      - ./config:/config
  
  git-server:
    build: ./servers/git-server
    environment:
      - GITLAB_URL=${GITLAB_URL}
      - GITLAB_TOKEN=${GITLAB_TOKEN}
    volumes:
      - ./config:/config
  
  project-mgmt:
    build: ./servers/project-mgmt
    environment:
      - JIRA_URL=${JIRA_URL}
      - JIRA_TOKEN=${JIRA_TOKEN}
    volumes:
      - ./config:/config
  
  database:
    build: ./servers/database
    environment:
      - DATABASE_URL=${DATABASE_URL}
    volumes:
      - ./config:/config
  
  monitoring:
    build: ./servers/monitoring
    environment:
      - GRAFANA_URL=${GRAFANA_URL}
      - GRAFANA_TOKEN=${GRAFANA_TOKEN}
    volumes:
      - ./config:/config

11.6 Claude Desktop 配置示例

{
  "mcpServers": {
    "enterprise-knowledge": {
      "command": "python",
      "args": ["/opt/mcp-servers/knowledge_base_server.py"],
      "env": {
        "CONFLUENCE_URL": "https://confluence.company.com",
        "CONFLUENCE_TOKEN": "your-token"
      }
    },
    "enterprise-git": {
      "command": "python",
      "args": ["/opt/mcp-servers/git_server.py"],
      "env": {
        "GITLAB_URL": "https://gitlab.company.com",
        "GITLAB_TOKEN": "your-token"
      }
    },
    "enterprise-jira": {
      "command": "python",
      "args": ["/opt/mcp-servers/jira_server.py"],
      "env": {
        "JIRA_URL": "https://jira.company.com",
        "JIRA_TOKEN": "your-token"
      }
    }
  }
}

11.7 使用场景示例

配置完成后，可以在 Claude Desktop 中这样使用：

场景1：查询项目文档

用户：帮我查找关于"用户认证模块"的设计文档

Claude：我来帮你搜索知识库中的相关文档。
[调用 search_docs 工具]
找到以下相关文档：
1. 《用户认证模块技术设计》- 文档ID: 12345
2. 《OAuth2.0 集成指南》- 文档ID: 12346
...

场景2：查看 Git 提交

用户：查看 main 分支最近的提交记录

Claude：我来查看 Git 仓库的最近提交。
[调用 git_server 工具]
最近5次提交：
- abc1234: 修复登录页面样式问题 (2小时前)
- def5678: 添加用户注册功能 (5小时前)
...

场景3：创建 Jira Issue

用户：创建一个 Bug：登录页面在 Safari 浏览器上显示异常

Claude：我来创建这个 Bug。
[调用 create_issue 工具]
已创建 Issue：PROJ-123
标题：登录页面在 Safari 浏览器上显示异常
类型：Bug
优先级：High

---

总结

本教程从零开始，全面介绍了 MCP 模型上下文协议：

1. 协议概述：理解了 MCP 的设计理念和应用场景
2. 架构详解：掌握了 Client-Server 模型、传输层、消息格式
3. 核心能力：深入学习了 Resources、Tools、Prompts 三大能力
4. Server 开发：使用 Python 和 Node.js SDK 开发了多个 Server
5. Client 开发：实现了 Python 和 Node.js Client
6. 主流 Server 实战：配置了文件系统、数据库、GitHub 等 Server
7. Claude Desktop 集成：完成了详细的配置指南
8. Cursor/VSCode 集成：支持多种 IDE 环境
9. 最佳实践：学习了安全、性能、测试等方面的经验
10. 生态系统：了解了 MCP 的发展方向
11. 实战项目：构建了完整的企业 MCP Server 生态

MCP 协议正在重新定义 AI 与外部世界的交互方式。作为开发者，掌握 MCP 意味着你能够：

• 为 AI 模型构建标准化的工具接口
• 安全地将 AI 能力集成到现有系统
• 构建可复用、可扩展的 AI 工具生态
• 走在 AI 技术发展的前沿

随着 MCP 生态的不断成熟，它将成为 AI 应用开发的基础设施。现在开始学习和实践 MCP，将为你的技术栈增添一项重要的能力。