MCP（Model Context Protocol）协议开发完全教程

一、概述与背景

1.1 什么是MCP

MCP（Model Context Protocol，模型上下文协议）是由Anthropic于2024年11月正式发布的一个开放标准协议。它的核心目标是为大语言模型（LLM）提供一种标准化的方式来连接外部数据源和工具，解决LLM与外部系统之间的"信息孤岛"问题。

在MCP出现之前，每个AI应用都需要为每个外部服务编写定制化的集成代码。如果你有N个AI应用需要连接M个外部服务，理论上你需要编写N×M个集成适配器。MCP的出现将这个复杂度降低为N+M——每个应用只需实现一次MCP Client，每个服务只需实现一次MCP Server。

1.2 为什么需要MCP

传统AI应用开发中存在以下痛点：

集成碎片化：每个LLM平台（OpenAI、Anthropic、Google等）都有各自的Function Calling规范，开发者需要为不同平台重复适配。

上下文受限：LLM的知识截止于训练数据，无法实时访问企业内部数据库、文件系统、API等动态数据源。

安全合规困难：直接让LLM访问原始数据源存在安全隐患，缺乏统一的权限控制和审计机制。

生态割裂：各厂商的插件系统互不兼容，开发者投入的适配工作无法复用。

MCP通过定义一个标准化的通信协议，让LLM能够以统一的方式访问各种外部能力，同时提供了安全、可审计的访问控制机制。

1.3 MCP与现有方案的对比

二、MCP架构设计

2.1 三层架构

MCP采用Host/Client/Server三层架构设计：

┌─────────────────────────────────┐
│           Host 应用              │
│  (Claude Desktop / IDE / 聊天)   │
│                                 │
│  ┌───────────────────────────┐  │
│  │      MCP Client           │  │
│  │  (协议客户端实例)          │  │
│  └─────┬─────────┬───────────┘  │
│        │         │              │
└────────┼─────────┼──────────────┘
         │         │
    ┌────┴───┐ ┌───┴────┐
    │MCP     │ │MCP     │
    │Server A│ │Server B│
    │(数据库)│ │(文件)  │
    └────────┘ └────────┘

Host（宿主）：用户直接交互的应用程序，如Claude Desktop、VS Code插件、自定义聊天界面。Host负责管理用户界面、控制权限、协调多个Client实例。

Client（客户端）：嵌入在Host中的协议客户端，每个Client与一个特定的Server建立一对一的连接。Client负责协议消息的序列化/反序列化、能力协商、请求路由。

Server（服务端）：提供具体能力的服务程序，如数据库查询、文件读取、API调用等。Server通过标准化接口暴露Resources、Tools、Prompts三大能力。

2.2 通信协议

MCP支持多种传输方式：

stdio（标准输入输出）：适用于本地进程间通信。Host启动Server子进程，通过stdin/stdout交换JSON-RPC消息。这是最简单、最安全的方式，适合桌面应用场景。

SSE（Server-Sent Events）：适用于远程服务通信。Server作为HTTP服务运行，Client通过SSE连接接收消息，通过HTTP POST发送请求。适合Web应用场景。

自定义传输：协议允许实现自定义传输层，如WebSocket、gRPC等，只要满足消息双向传递的基本要求。

2.3 消息格式

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

// 请求
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "query_database",
    "arguments": {
      "sql": "SELECT * FROM users LIMIT 10"
    }
  }
}

// 响应
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "[{\"id\": 1, \"name\": \"Alice\"}, ...]"
      }
    ]
  }
}

2.4 能力协商

连接建立时，Client和Server会进行能力协商：

// Client发起初始化
{
  "method": "initialize",
  "params": {
    "protocolVersion": "2024-11-05",
    "capabilities": {
      "roots": { "listChanged": true },
      "sampling": {}
    },
    "clientInfo": {
      "name": "MyApp",
      "version": "1.0.0"
    }
  }
}

// Server响应能力
{
  "result": {
    "protocolVersion": "2024-11-05",
    "capabilities": {
      "tools": { "listChanged": true },
      "resources": { "subscribe": true },
      "prompts": { "listChanged": true }
    },
    "serverInfo": {
      "name": "DatabaseServer",
      "version": "1.0.0"
    }
  }
}

三、核心概念详解

3.1 Resources（资源）

Resources是Server向Client暴露的数据源，类似于REST API中的GET端点。它们代表可以被读取的内容，如文件、数据库记录、API响应等。

每个Resource具有以下属性：

• uri：唯一标识符，如file:///path/to/file或db://table/users
• name：人类可读的名称
• description：资源描述（可选）
• mimeType：MIME类型（可选）

Resources支持两种读取方式：

• 直接读取：Client通过URI直接获取资源内容
• 动态模板：使用URI模板（如db://table/{name}）提供参数化的资源访问

3.2 Tools（工具）

Tools是Server暴露的可执行功能，类似于函数调用。Tools是LLM与外部世界交互的主要手段。

每个Tool定义包括：

• name：工具名称（在Server内唯一）
• description：功能描述，LLM根据此描述决定何时调用
• inputSchema：JSON Schema格式的输入参数定义

Tools的关键特性：

• 模型控制：Tools由LLM自主决定何时调用，而非由用户直接触发
• 类型安全：输入参数通过JSON Schema严格定义
• 幂等性建议：建议Tool实现幂等操作，避免重复调用产生副作用

3.3 Prompts（提示模板）

Prompts是Server提供的预定义提示模板，用于引导LLM完成特定任务。它们类似于"宏"或"快捷指令"。

每个Prompt定义包括：

• name：模板名称
• description：用途说明
• arguments：可选的参数列表

Prompts的典型应用场景：

• 代码审查模板：输入代码片段，输出结构化的审查意见
• 数据分析模板：输入数据描述，输出分析查询建议
• 文档生成模板：输入API定义，输出使用文档

3.4 Sampling（采样）

Sampling是MCP的一个高级特性，允许Server通过Client请求LLM的补全能力。这创造了一种"服务端驱动"的AI交互模式。

工作流程：

1. Server向Client发送sampling/createMessage请求
2. Client将请求转发给Host中的LLM
3. LLM生成响应
4. Client将响应返回给Server
5. Server继续处理

这种机制使得Server能够利用LLM的能力进行智能决策，而无需直接集成LLM API。

四、MCP Server开发实战

4.1 环境准备

使用Python SDK开发MCP Server，需要Python 3.10+环境：

pip install mcp

4.2 基础Server结构

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
    Tool,
    TextContent,
    CallToolResult,
    ListToolsResult,
)
import json

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

# 定义工具列表
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="calculate",
            description="执行数学计算，支持加减乘除",
            inputSchema={
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "数学表达式，如 '2 + 3 * 4'"
                    }
                },
                "required": ["expression"]
            }
        )
    ]

# 实现工具调用
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> CallToolResult:
    if name == "calculate":
        expression = arguments.get("expression", "")
        try:
            # 安全的数学计算
            result = eval(expression, {"__builtins__": {}}, {})
            return CallToolResult(
                content=[TextContent(
                    type="text",
                    text=f"计算结果: {expression} = {result}"
                )]
            )
        except Exception as e:
            return CallToolResult(
                content=[TextContent(
                    type="text",
                    text=f"计算错误: {str(e)}"
                )],
                isError=True
            )

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

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

4.3 添加Resources支持

from mcp.types import Resource, ReadResourceResult

@app.list_resources()
async def list_resources() -> list[Resource]:
    return [
        Resource(
            uri="config://app/settings",
            name="应用配置",
            description="获取当前应用的配置信息",
            mimeType="application/json"
        )
    ]

@app.read_resource()
async def read_resource(uri: str) -> ReadResourceResult:
    if uri == "config://app/settings":
        config = {
            "app_name": "MyApp",
            "version": "1.0.0",
            "debug": False
        }
        return ReadResourceResult(
            contents=[{
                "uri": uri,
                "mimeType": "application/json",
                "text": json.dumps(config, ensure_ascii=False, indent=2)
            }]
        )
    raise ValueError(f"Unknown resource: {uri}")

4.4 添加Prompts支持

from mcp.types import Prompt, GetPromptResult, PromptMessage

@app.list_prompts()
async def list_prompts() -> list[Prompt]:
    return [
        Prompt(
            name="code_review",
            description="代码审查模板",
            arguments=[
                {
                    "name": "code",
                    "description": "待审查的代码",
                    "required": True
                },
                {
                    "name": "language",
                    "description": "编程语言",
                    "required": False
                }
            ]
        )
    ]

@app.get_prompt()
async def get_prompt(name: str, arguments: dict) -> GetPromptResult:
    if name == "code_review":
        code = arguments.get("code", "")
        language = arguments.get("language", "Python")
        return GetPromptResult(
            messages=[
                PromptMessage(
                    role="user",
                    content=TextContent(
                        type="text",
                        text=f"请审查以下{language}代码，"
                             f"指出潜在问题和改进建议：\n\n"
                             f"```{language}\n{code}\n```"
                    )
                )
            ]
        )

五、MCP Client开发实战

5.1 基础Client实现

from mcp.client import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client
import json

async def run_client():
    # 配置Server连接参数
    server_params = StdioServerParameters(
        command="python",
        args=["my_server.py"]
    )

    # 建立连接
    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(
                "calculate",
                arguments={"expression": "2 + 3 * 4"}
            )
            print(f"调用结果: {result.content[0].text}")

            # 读取资源
            resources = await session.list_resources()
            for resource in resources.resources:
                content = await session.read_resource(resource.uri)
                print(f"资源 {resource.uri}: {content}")

if __name__ == "__main__":
    import asyncio
    asyncio.run(run_client())

5.2 动态Server发现

import subprocess
import json

class MCPClientManager:
    def __init__(self):
        self.servers = {}
        self.sessions = {}

    async def add_server(self, name: str, command: str, args: list):
        """动态添加并连接Server"""
        server_params = StdioServerParameters(
            command=command,
            args=args
        )
        read, write = await stdio_client(server_params)
        session = ClientSession(read, write)
        await session.initialize()

        self.servers[name] = server_params
        self.sessions[name] = session

        # 获取Server能力
        tools = await session.list_tools()
        print(f"Server '{name}' 已连接，提供 {len(tools.tools)} 个工具")
        return session

    async def call_tool(self, server_name: str, tool_name: str, arguments: dict):
        """调用指定Server的工具"""
        session = self.sessions.get(server_name)
        if not session:
            raise ValueError(f"Server '{server_name}' 未连接")
        return await session.call_tool(tool_name, arguments)

    async def find_tool(self, tool_name: str):
        """跨Server查找工具"""
        for name, session in self.sessions.items():
            tools = await session.list_tools()
            for tool in tools.tools:
                if tool.name == tool_name:
                    return name, tool
        return None, None

六、与Claude Desktop集成

6.1 配置文件

Claude Desktop通过配置文件管理MCP Server连接，配置文件位于：

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

{
  "mcpServers": {
    "my-database": {
      "command": "python",
      "args": ["/path/to/database_server.py"],
      "env": {
        "DB_HOST": "localhost",
        "DB_PORT": "5432"
      }
    },
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "/Users/me/documents"]
    }
  }
}

6.2 调试技巧

使用MCP Inspector进行调试：

npx @modelcontextprotocol/inspector python my_server.py

Inspector提供了一个Web界面，可以：

• 查看Server暴露的所有能力
• 手动调用Tools并查看结果
• 检查Resources内容
• 测试Prompts模板
• 查看通信日志

七、与LangChain/LlamaIndex集成

7.1 LangChain集成

from langchain_mcp import MCPToolkit
from langchain.agents import create_react_agent, AgentExecutor
from langchain_openai import ChatOpenAI

async def create_mcp_agent():
    # 连接MCP Server
    toolkit = await MCPToolkit.from_server(
        command="python",
        args=["my_mcp_server.py"]
    )

    # 获取工具
    tools = toolkit.get_tools()

    # 创建Agent
    llm = ChatOpenAI(model="gpt-4")
    agent = create_react_agent(llm, tools, prompt_template)
    executor = AgentExecutor(agent=agent, tools=tools)

    # 执行查询
    result = await executor.ainvoke({
        "input": "查询数据库中的用户数量"
    })
    return result

7.2 LlamaIndex集成

from llama_index.core.tools import FunctionTool
from llama_index.agent.openai import OpenAIAgent
import mcp

async def create_llama_agent():
    # 连接MCP Server
    session = await connect_mcp_server("python", ["my_server.py"])

    # 获取工具列表
    tools_result = await session.list_tools()

    # 转换为LlamaIndex工具
    llama_tools = []
    for tool in tools_result.tools:
        async def call_mcp(name=tool.name, **kwargs):
            result = await session.call_tool(name, kwargs)
            return result.content[0].text

        llama_tool = FunctionTool.from_defaults(
            fn=call_mcp,
            name=tool.name,
            description=tool.description
        )
        llama_tools.append(llama_tool)

    # 创建Agent
    agent = OpenAIAgent.from_tools(llama_tools, verbose=True)
    response = await agent.achat("帮我查询最近的销售数据")
    return response

八、企业级MCP部署方案

8.1 部署架构

企业环境中，MCP Server通常需要以下部署组件：

┌─────────────────────────────────────────┐
│              负载均衡器                   │
└──────────┬──────────────┬───────────────┘
           │              │
    ┌──────┴──┐    ┌──────┴──┐
    │MCP网关A │    │MCP网关B │
    └────┬────┘    └────┬────┘
         │              │
    ┌────┴──────────────┴────┐
    │      MCP Server集群     │
    │  ┌────┐ ┌────┐ ┌────┐  │
    │  │ S1 │ │ S2 │ │ S3 │  │
    │  └────┘ └────┘ └────┘  │
    └─────────┬──────────────┘
              │
    ┌─────────┴──────────────┐
    │    共享存储/数据库       │
    └────────────────────────┘

8.2 使用SSE传输的远程Server

from mcp.server import Server
from mcp.server.sse import SseServerTransport
from starlette.applications import Starlette
from starlette.routing import Route, Mount

app = Server("remote-server")

# ... 定义tools, resources, prompts ...

# 创建SSE传输
sse_transport = SseServerTransport("/messages")

async def handle_sse(request):
    async with sse_transport.connect_sse(
        request.scope, request.receive, request._send
    ) as streams:
        await app.run(streams[0], streams[1])

async def handle_messages(request):
    await sse_transport.handle_post_message(
        request.scope, request.receive, request._send
    )

# 创建Starlette应用
starlette_app = Starlette(
    routes=[
        Route("/sse", endpoint=handle_sse),
        Route("/messages", endpoint=handle_messages, methods=["POST"]),
    ]
)

# 启动: uvicorn server_app:starlette_app --host 0.0.0.0 --port 8080

8.3 容器化部署

FROM python:3.11-slim

WORKDIR /app

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

COPY . .

EXPOSE 8080

CMD ["uvicorn", "server_app:starlette_app", "--host", "0.0.0.0", "--port", "8080"]

# docker-compose.yml
version: '3.8'
services:
  mcp-server:
    build: .
    ports:
      - "8080:8080"
    environment:
      - DB_HOST=postgres
      - REDIS_URL=redis://redis:6379
    depends_on:
      - postgres
      - redis
    deploy:
      replicas: 3
      resources:
        limits:
          memory: 512M
          cpus: '0.5'

  postgres:
    image: postgres:15
    environment:
      POSTGRES_DB: mcp
      POSTGRES_PASSWORD: secret
    volumes:
      - pgdata:/var/lib/postgresql/data

  redis:
    image: redis:7-alpine

volumes:
  pgdata:

九、安全与权限管理

9.1 认证机制

MCP协议本身不定义认证方式，但可以在传输层实施：

from mcp.server import Server
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware

class TokenAuthBackend:
    async def authenticate(self, conn):
        token = conn.headers.get("Authorization", "").replace("Bearer ", "")
        if not token:
            return None
        # 验证token
        user = await verify_token(token)
        if user:
            return AuthCredentials(["authenticated"]), user
        return None

# 添加认证中间件
starlette_app = Starlette(
    routes=[...],
    middleware=[
        Middleware(AuthenticationMiddleware, backend=TokenAuthBackend())
    ]
)

9.2 细粒度权限控制

class PermissionManager:
    def __init__(self):
        self.permissions = {
            "admin": {
                "tools": ["*"],
                "resources": ["*"],
                "rate_limit": 1000
            },
            "user": {
                "tools": ["query_database", "read_file"],
                "resources": ["public://*"],
                "rate_limit": 100
            }
        }

    def check_tool_permission(self, role: str, tool_name: str) -> bool:
        allowed = self.permissions.get(role, {}).get("tools", [])
        return "*" in allowed or tool_name in allowed

    def check_resource_permission(self, role: str, uri: str) -> bool:
        allowed = self.permissions.get(role, {}).get("resources", [])
        return any(
            pattern.replace("*", "") in uri
            for pattern in allowed
        )

perm_manager = PermissionManager()

@app.call_tool()
async def call_tool(name: str, arguments: dict, context=None):
    role = get_user_role(context)
    if not perm_manager.check_tool_permission(role, name):
        return CallToolResult(
            content=[TextContent(type="text", text="权限不足")],
            isError=True
        )
    # 执行工具逻辑...

9.3 审计日志

import logging
from datetime import datetime

audit_logger = logging.getLogger("mcp.audit")

def log_tool_call(user: str, tool: str, args: dict, result: str, success: bool):
    audit_logger.info(json.dumps({
        "timestamp": datetime.utcnow().isoformat(),
        "user": user,
        "action": "tool_call",
        "tool": tool,
        "arguments": args,
        "result_summary": result[:200],
        "success": success
    }, ensure_ascii=False))

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    user = get_current_user()
    try:
        result = await execute_tool(name, arguments)
        log_tool_call(user, name, arguments, str(result), True)
        return result
    except Exception as e:
        log_tool_call(user, name, arguments, str(e), False)
        raise

十、实战案例

10.1 构建数据库查询MCP Server

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent, CallToolResult
import sqlite3
import json

app = Server("database-server")

# 数据库连接池
def get_db():
    return sqlite3.connect("app.db")

@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="query_sql",
            description="执行SQL查询（仅SELECT语句）",
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {
                        "type": "string",
                        "description": "SQL查询语句"
                    },
                    "limit": {
                        "type": "integer",
                        "description": "最大返回行数",
                        "default": 100
                    }
                },
                "required": ["sql"]
            }
        ),
        Tool(
            name="list_tables",
            description="列出数据库中的所有表",
            inputSchema={"type": "object", "properties": {}}
        ),
        Tool(
            name="describe_table",
            description="查看表结构",
            inputSchema={
                "type": "object",
                "properties": {
                    "table_name": {
                        "type": "string",
                        "description": "表名"
                    }
                },
                "required": ["table_name"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    db = get_db()
    cursor = db.cursor()

    try:
        if name == "query_sql":
            sql = arguments["sql"].strip()
            # 安全检查：只允许SELECT
            if not sql.upper().startswith("SELECT"):
                return CallToolResult(
                    content=[TextContent(
                        type="text",
                        text="安全限制：只允许执行SELECT查询"
                    )],
                    isError=True
                )

            limit = arguments.get("limit", 100)
            if "LIMIT" not in sql.upper():
                sql += f" LIMIT {limit}"

            cursor.execute(sql)
            columns = [desc[0] for desc in cursor.description]
            rows = cursor.fetchall()
            result = [dict(zip(columns, row)) for row in rows]

            return CallToolResult(
                content=[TextContent(
                    type="text",
                    text=json.dumps(result, ensure_ascii=False, indent=2)
                )]
            )

        elif name == "list_tables":
            cursor.execute(
                "SELECT name FROM sqlite_master WHERE type='table'"
            )
            tables = [row[0] for row in cursor.fetchall()]
            return CallToolResult(
                content=[TextContent(
                    type="text",
                    text=json.dumps(tables, ensure_ascii=False)
                )]
            )

        elif name == "describe_table":
            table = arguments["table_name"]
            cursor.execute(f"PRAGMA table_info({table})")
            columns = cursor.fetchall()
            schema = [
                {"name": col[1], "type": col[2], "notnull": col[3]}
                for col in columns
            ]
            return CallToolResult(
                content=[TextContent(
                    type="text",
                    text=json.dumps(schema, ensure_ascii=False, indent=2)
                )]
            )
    finally:
        db.close()

10.2 构建文件系统MCP Server

from mcp.server import Server
from mcp.types import Tool, Resource, TextContent, CallToolResult
import os
from pathlib import Path

app = Server("filesystem-server")

# 限制访问的根目录
ROOT_DIR = Path("/safe/workspace")

def safe_path(requested: str) -> Path:
    """确保路径在安全范围内"""
    resolved = (ROOT_DIR / requested).resolve()
    if not str(resolved).startswith(str(ROOT_DIR.resolve())):
        raise ValueError("路径越界访问")
    return resolved

@app.list_tools()
async def list_tools():
    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": "目录路径",
                        "default": "."
                    }
                }
            }
        ),
        Tool(
            name="search_files",
            description="按文件名搜索文件",
            inputSchema={
                "type": "object",
                "properties": {
                    "pattern": {
                        "type": "string",
                        "description": "搜索模式（支持通配符）"
                    }
                },
                "required": ["pattern"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    try:
        if name == "read_file":
            path = safe_path(arguments["path"])
            content = path.read_text(encoding="utf-8")
            return CallToolResult(
                content=[TextContent(type="text", text=content)]
            )

        elif name == "write_file":
            path = safe_path(arguments["path"])
            path.parent.mkdir(parents=True, exist_ok=True)
            path.write_text(arguments["content"], encoding="utf-8")
            return CallToolResult(
                content=[TextContent(
                    type="text",
                    text=f"文件已写入: {path}"
                )]
            )

        elif name == "list_directory":
            path = safe_path(arguments.get("path", "."))
            entries = []
            for item in sorted(path.iterdir()):
                entries.append({
                    "name": item.name,
                    "type": "dir" if item.is_dir() else "file",
                    "size": item.stat().st_size if item.is_file() else None
                })
            return CallToolResult(
                content=[TextContent(
                    type="text",
                    text=json.dumps(entries, ensure_ascii=False, indent=2)
                )]
            )

        elif name == "search_files":
            pattern = arguments["pattern"]
            matches = list(ROOT_DIR.rglob(pattern))
            results = [
                str(p.relative_to(ROOT_DIR))
                for p in matches[:50]  # 限制结果数量
            ]
            return CallToolResult(
                content=[TextContent(
                    type="text",
                    text=json.dumps(results, ensure_ascii=False, indent=2)
                )]
            )
    except Exception as e:
        return CallToolResult(
            content=[TextContent(type="text", text=f"错误: {str(e)}")],
            isError=True
        )

十一、最佳实践

11.1 Server设计原则

单一职责：每个Server专注于一类能力。数据库Server只处理数据查询，文件Server只处理文件操作。避免创建"万能Server"。

清晰的描述：Tool的description要足够详细，让LLM能准确判断何时使用。例如不要写"查询数据"，而要写"从用户表中按条件查询用户信息，支持按姓名、邮箱、注册时间筛选"。

输入验证：严格验证所有输入参数，使用JSON Schema定义约束。永远不要信任来自Client的输入。

错误处理：返回清晰的错误信息，帮助LLM理解问题所在并尝试修正。使用isError标记区分正常结果和错误结果。

11.2 性能优化

import asyncio
from functools import lru_cache

# 缓存频繁访问的资源
@lru_cache(maxsize=100)
def get_cached_resource(uri: str):
    return load_resource(uri)

# 批量操作
@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "batch_query":
        queries = arguments["queries"]
        # 并行执行多个查询
        tasks = [execute_query(q) for q in queries]
        results = await asyncio.gather(*tasks)
        return CallToolResult(
            content=[TextContent(
                type="text",
                text=json.dumps(results)
            )]
        )

# 流式大文件
@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "read_large_file":
        path = arguments["path"]
        chunks = []
        async with aiofiles.open(path, 'r') as f:
            async for chunk in f:
                chunks.append(chunk)
                if len(chunks) >= 100:
                    yield chunks
                    chunks = []

11.3 日志与监控

import logging
from prometheus_client import Counter, Histogram

# Prometheus指标
tool_calls = Counter(
    'mcp_tool_calls_total',
    'Total tool calls',
    ['tool_name', 'status']
)
tool_duration = Histogram(
    'mcp_tool_duration_seconds',
    'Tool execution duration',
    ['tool_name']
)

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    with tool_duration.labels(tool_name=name).time():
        try:
            result = await execute_tool(name, arguments)
            tool_calls.labels(tool_name=name, status='success').inc()
            return result
        except Exception as e:
            tool_calls.labels(tool_name=name, status='error').inc()
            raise

十二、常见问题

Q1: MCP Server连接失败怎么办？

检查以下几点：

1. Server进程是否正常启动
2. 配置文件中的命令和参数是否正确
3. 环境变量是否正确设置
4. 查看Claude Desktop日志：~/Library/Logs/Claude/ (macOS)

Q2: 如何处理长时间运行的工具？

对于耗时操作，建议：

1. 返回进度信息，让用户知道正在处理
2. 实现取消机制，支持中断长时间任务
3. 考虑异步模式，先返回任务ID，再轮询结果

Q3: MCP支持多语言吗？

MCP协议本身是语言无关的。目前官方提供TypeScript和Python SDK，社区也有Go、Rust等语言的实现。只要能实现JSON-RPC通信，任何语言都可以开发MCP Server。

Q4: 如何测试MCP Server？

推荐测试策略：

1. 使用MCP Inspector进行手动测试
2. 编写单元测试覆盖每个Tool的逻辑
3. 使用集成测试验证Client-Server通信
4. 使用Mock LLM测试完整工作流

Q5: MCP与OpenAI Function Calling可以共存吗？

可以。MCP和Function Calling是不同层面的概念。你可以在MCP Server中封装Function Calling的逻辑，也可以在使用OpenAI API的应用中通过MCP Client访问MCP Server。两者可以互为补充。

十三、总结

MCP协议为AI应用与外部系统的集成提供了一个标准化的解决方案。通过Host/Client/Server三层架构，它实现了：

• 标准化：统一的协议规范，消除碎片化
• 安全性：细粒度权限控制和审计能力
• 可扩展：支持多种传输方式和自定义扩展
• 互操作：跨平台、跨语言的兼容性

随着AI应用的普及，MCP有望成为连接LLM与外部世界的标准桥梁。掌握MCP开发，将使你能够构建更强大、更安全、更易集成的AI应用。

推荐学习资源

• MCP官方规范文档
• MCP TypeScript SDK GitHub仓库
• MCP Python SDK GitHub仓库
• Claude Desktop MCP配置指南
• MCP Inspector调试工具

本教程基于MCP协议2024-11-05版本编写，协议仍在持续演进中，请关注官方更新。