MCP 协议深度解析：当 AI Agent 学会「即插即用」——从 USB-C 思想到生产级工具生态完全指南（2026）

前言：一个困扰 AI 开发者四年的终极问题

2022 年 ChatGPT 爆火之后，所有人都在问同一个问题：大模型再强，也只是被困在对话框里的「聪明大脑」，它怎么才能真正连接外部世界——操作数据库、调用 API、读写文件、控制智能硬件？

四年过去了，答案经历了一代代演进。从最早的 Function Calling，到 LangChain 的 Tool 封装，再到各厂各搞一套的私有协议……每家 AI 厂商都说自己的方案最好，但现实是：开发者换一个模型就要重写一遍工具集成代码，每个 AI 应用都要重复造一遍轮子。

直到 2024 年 11 月，Anthropic 发布了 Model Context Protocol（MCP），把这个困扰行业的问题用一句话总结清楚了——

MCP 就是 AI 世界的 USB-C 接口。

就像 USB-C 统一了充电线、数据线、音视频输出等多种功能，MCP 统一了 AI 模型连接外部工具和数据源的方式：只要一个协议，就能让任何兼容 MCP 的 AI 应用，无缝调用任何兼容 MCP 的外部工具。

这不是小打小闹的功能更新，而是整个 AI 工具生态的基础设施革命。2026 年，MCP 已经从极客玩具变成了企业级标配，Cursor、Claude Desktop、VS Code、Cline、Trae 等主流 AI 编程工具全部支持 MCP，GitHub 上相关项目超过 10 万个。

本文将从协议设计哲学出发，深入 MCP 的架构原理、JSON-RPC 通信机制、Server/Client 开发实战、安全模型、性能调优，以及 2026 年最新的生态进展——让你不仅会用，还要懂为什么这样设计，真正掌握这门 AI 时代的「连接协议」。

一、为什么需要 MCP？——从碎片化到标准化的必然之路

1.1 AI 工具集成的「巴别塔困境」

在 MCP 出现之前，AI 开发者面临的是一个噩梦般的碎片化生态：

OpenAI 的方案：

{
  "name": "get_weather",
  "description": "获取天气信息",
  "parameters": {
    "type": "object",
    "properties": {
      "location": {"type": "string"}
    }
  }
}

Anthropic 的方案：

{
  "tool": "weather_tool",
  "description": "查询天气",
  "input_schema": {...}
}

Google 的方案：

{
  "function_declarations": [{
    "name": "weather",
    "parameters": {...}
  }]
}

每换一个大模型，工具定义格式就要重写一遍。更糟糕的是，就算格式统一了，工具的认证方式（API Key、Bearer Token、OAuth）、传输方式（HTTP、WebSocket、本地进程）、错误处理、重试策略全部各不相同。

这意味着一个开发团队想把 AI 接入 PostgreSQL 数据库，要为 GPT、Claude、Gemini 分别写三套集成代码；想把 AI 接入 GitHub，要再写三套；想同时支持多个数据库和 API，工作量直接爆炸。

这就是 AI 工具集成的「巴别塔困境」——每种语言的语法各不相同，彼此无法沟通。

1.2 USB-C 为什么能成功？MCP 学到了什么

USB-C 的成功不是因为技术最先进（雷电 4 肯定比 USB 4.0 强），而是因为它解决了三个根本问题：

1. 物理层统一：不管你是显示器、硬盘还是充电宝，插上就能用
2. 协议层抽象：设备不需要知道上层传输的是什么数据，只关心标准接口
3. 向后兼容：USB 2.0 的设备插在 USB-C 口上依然能用

MCP 完美借鉴了这个思路：

┌─────────────────────────────────────────────────────┐
│                      AI 应用                         │
│   (Claude Desktop / Cursor / Cline / 任意 AI App)   │
└──────────────────────┬──────────────────────────────┘
                       │ MCP Client（统一接口）
                       │ 标准 JSON-RPC 2.0 协议
┌──────────────────────┴──────────────────────────────┐
│                   MCP Host                           │
│              (统一认证、安全沙箱、生命周期管理)            │
└──────────────────────┬──────────────────────────────┘
                       │ MCP Server（可插拔）
        ┌──────────────┼──────────────┬──────────────┐
        ▼              ▼              ▼              ▼
   ┌────────┐    ┌──────────┐  ┌──────────┐  ┌────────┐
   │文件系统 │    │ 数据库   │  │ GitHub API│  │自定义工具│
   │ Server │    │  Server  │  │   Server  │  │ Server │
   └────────┘    └──────────┘  └──────────┘  └────────┘

MCP 的核心设计原则：

• 工具开发者：只需要实现一个 MCP Server，就能被所有 MCP Client 使用
• AI 应用开发者：只需要实现一个 MCP Client，就能调用所有 MCP Server
• 双向解耦：双方独立演进，互不依赖

1.3 MCP 的技术选型：为什么是 JSON-RPC 2.0？

MCP 选择 JSON-RPC 2.0 而不是 gRPC、REST 或 GraphQL，有非常务实的考量：

JSON-RPC 2.0 的优势在于：

• 零依赖：任何语言都有 JSON 解析库，不需要额外引入
• 无状态：非常适合 AI tool calling 的独立调用模式
• 轻量：协议头部极小，减少 token 消耗
• 成熟：2009 年就成为正式规范，经过十多年实战验证

二、MCP 架构深度解析

2.1 三大核心能力：Tools、Resources、Prompts

MCP Server 暴露三类核心能力，理解这三者的区别是掌握 MCP 的关键。

2.1.1 Tools（工具）：AI 模型可以「执行」的操作

Tools 是 AI 模型主动触发的操作。模型通过 tools/call 方法调用服务器上的工具，服务器执行后返回结果。

典型场景：

• 搜索网页、执行 SQL 查询、发送 API 请求
• 操作文件系统、运行 shell 命令
• 调用任何外部服务

// MCP Server 暴露一个 Tool
const myServer: McpServer = new McpServer({
  name: "github-tools",
  version: "1.0.0"
});

myServer.setRequestHandler(ListToolsRequestSchema, async () => {
  return {
    tools: [
      {
        name: "search_repositories",
        description: "搜索 GitHub 仓库，支持关键词、语言、排序",
        inputSchema: {
          type: "object",
          properties: {
            query: { type: "string", description: "搜索关键词" },
            language: { type: "string", description: "编程语言，如 python" },
            sort: { 
              type: "string", 
              enum: ["stars", "forks", "updated"],
              default: "stars"
            }
          },
          required: ["query"]
        }
      },
      {
        name: "create_issue",
        description: "在指定仓库创建 Issue",
        inputSchema: {
          type: "object",
          properties: {
            owner: { type: "string" },
            repo: { type: "string" },
            title: { type: "string" },
            body: { type: "string" }
          },
          required: ["owner", "repo", "title"]
        }
      }
    ]
  };
});

// 工具调用处理器
myServer.setRequestHandler(CallToolRequestSchema, async (request) => {
  const { name, arguments: args } = request.params;
  
  if (name === "search_repositories") {
    const response = await fetch(
      `https://api.github.com/search/repositories?q=${args.query}&sort=${args.sort || "stars"}`,
      { headers: { Authorization: `token ${process.env.GITHUB_TOKEN}` }}
    );
    const data = await response.json();
    return { content: [{ type: "text", text: JSON.stringify(data.items) }] };
  }
  
  if (name === "create_issue") {
    const response = await fetch(
      `https://api.github.com/repos/${args.owner}/${args.repo}/issues`,
      {
        method: "POST",
        headers: { 
          Authorization: `token ${process.env.GITHUB_TOKEN}`,
          "Content-Type": "application/json"
        },
        body: JSON.stringify({ title: args.title, body: args.body })
      }
    );
    return { content: [{ type: "text", text: JSON.stringify(await response.json()) }] };
  }
  
  throw new Error(`Unknown tool: ${name}`);
});

2.1.2 Resources（资源）：AI 可以「读取」的数据

Resources 是 AI 应用主动读取的数据，可以理解为「AI 的知识库」。和 Tools 不同，Resources 是只读的，模型通过 resources/list 发现可用的资源，通过 resources/read 读取内容。

典型场景：

• 本地文件内容（项目代码、配置、数据）
• 数据库 schema、记录
• 知识库文档、API 文档

// 暴露文件系统作为 Resource
myServer.setRequestHandler(ListResourcesRequestSchema, async () => {
  return {
    resources: [
      {
        uri: "file:///Users/me/projects/ai-agent/README.md",
        name: "项目 README",
        description: "AI Agent 项目的说明文档",
        mimeType: "text/markdown"
      },
      {
        uri: "db://production/users/schema",
        name: "生产数据库 Users 表结构",
        description: "PostgreSQL users 表的完整 schema",
        mimeType: "application/json"
      },
      {
        uri: "http://internal-api.company.com/openapi.json",
        name: "内部 API 文档",
        description: "公司内部 REST API 的 OpenAPI 规范",
        mimeType: "application/json"
      }
    ]
  };
});

myServer.setRequestHandler(ReadResourceRequestSchema, async (request) => {
  const { uri } = request.params;
  
  if (uri.startsWith("file://")) {
    const filePath = uri.replace("file://", "");
    const content = await fs.readFile(filePath, "utf-8");
    const mimeType = getMimeType(filePath);
    return { contents: [{ uri, mimeType, text: content }] };
  }
  
  if (uri.startsWith("db://")) {
    // 解析数据库连接和查询
    const { host, table } = parseDbUri(uri);
    const data = await db.query(`SELECT * FROM ${table} LIMIT 10`);
    return { contents: [{ uri, mimeType: "application/json", text: JSON.stringify(data) }] };
  }
  
  throw new Error(`Unknown resource URI: ${uri}`);
});

Tools vs Resources 的关键区别：

2.1.3 Prompts（提示模板）：可复用的提示词片段

Prompts 是服务器预定义的提示词模板，可以帮助 AI 模型更好地调用特定工具或处理特定场景。模型通过 prompts/list 发现可用模板，通过 prompts/get 获取具体内容。

myServer.setRequestHandler(ListPromptsRequestSchema, async () => {
  return {
    prompts: [
      {
        name: "code_review",
        description: "对提交代码进行全面评审，输出结构化反馈",
        arguments: [
          { name: "language", description: "编程语言", required: true },
          { name: "focus", description: "评审重点：security/performance/readability", required: false }
        ]
      },
      {
        name: "database_schema_explainer",
        description: "解释数据库表的结构和关系，生成中文说明",
        arguments: [
          { name: "table_name", description: "表名", required: true }
        ]
      }
    ]
  };
});

myServer.setRequestHandler(GetPromptRequestSchema, async (request) => {
  if (request.params.name === "code_review") {
    return {
      description: "代码评审提示词",
      messages: [{
        role: "user",
        content: {
          type: "text",
          text: `请对以下 ${request.params.arguments?.language || "任意语言"} 代码进行评审。\n\n重点关注：${request.params.arguments?.focus || "整体质量"}\n\n请按以下格式输出：\n1. 代码质量评分（1-10）\n2. 发现的问题（严重/中等/轻微）\n3. 具体改进建议\n4. 优化后的参考代码\n\n待评审代码：\n${request.params.arguments?.code || ""}`
        }
      }]
    };
  }
  throw new Error(`Unknown prompt: ${request.params.name}`);
});

2.2 传输层：stdio vs HTTP+SSE vs WebSocket

MCP 支持三种传输层协议，适用于不同的部署场景：

2.2.1 stdio（标准输入输出）：本地进程通信

最常用、最推荐的模式，适合本地开发和桌面应用。

// Claude Desktop 的 MCP 配置文件
{
  "mcpServers": {
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/Users/me/Desktop"],
      "env": {}
    },
    "postgres": {
      "command": "uv",
      "args": ["run", "mcp-server-postgres"],
      "env": {
        "DATABASE_URL": "postgresql://localhost:5432/mydb"
      }
    }
  }
}

工作原理：

AI 应用 (MCP Client)              MCP Server (独立进程)
       │                                    │
       │  fork() + exec() 启动进程            │
       │ ──────────────────────────────────►│
       │         stdin: JSON-RPC 请求        │
       │                                    │
       │  stdout: JSON-RPC 响应              │
       │ ◄───────────────────────────────── │

stdio 的优势：

• 进程隔离：Server 崩溃不影响 AI 应用
• 零网络开销：本地通信，延迟极低
• 天然安全：不需要暴露网络端口
• 简单部署：打包成 CLI 工具即可

stdio 的劣势：

• 只能在本地使用，无法远程调用
• Server 之间无法直接通信

2.2.2 HTTP + SSE（Server-Sent Events）：远程服务器通信

适合需要远程部署的企业级场景。Client 通过 HTTP POST 发送请求，Server 通过 SSE 流式推送响应。

// TypeScript 实现 HTTP+SSE 传输层
import express from "express";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";

const app = express();
app.use(express.json());

const transport = new StreamableHTTPServerTransport({
  sessionIdGenerator: () => crypto.randomUUID()
});

server.connect(transport);

app.post("/mcp", async (req, res) => {
  // 关键：设置 SSE 流式响应头
  res.setHeader("Content-Type", "text/event-stream");
  res.setHeader("Cache-Control", "no-cache");
  res.setHeader("Connection", "keep-alive");
  res.flushHeaders();
  
  await transport.handleRequest(req, res, req.body);
});

app.get("/mcp", async (req, res) => {
  // SSE 连接，用于服务端主动推送
  res.setHeader("Content-Type", "text/event-stream");
  res.setHeader("Cache-Control", "no-cache");
  res.setHeader("Connection", "keep-alive");
  res.flushHeaders();
  
  await transport.handleRequest(req, res);
});

app.listen(3000);

2.2.3 WebSocket：双向流式通信

适合需要实时双向通信的场景，比如 AI 应用需要实时接收外部事件通知。

import { WebSocketServer } from "ws";
import { WebSocketTransport } from "@modelcontextprotocol/sdk/server/websocket.js";

const wss = new WebSocketServer({ port: 8080 });
wss.on("connection", async (ws) => {
  const transport = new WebSocketTransport(ws);
  await server.connect(transport);
});

2.3 JSON-RPC 2.0 消息格式详解

MCP 的所有通信都基于 JSON-RPC 2.0，理解消息格式是调试和开发 MCP 应用的基础。

请求消息

// 工具调用请求
{
  "jsonrpc": "2.0",
  "id": 42,
  "method": "tools/call",
  "params": {
    "name": "search_repositories",
    "arguments": {
      "query": "rust async",
      "language": "rust",
      "sort": "stars"
    }
  }
}

响应消息

// 成功响应
{
  "jsonrpc": "2.0",
  "id": 42,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "[{\"name\":\"tokio-rs/tokio\", \"stars\":28000}, ...]"
      }
    ],
    "isError": false
  }
}

// 错误响应
{
  "jsonrpc": "2.0",
  "id": 42,
  "error": {
    "code": -32602,
    "message": "Invalid params: missing required argument 'query'",
    "data": {
      "argument": "query"
    }
  }
}

标准错误码：

通知消息（无响应）

// 不需要响应的通知
{
  "jsonrpc": "2.0",
  "method": "notifications/initialized",
  "params": {}
}

三、MCP Server 开发实战：从零构建一个 GitHub 工具服务器

3.1 项目初始化

本节用一个完整的 GitHub MCP Server 演示生产级开发流程。

# 使用 uv 初始化项目（比 pip 快 10-100 倍）
mkdir github-mcp-server && cd github-mcp-server
uv init --name github-mcp-server

# 安装 MCP SDK（Python）
uv add "mcp[cli]" httpx aiohttp pydantic python-dotenv

# 安装类型提示和开发工具
uv add --dev pytest pytest-asyncio mypy ruff

3.2 完整的 MCP Server 实现

"""
GitHub MCP Server - 生产级完整实现
支持：仓库搜索、Issue 管理、PR 评审、文件操作
"""

import os
import json
from typing import Any, TypedDict
from dataclasses import dataclass, field

import httpx
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
    Tool, TextContent, Resource, Prompt,
    ListToolsRequest, CallToolRequest,
    ListResourcesRequest, ReadResourceRequest,
    GetPromptRequest
)

# ============================================================
# 配置管理
# ============================================================

@dataclass
class GitHubConfig:
    token: str = field(default_factory=lambda: os.getenv("GITHUB_TOKEN", ""))
    owner: str = field(default_factory=lambda: os.getenv("GITHUB_OWNER", ""))
    base_url: str = "https://api.github.com"
    per_page: int = 30
    timeout: int = 30

    def headers(self) -> dict[str, str]:
        headers = {
            "Accept": "application/vnd.github+json",
            "X-GitHub-Api-Version": "2022-11-28",
            "User-Agent": "MCP-GitHub-Server/1.0"
        }
        if self.token:
            headers["Authorization"] = f"Bearer {self.token}"
        return headers

config = GitHubConfig()

# ============================================================
# MCP Server 初始化
# ============================================================

server = Server("github-mcp-server", version="1.0.0")

# ============================================================
# 辅助工具函数
# ============================================================

async def github_get(path: str, params: dict | None = None) -> dict[str, Any]:
    """统一的 GitHub API 请求方法"""
    url = f"{config.base_url}{path}"
    async with httpx.AsyncClient(
        timeout=httpx.Timeout(config.timeout),
        headers=config.headers()
    ) as client:
        response = await client.get(url, params=params)
        
        # 处理速率限制
        if response.status_code == 403:
            reset_time = response.headers.get("X-RateLimit-Reset")
            if reset_time:
                import time
                wait_seconds = int(reset_time) - int(time.time())
                raise Exception(
                    f"GitHub API 速率限制已达到。请等待 {wait_seconds} 秒后重试。"
                )
            raise Exception("GitHub API 速率限制已达到。")
        
        if response.status_code == 404:
            raise Exception(f"GitHub 资源未找到: {path}")
        
        response.raise_for_status()
        return response.json()

async def github_post(path: str, data: dict) -> dict[str, Any]:
    """统一的 GitHub API POST 请求方法"""
    url = f"{config.base_url}{path}"
    async with httpx.AsyncClient(
        timeout=httpx.Timeout(config.timeout),
        headers=config.headers()
    ) as client:
        response = await client.post(url, json=data)
        response.raise_for_status()
        return response.json()

def format_repository(repo: dict) -> str:
    """格式化仓库信息为可读文本"""
    return (
        f"**{repo['full_name']}**\n"
        f"⭐ {repo['stargazers_count']:,} | "
        f"🍴 {repo['forks_count']:,} | "
        f"👁 {repo['watchers_count']:,}\n"
        f"📝 {repo['description'] or '无描述'}\n"
        f"🔗 {repo['html_url']}\n"
        f"🏷️ 语言: {repo.get('language', '未知')} | "
        f"📅 更新时间: {repo['updated_at'][:10]}\n"
    )

# ============================================================
# Tool 定义
# ============================================================

@server.list_tools()
async def list_tools() -> list[Tool]:
    """列出所有可用的 GitHub 工具"""
    return [
        Tool(
            name="search_repos",
            description="搜索 GitHub 仓库，支持关键词、语言、排序和过滤条件。"
                        "适合寻找开源项目、分析技术趋势、研究竞品。",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索关键词，支持 GitHub 高级搜索语法。"
                                       "如: 'machine learning language:python stars:>1000'"
                    },
                    "max_results": {
                        "type": "integer",
                        "description": "最大返回结果数，默认 10，最大 100",
                        "default": 10,
                        "minimum": 1,
                        "maximum": 100
                    },
                    "sort": {
                        "type": "string",
                        "description": "排序方式",
                        "enum": ["stars", "forks", "updated"],
                        "default": "stars"
                    }
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="get_repo_info",
            description="获取指定仓库的详细信息，包括统计、贡献者、提交活动等。"
                        "用于分析仓库健康度和活跃度。",
            inputSchema={
                "type": "object",
                "properties": {
                    "owner": {"type": "string", "description": "仓库所有者"},
                    "repo": {"type": "string", "description": "仓库名称"}
                },
                "required": ["owner", "repo"]
            }
        ),
        Tool(
            name="list_issues",
            description="列出仓库的 Issue，支持按状态、标签、负责人过滤。"
                        "用于追踪项目进度、发现 good first issue。",
            inputSchema={
                "type": "object",
                "properties": {
                    "owner": {"type": "string", "description": "仓库所有者"},
                    "repo": {"type": "string", "description": "仓库名称"},
                    "state": {
                        "type": "string",
                        "description": "Issue 状态",
                        "enum": ["open", "closed", "all"],
                        "default": "open"
                    },
                    "labels": {
                        "type": "string",
                        "description": "逗号分隔的标签，如: 'bug,help wanted'"
                    },
                    "max_results": {
                        "type": "integer",
                        "default": 20,
                        "maximum": 100
                    }
                },
                "required": ["owner", "repo"]
            }
        ),
        Tool(
            name="create_issue",
            description="在指定仓库创建 Issue。可设置标题、正文、标签、指派人。"
                        "用于自动化工单创建、bug 报告。",
            inputSchema={
                "type": "object",
                "properties": {
                    "owner": {"type": "string", "description": "仓库所有者"},
                    "repo": {"type": "string", "description": "仓库名称"},
                    "title": {"type": "string", "description": "Issue 标题"},
                    "body": {
                        "type": "string",
                        "description": "Issue 正文，支持 Markdown"
                    },
                    "labels": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "标签列表"
                    }
                },
                "required": ["owner", "repo", "title"]
            }
        ),
        Tool(
            name="create_pull_request",
            description="创建 Pull Request。需要指定源分支和目标分支。"
                        "用于代码审查工作流自动化。",
            inputSchema={
                "type": "object",
                "properties": {
                    "owner": {"type": "string"},
                    "repo": {"type": "string"},
                    "title": {"type": "string"},
                    "body": {"type": "string"},
                    "head": {"type": "string", "description": "源分支名"},
                    "base": {
                        "type": "string",
                        "description": "目标分支名",
                        "default": "main"
                    },
                    "draft": {
                        "type": "boolean",
                        "description": "是否为草稿 PR",
                        "default": False
                    }
                },
                "required": ["owner", "repo", "title", "head"]
            }
        ),
        Tool(
            name="get_file_content",
            description="获取仓库中指定文件的原始内容。"
                        "用于代码审查、文档读取、配置文件分析。",
            inputSchema={
                "type": "object",
                "properties": {
                    "owner": {"type": "string"},
                    "repo": {"type": "string"},
                    "path": {"type": "string", "description": "文件路径，如: 'src/main.py'"},
                    "ref": {
                        "type": "string",
                        "description": "分支名、tag 或 commit SHA，默认使用默认分支"
                    }
                },
                "required": ["owner", "repo", "path"]
            }
        ),
        Tool(
            name="list_commits",
            description="获取仓库的最近提交记录，了解项目活跃度和开发动态。",
            inputSchema={
                "type": "object",
                "properties": {
                    "owner": {"type": "string"},
                    "repo": {"type": "string"},
                    "branch": {
                        "type": "string",
                        "description": "分支名，默认主分支"
                    },
                    "max_results": {
                        "type": "integer",
                        "default": 20,
                        "maximum": 100
                    }
                },
                "required": ["owner", "repo"]
            }
        )
    ]

# ============================================================
# Tool 调用处理器
# ============================================================

@server.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
    """处理工具调用请求"""
    
    try:
        if name == "search_repos":
            params = {
                "q": arguments["query"],
                "sort": arguments.get("sort", "stars"),
                "per_page": min(arguments.get("max_results", 10), 100)
            }
            data = await github_get("/search/repositories", params)
            
            if not data.get("items"):
                return [TextContent(type="text", text="未找到匹配的仓库。")]
            
            results = [
                f"## 搜索结果 ({data['total_count']:,} 个结果)\n"
            ]
            for repo in data["items"][:params["per_page"]]:
                results.append(format_repository(repo))
                results.append("---")
            
            total = len(data["items"])
            shown = min(total, params["per_page"])
            results.append(f"\n共找到 {data['total_count']:,} 个仓库，展示前 {shown} 个。")
            
            return [TextContent(type="text", text="\n".join(results))]
        
        elif name == "get_repo_info":
            owner, repo = arguments["owner"], arguments["repo"]
            
            # 并行获取多个数据源
            async with httpx.AsyncClient(timeout=httpx.Timeout(config.timeout)) as client:
                tasks = [
                    client.get(f"{config.base_url}/repos/{owner}/{repo}", headers=config.headers()),
                    client.get(f"{config.base_url}/repos/{owner}/{repo}/contributors", headers=config.headers(), params={"per_page": "5"}),
                    client.get(f"{config.base_url}/repos/{owner}/{repo}/traffic/views", headers=config.headers()),
                ]
                repo_data, contributors, traffic = await asyncio.gather(*tasks)
                repo_data.raise_for_status()
                contributors.raise_for_status()
                traffic.raise_for_status()
            
            repo = repo_data.json()
            contribs = contributors.json()
            views = traffic.json()
            
            info = f"""# {repo['full_name']}

## 基本信息
- **描述**: {repo['description'] or '无'}
- **语言**: {repo.get('language', '未知')}
- **许可证**: {repo.get('license', {}).get('name', '未知')}
- **创建时间**: {repo['created_at'][:10]}
- **更新时间**: {repo['updated_at'][:10]}

## 统计数据
- ⭐ Stars: {repo['stargazers_count']:,}
- 🍴 Forks: {repo['forks_count']:,}
- 👁 Watchers: {repo['watchers_count']:,}
- 🐛 Issues: {repo['open_issues_count']:,}
- 📦 Releases: {repo['releases_url'].count('{') and '有' or '有'} (点击查看)

## 活跃度
- 最近一周浏览量: {views.get('count', 'N/A')}
- 最近一周唯一访客: {views.get('uniques', 'N/A')}

## Top 贡献者
{chr(10).join(f"{i+1}. [{c['login']}](https://github.com/{c['login']}) ({c['contributions']} 次提交)" for i, c in enumerate(contribs[:5]))}

## 链接
- 🌐 主页: {repo['html_url']}
- 📖 文档: {repo.get('homepage') or '无'}
- 🐛 Issue: {repo['html_url']}/issues
- 🔀 Pull Requests: {repo['html_url']}/pulls
"""
            return [TextContent(type="text", text=info)]
        
        elif name == "list_issues":
            owner, repo = arguments["owner"], arguments["repo"]
            params = {
                "state": arguments.get("state", "open"),
                "per_page": min(arguments.get("max_results", 20), 100)
            }
            if arguments.get("labels"):
                params["labels"] = arguments["labels"]
            
            data = await github_get(f"/repos/{owner}/{repo}/issues", params)
            
            if not data:
                return [TextContent(type="text", text="没有找到 Issue。")]
            
            results = [f"## {repo} 的 Issues ({arguments.get('state', 'open')}) 共 {len(data)} 条\n"]
            for issue in data:
                labels = ", ".join(f"`{l['name']}`" for l in issue.get("labels", []))
                results.append(
                    f"### [{issue['number']}] {issue['title']}\n"
                    f"🔗 {issue['html_url']}\n"
                    f"👤 {issue['user']['login']} | "
                    f"💬 {issue['comments']} 评论 | "
                    f"📅 {issue['created_at'][:10]}\n"
                    + (f"🏷️ {labels}\n" if labels else "")
                    + (f"📌 指派给: {', '.join(a['login'] for a in issue.get('assignees', []))}\n" if issue.get('assignees') else "")
                )
            
            return [TextContent(type="text", text="\n".join(results))]
        
        elif name == "create_issue":
            owner, repo = arguments["owner"], arguments["repo"]
            data = {
                "title": arguments["title"],
                "body": arguments.get("body", ""),
                "labels": arguments.get("labels", [])
            }
            result = await github_post(f"/repos/{owner}/{repo}/issues", data)
            return [TextContent(
                type="text",
                text=f"✅ Issue 创建成功！\n\n**标题**: {result['title']}\n**编号**: #{result['number']}\n**链接**: {result['html_url']}"
            )]
        
        elif name == "create_pull_request":
            owner, repo = arguments["owner"], arguments["repo"]
            data = {
                "title": arguments["title"],
                "body": arguments.get("body", ""),
                "head": arguments["head"],
                "base": arguments.get("base", "main"),
                "draft": arguments.get("draft", False)
            }
            result = await github_post(f"/repos/{owner}/{repo}/pulls", data)
            return [TextContent(
                type="text",
                text=f"✅ Pull Request 创建成功！\n\n**标题**: {result['title']}\n**编号**: #{result['number']}\n**链接**: {result['html_url']}\n**状态**: {'草稿' if result.get('draft') else '活跃'}"
            )]
        
        elif name == "get_file_content":
            owner, repo = arguments["owner"], arguments["repo"]
            path = arguments["path"]
            ref = arguments.get("ref")
            
            url = f"/repos/{owner}/{repo}/contents/{path}"
            if ref:
                url += f"?ref={ref}"
            
            content_data = await github_get(url)
            
            # 文件内容是 Base64 编码的
            import base64
            content = base64.b64decode(content_data["content"]).decode("utf-8")
            
            # 截断超长文件（超过 5000 行）
            lines = content.split("\n")
            truncated = len(lines) > 5000
            if truncated:
                content = "\n".join(lines[:5000])
                content += f"\n\n... [文件已被截断，共 {len(lines)} 行，显示前 5000 行]"
            
            return [TextContent(
                type="text",
                text=f"**文件**: {owner}/{repo}/{path}\n**分支**: {ref or '默认分支'}\n\n```\n{content}\n```"
            )]
        
        elif name == "list_commits":
            owner, repo = arguments["owner"], arguments["repo"]
            params = {
                "per_page": min(arguments.get("max_results", 20), 100)
            }
            if arguments.get("branch"):
                params["sha"] = arguments["branch"]
            
            data = await github_get(f"/repos/{owner}/{repo}/commits", params)
            
            results = [f"## {owner}/{repo} 的最近提交记录\n"]
            for commit in data:
                results.append(
                    f"**{commit['sha'][:7]}** {commit['commit']['message'].split(chr(10))[0]}\n"
                    f"👤 {commit['commit']['author']['name']} | "
                    f"📅 {commit['commit']['author']['date'][:10]} | "
                    f"📁 {commit.get('files', [{}])[0].get('filename', '多文件') if 'files' in commit else '多文件'}\n"
                )
            
            return [TextContent(type="text", text="\n".join(results))]
        
        else:
            raise ValueError(f"未知工具: {name}")
    
    except httpx.HTTPStatusError as e:
        return [TextContent(
            type="text",
            text=f"❌ GitHub API 错误: {e.response.status_code} - {e.response.text}"
        )]
    except Exception as e:
        return [TextContent(type="text", text=f"❌ 错误: {str(e)}")]

# ============================================================
# Resource 定义
# ============================================================

@server.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri=f"github://{config.owner}/repos",
            name="我的仓库列表",
            description="当前配置的 GitHub 用户的仓库列表",
            mimeType="application/json"
        )
    ]

# ============================================================
# 启动服务器
# ============================================================

async def main():
    import asyncio
    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())

3.3 Claude Desktop 配置

将上述 Server 配置到 Claude Desktop：

// macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
// Windows: %APPDATA%/Claude/claude_desktop_config.json
{
  "mcpServers": {
    "github": {
      "command": "uv",
      "args": ["run", "--directory", "/path/to/github-mcp-server", "python", "-m", "github_mcp_server"],
      "env": {
        "GITHUB_TOKEN": "ghp_xxxxxxxxxxxx",
        "GITHUB_OWNER": "your-username"
      }
    }
  }
}

配置完成后重启 Claude Desktop，AI 就能直接调用 GitHub 工具了：

用户: 帮我搜索最近 Rust 异步编程相关的仓库，找出 star 数超过 1000 的

AI: (调用 search_repos，参数: query="rust async", sort="stars")
    
    找到以下仓库：
    
    1. **tokio-rs/tokio** ⭐ 28,000+
       最流行的 Rust 异步运行时...
    
    2. **async-graphql/async-graphql** ⭐ 8,000+
       Rust 的 GraphQL 实现...

四、安全模型深度分析：MCP 不只是「能用」，还要「敢用」

4.1 信任模型的三层架构

MCP 的安全设计基于一个核心原则：最小权限原则 + 显式授权。

┌─────────────────────────────────────────────┐
│              MCP Host（Claude Desktop 等）    │
│                                             │
│  认证层: Bearer Token / API Key / OAuth     │
│  权限层: RBAC（基于角色的访问控制）           │
│  沙箱层: 进程隔离 + 资源限制                  │
└─────────────────────────────────────────────┘

第一层：传输安全

// stdio 模式：进程隔离，无需网络认证
// 但敏感操作需要用户显式确认

// HTTP+SSE 模式：支持 TLS + JWT
{
  "security": {
    "transport": "https",
    "auth": {
      "type": "bearer",
      "token_env": "MCP_API_TOKEN"
    },
    "allowed_origins": ["https://claude.ai", "https://anthropic.com"]
  }
}

第二层：工具级别权限控制

// 在 MCP Server 中实现细粒度权限控制
class PermissionedMcpServer extends McpServer {
  private userPermissions: Map<string, string[]> = new Map();
  
  constructor() {
    super({ name: "permissioned-server", version: "1.0.0" });
    this.setupPermissionMiddleware();
  }
  
  private setupPermissionMiddleware() {
    // 拦截所有工具调用，进行权限检查
    this.setRequestHandler(CallToolRequestSchema, async (request) => {
      const userId = request.headers["x-user-id"];
      const toolName = request.params.name;
      
      const allowed = this.userPermissions.get(userId) || [];
      if (!allowed.includes(toolName)) {
        throw new Error(
          `权限不足: 用户 ${userId} 无权调用工具 ${toolName}`
        );
      }
      
      return this.callTool(toolName, request.params.arguments);
    });
  }
  
  // 管理员可以动态分配权限
  setUserPermissions(userId: string, tools: string[]) {
    this.userPermissions.set(userId, tools);
  }
}

第三层：用户确认机制

MCP 设计了危险操作二次确认机制：

// 高危工具需要用户手动确认
const DANGEROUS_TOOLS = [
  "delete_file",
  "drop_database", 
  "execute_shell",
  "send_email",
  "delete_repository"
];

server.setRequestHandler(CallToolRequestSchema, async (request) => {
  const toolName = request.params.name;
  
  if (DANGEROUS_TOOLS.includes(toolName)) {
    // 返回确认请求
    return {
      content: [{
        type: "request_confirmation",
        data: {
          tool: toolName,
          arguments: request.params.arguments,
          warning: getDangerWarning(toolName)
        }
      }]
    };
  }
  
  return this.callTool(toolName, request.params.arguments);
});

4.2 企业级安全最佳实践

1. Network Isolation（网络隔离）：

# MCP Server 只监听本地端口，不对外暴露
server.listen("127.0.0.1", 8080)

# 如果需要远程访问，通过 VPN 或 SSH 隧道
ssh -L 8080:localhost:8080 user@mcp-server-host

2. Secrets 管理：

# 永远不要在代码或配置文件中硬编码密钥
# 使用环境变量或密钥管理服务

# ✅ 正确
config = GitHubConfig(token=os.getenv("GITHUB_TOKEN"))

# ✅ 更好：使用专门的密钥管理
from keyring import get_keyring
token = get_keyring().get_password("github", "mcp-server")

# ✅ 企业级：使用 Vault
import hvac
vault = hvac.Client(url=os.getenv("VAULT_ADDR"))
secret = vault.secrets.kv.v2.read_secret_version(
    path="github/mcp-server",
    mount_point="secret"
)

3. 审计日志：

import structlog
from datetime import datetime

logger = structlog.get_logger()

@server.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
    # 记录所有工具调用
    logger.info(
        "tool_call",
        tool=name,
        arguments=sanitize_for_logging(arguments),  # 脱敏敏感信息
        timestamp=datetime.utcnow().isoformat(),
        user=os.getenv("MCP_USER_ID", "unknown")
    )
    
    try:
        result = await execute_tool(name, arguments)
        logger.info("tool_success", tool=name)
        return result
    except Exception as e:
        logger.error("tool_error", tool=name, error=str(e))
        raise

五、性能优化：让 MCP 在生产环境跑得更稳更快

5.1 连接池与并发管理

import asyncio
from contextlib import asynccontextmanager

class GitHubConnectionPool:
    """GitHub API 连接池，避免频繁建立连接"""
    
    def __init__(self, max_connections: int = 10, rate_limit_delay: float = 1.0):
        self.semaphore = asyncio.Semaphore(max_connections)
        self.rate_limit_delay = rate_limit_delay
        self.last_request_time = 0.0
        self._lock = asyncio.Lock()
    
    @asynccontextmanager
    async def acquire(self):
        async with self.semaphore:
            async with self._lock:
                # 速率限制：两次请求之间至少间隔 rate_limit_delay 秒
                elapsed = asyncio.get_event_loop().time() - self.last_request_time
                if elapsed < self.rate_limit_delay:
                    await asyncio.sleep(self.rate_limit_delay - elapsed)
                self.last_request_time = asyncio.get_event_loop().time()
            
            yield self.client

pool = GitHubConnectionPool(max_connections=5, rate_limit_delay=0.5)

5.2 响应缓存：减少 token 消耗的利器

from functools import lru_cache, wraps
import hashlib
import json
import time

def cache_response(ttl_seconds: int = 300):
    """缓存工具响应，减少重复调用"""
    cache: dict[str, tuple[float, Any]] = {}
    
    def decorator(func):
        @wraps(func)
        async def wrapper(name: str, arguments: dict) -> list[TextContent]:
            # 生成缓存键（包含参数哈希）
            cache_key = f"{name}:{hashlib.md5(json.dumps(arguments, sort_keys=True).encode()).hexdigest()}"
            
            now = time.time()
            if cache_key in cache:
                expires_at, cached_result = cache[cache_key]
                if now < expires_at:
                    # 命中缓存，标记来源
                    return [TextContent(
                        type="text",
                        text=f"[缓存] {cached_result.text}"
                    )]
            
            # 未命中缓存，执行真实调用
            result = await func(name, arguments)
            
            # 写入缓存
            cache[cache_key] = (now + ttl_seconds, result[0])
            
            # 清理过期缓存
            expired = [k for k, (exp, _) in cache.items() if now >= exp]
            for k in expired:
                del cache[k]
            
            return result
        
        return wrapper
    return decorator

# 使用缓存
@cache_response(ttl_seconds=300)  # 5 分钟缓存
async def search_repos_cached(name: str, arguments: dict) -> list[TextContent]:
    return await search_repos(name, arguments)

5.3 流式响应：给 AI 更快的首字节体验

import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";

// 使用流式传输，AI 可以边接收边处理
server.setRequestHandler(CallToolRequestSchema, async (request, sendResult) => {
  const { name, arguments: args } = request.params;
  
  if (name === "stream_logs") {
    // 模拟日志流
    const transport = new StreamableHTTPServerTransport(/*...*/);
    
    // 立即发送初始响应
    await sendResult({
      content: [{ type: "text", text: "开始接收日志流...\n" }],
      _isStreaming: true
    });
    
    // 流式发送日志内容
    const logStream = getLogStream(args.service);
    for await (const logLine of logStream) {
      await sendResult({
        content: [{ type: "text", text: logLine }]
      });
    }
    
    await sendResult({
      content: [{ type: "text", text: "\n[日志流结束]" }],
      _isStreaming: false
    });
  }
});

六、2026 年 MCP 生态全景图

6.1 官方与社区 Server 一览

6.2 MCP 在 AI 编程工具中的实际应用

Cursor：

// .cursor/mcp.json
{
  "mcpServers": {
    "database": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-postgres"],
      "env": {
        "DATABASE_URL": "postgresql://prod-db:5432/app"
      }
    },
    "aws": {
      "command": "python",
      "args": ["/Users/me/tools/aws-mcp-server/main.py"],
      "env": {
        "AWS_ACCESS_KEY_ID": "${AWS_ACCESS_KEY_ID}",
        "AWS_SECRET_ACCESS_KEY": "${AWS_SECRET_ACCESS_KEY}"
      }
    }
  }
}

配置后，在 Cursor 中直接说：

帮我查一下生产数据库里过去 7 天注册用户数最多的前 10 个城市，用中文给我做个分析报告

Cursor 会自动调用 PostgreSQL MCP Server 执行 SQL，把结果整理成报告——整个过程对用户透明，不需要关心底层数据库连接细节。

6.3 MCP 的局限性：它不是银弹

冷静客观地说，MCP 也有其局限性：

1. 协议仍然年轻，版本稳定性有待观察

截至 2026 年，MCP 规范仍在快速迭代中。部分 API 存在破坏性变更的可能，企业在生产环境中使用时应锁定版本号。

2. 认证生态尚不完善

MCP 目前没有统一的认证标准。每个 Server 自己实现认证，GitHub 用 Token，数据库用连接字符串，云服务用 IAM 凭证。这在短期内造成了碎片化。

3. 分布式 MCP 仍在探索

多 Server 协同、Server-to-Server 通信等场景，官方尚未给出标准方案。企业内部往往需要自行封装。

4. Tool Calling 的 token 消耗

MCP 的 tool descriptions 和 schema 本身会消耗上下文 token。当 Server 暴露大量工具时，需要精心设计 schema 精简策略。

七、实战：构建一个「上下文感知」的 RAG MCP Server

7.1 需求分析

我们需要一个 MCP Server，能做到：

• 读取项目文档和代码
• 理解当前对话的上下文
• 智能检索相关内容
• 让 AI 在回答时引用真实项目信息

7.2 完整实现

"""
RAG MCP Server - 让 AI 拥有项目专属知识库
使用向量数据库（ChromaDB）存储文档，支持语义搜索
"""

import os
import hashlib
from pathlib import Path
from typing import Any
from dataclasses import dataclass

import chromadb
from chromadb.config import Settings
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent, Resource

# ============================================================
# 向量数据库初始化
# ============================================================

@dataclass
class RAGConfig:
    data_dir: Path = Path.home() / ".mcp-rag" / "data"
    persist_dir: Path = Path.home() / ".mcp-rag" / "chroma"
    chunk_size: int = 1000
    chunk_overlap: int = 200
    max_results: int = 5

config = RAGConfig()
config.data_dir.mkdir(parents=True, exist_ok=True)
config.persist_dir.mkdir(parents=True, exist_ok=True)

# ChromaDB 客户端
chroma_client = chromadb.PersistentClient(path=str(config.persist_dir))
collection = chroma_client.get_or_create_collection(
    name="project_docs",
    metadata={"description": "项目文档向量库"}
)

# ============================================================
# 文档处理
# ============================================================

def chunk_text(text: str, chunk_size: int = 1000, overlap: int = 200) -> list[str]:
    """将长文本分块，支持重叠以保持上下文连续性"""
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunks.append(text[start:end])
        start = end - overlap
    return chunks

def compute_file_hash(file_path: Path) -> str:
    """计算文件哈希，用于检测内容变化"""
    return hashlib.sha256(file_path.read_bytes()).hexdigest()[:16]

async def index_file(file_path: Path) -> dict[str, Any]:
    """索引单个文件到向量库"""
    if file_path.suffix not in [".md", ".txt", ".py", ".js", ".ts", ".json", ".yaml", ".yml", ".toml", ".rst"]:
        return {"skipped": True, "reason": "不支持的文件类型"}
    
    content = file_path.read_text(encoding="utf-8", errors="ignore")
    if len(content) < 50:
        return {"skipped": True, "reason": "内容过短"}
    
    chunks = chunk_text(content, config.chunk_size, config.chunk_overlap)
    
    # 为每个 chunk 生成 ID
    file_hash = compute_file_hash(file_path)
    ids = [f"{file_hash}_{i}" for i in range(len(chunks))]
    metadatas = [{
        "file": str(file_path),
        "chunk_index": i,
        "total_chunks": len(chunks),
        "file_hash": file_hash
    } for i in range(len(chunks))]
    
    # 存储到 ChromaDB
    # 注意：实际使用中需要调用 embedding API（如 OpenAI）获取向量
    # 这里用占位符，实际生产环境请替换为真实 embedding
    try:
        collection.upsert(
            ids=ids,
            documents=chunks,
            metadatas=metadatas
        )
    except Exception as e:
        return {"error": str(e)}
    
    return {
        "indexed": True,
        "file": str(file_path),
        "chunks": len(chunks)
    }

async def index_directory(dir_path: Path, extensions: list[str] | None = None) -> dict[str, Any]:
    """递归索引整个目录"""
    extensions = extensions or [".md", ".txt", ".py", ".js", ".ts", ".json"]
    results = {"indexed": 0, "skipped": 0, "errors": []}
    
    for file_path in dir_path.rglob("*"):
        if file_path.is_file() and file_path.suffix in extensions:
            result = await index_file(file_path)
            if result.get("indexed"):
                results["indexed"] += 1
            elif result.get("skipped"):
                results["skipped"] += 1
            elif result.get("error"):
                results["errors"].append({str(file_path): result["error"]})
    
    return results

# ============================================================
# MCP Server 定义
# ============================================================

server = Server("rag-mcp-server", version="1.0.0")

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="index_directory",
            description="将指定目录下的所有代码和文档文件索引到向量数据库。"
                        "支持 .py/.js/.ts/.md/.json/.yaml 等格式。"
                        "用于建立项目知识库，使 AI 能够基于真实项目内容回答问题。",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "要索引的目录路径"},
                    "extensions": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "要索引的文件扩展名列表"
                    }
                },
                "required": ["path"]
            }
        ),
        Tool(
            name="search_knowledge_base",
            description="在项目知识库中进行语义搜索，找到与查询最相关的文档片段。"
                        "返回相关内容的原文和来源文件路径。用于 AI 回答项目相关问题时引用真实代码和文档。",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索查询（自然语言），如 '如何配置数据库连接池'"
                    },
                    "max_results": {
                        "type": "integer",
                        "description": "最多返回的相关片段数",
                        "default": 5,
                        "maximum": 20
                    },
                    "filter_file": {
                        "type": "string",
                        "description": "只搜索指定文件或目录（如：'src/api'）"
                    }
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="get_file_context",
            description="获取指定文件的内容，并标注其是否已被索引。"
                        "用于确认文件是否已建立向量索引。",
            inputSchema={
                "type": "object",
                "properties": {
                    "path": {"type": "string"},
                    "max_lines": {
                        "type": "integer",
                        "description": "最多显示行数",
                        "default": 200
                    }
                },
                "required": ["path"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
    
    if name == "index_directory":
        dir_path = Path(arguments["path"]).expanduser().resolve()
        if not dir_path.exists():
            return [TextContent(type="text", text=f"❌ 目录不存在: {dir_path}")]
        if not dir_path.is_dir():
            return [TextContent(type="text", text=f"❌ 不是有效目录: {dir_path}")]
        
        results = await index_directory(
            dir_path,
            arguments.get("extensions")
        )
        
        return [TextContent(
            type="text",
            text=f"✅ 索引完成！\n\n- 成功索引: {results['indexed']} 个文件\n"
                 f"- 跳过: {results['skipped']} 个文件\n"
                 f"- 错误: {len(results['errors'])} 个\n"
                 + (f"\n错误详情:\n" + "\n".join(str(e) for e in results['errors']) if results['errors'] else "")
        )]
    
    elif name == "search_knowledge_base":
        query = arguments["query"]
        max_results = arguments.get("max_results", 5)
        
        # 执行向量搜索
        # 注意：实际生产中需要先调用 embedding API
        # results = collection.query(query_texts=[query], n_results=max_results)
        
        # 模拟搜索结果（实际使用中替换为真实向量搜索）
        results = collection.query(
            query_texts=[query],
            n_results=max_results,
            where=arguments.get("filter_file") and {"file": {"$contains": arguments["filter_file"]}}
        )
        
        if not results["documents"] or not results["documents"][0]:
            return [TextContent(
                type="text",
                text=f"❌ 在知识库中未找到与「{query}」相关的内容。\n\n"
                     f"提示：可以先运行 index_directory 索引项目目录。"
            )]
        
        output = [f"## 知识库搜索结果: {query}\n"]
        for i, (doc, meta) in enumerate(zip(results["documents"][0], results["metadatas"][0])):
            output.append(
                f"\n### {i+1}. {meta['file']} (chunk {meta['chunk_index']+1}/{meta['total_chunks']})\n"
                f"```\n{doc}\n```\n"
            )
        
        output.append(f"\n_共返回 {len(results['documents'][0])} 条相关结果_")
        
        return [TextContent(type="text", text="\n".join(output))]
    
    elif name == "get_file_context":
        file_path = Path(arguments["path"]).expanduser().resolve()
        if not file_path.exists():
            return [TextContent(type="text", text=f"❌ 文件不存在: {file_path}")]
        
        content = file_path.read_text(encoding="utf-8", errors="ignore")
        lines = content.split("\n")
        max_lines = arguments.get("max_lines", 200)
        
        # 检查是否已索引
        file_hash = compute_file_hash(file_path)
        existing = collection.get(where={"file_hash": file_hash})
        indexed = len(existing.get("ids", [])) > 0
        
        display_content = "\n".join(lines[:max_lines])
        if len(lines) > max_lines:
            display_content += f"\n\n... [文件共 {len(lines)} 行，显示前 {max_lines} 行]"
        
        return [TextContent(
            type="text",
            text=f"**文件**: {file_path}\n"
                 f"**大小**: {file_path.stat().st_size:,} 字节\n"
                 f"**修改时间**: {file_path.stat().st_mtime}\n"
                 f"**向量索引**: {'✅ 已索引' if indexed else '❌ 未索引'}\n"
                 f"\n```\n{display_content}\n```"
        )]
    
    raise ValueError(f"未知工具: {name}")

# ============================================================
# 启动
# ============================================================

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__":
    import asyncio
    asyncio.run(main())

八、总结：MCP 正在重塑 AI 的边界

MCP 的出现，标志着 AI 应用开发进入了一个新阶段。

从碎片化到标准化：曾经 AI 工具集成是每个团队的噩梦，每换一个模型就要重写一遍，每接一个外部服务就要写一套适配代码。MCP 把这件事变得像 USB-C 一样简单——写一次，到处能用。

从「能用」到「敢用」：MCP 的安全模型（进程隔离、RBAC、审计日志）让企业在生产环境中部署 AI 工具成为可能。AI 不再只是一个玩具，而是真正可以操作生产系统的智能助手。

从单点智能到生态智能：当所有 AI 应用和所有工具都通过同一个协议互联，形成的正反馈效应远超任何单一厂商的私有方案。工具开发者只需要维护一套代码，用户却能在所有 AI 平台上无缝使用——这是真正多赢的生态。

2026 年的今天，MCP 已经不再是一个实验性项目，而是 AI 工具生态的基础设施。无论是 AI 编程（Cursor、Cline、VS Code Agent）、AI 数据库工具、还是 AI 驱动的 DevOps 平台，MCP 都在底层默默支撑着这一切。

对于每一个想在 AI 时代保持竞争力的开发者来说，理解 MCP、掌握 MCP、用好 MCP，已经不是「锦上添花」的技能，而是必须具备的核心能力。

下一个改变 AI 世界的协议，也许已经在路上了。但在那之前，MCP 就是你现在就要掌握的连接标准。

附录：快速上手资源

官方资源：

• MCP 规范文档：https://modelcontextprotocol.io
• 官方 SDK：https://github.com/modelcontextprotocol
• MCP TypeScript SDK：npm install @modelcontextprotocol/sdk
• MCP Python SDK：pip install mcp

社区资源：

• awesome-mcp：https://github.com/xxxx
• MCP Registry：https://mcp.so（第三方 MCP Server 索引）

推荐学习路径：

1. 先在 Claude Desktop 或 Cursor 中体验官方 Server
2. 用 Python SDK 从零构建一个简单的 MCP Server
3. 尝试接入向量数据库，实现 RAG 功能
4. 在生产环境中部署，配置安全策略和监控

本文基于 MCP 协议 2026 年最新版本编写，部分 API 细节可能随版本更新而变化，请以官方文档为准。