第三十三章：MCP模型上下文协议

一、MCP简介：为什么需要模型上下文协议

在大语言模型快速发展的今天，每个AI应用都需要与外部数据源、工具和服务进行交互。然而，如果没有统一的标准，每个集成都需要从头开发，导致了大量重复劳动和碎片化问题。MCP（Model Context Protocol）正是为了解决这个问题而诞生的开放标准协议。

MCP与函数调用的区别

很多开发者会混淆MCP与LLM的Function Calling。实际上，Function Calling是LLM内部的能力，而MCP是外部的标准化通信协议。两者是互补的关系，而非替代关系。

维度	Function Calling	MCP
性质	LLM内部能力	外部标准化协议
定义位置	应用代码中	独立的MCP Server
复用性	每个应用独立实现	跨应用复用
协议规范	各厂商各自定义	统一标准（JSON-RPC 2.0）
安全边界	由应用控制	协议层内置权限控制

二、架构设计：Host、Client、Server

MCP采用分层架构，包含三个核心角色：Host（宿主）、Client（客户端）和Server（服务端）。这种分层设计使得协议既灵活又安全。

一个Host可以同时连接多个Client，每个Client对应一个Server。这种设计使得Host能够聚合来自不同Server的能力，同时保持清晰的权限边界。

连接生命周期

MCP Client与Server之间的连接遵循标准的生命周期管理流程：

这种设计确保了连接的可靠性和可预测性，使得错误处理和资源清理变得更加容易。

消息流动模式

MCP中的消息流动遵循严格的请求-响应和通知两种模式。理解这些模式对于正确实现协议至关重要。

JSON - 通知示例

// Server → Client: 进度通知
{
  "jsonrpc": "2.0",
  "method": "notifications/progress",
  "params": {
    "progressToken": "abc-123",
    "progress": 50,
    "total": 100,
    "message": "正在处理第50条记录..."
  }
}

// Server → Client: 资源变更通知
{
  "jsonrpc": "2.0",
  "method": "notifications/resources/updated",
  "params": {
    "uri": "file:///data/report.csv"
  }
}

多Server聚合模式

在实际应用中，一个Host可能需要连接多个Server来聚合不同的能力。MCP的架构设计天然支持这种多Server聚合模式。

Python - 多Server管理

"""
多Server管理示例
展示如何在Host中管理多个MCP Server
"""
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

class MultiServerHost:
    def __init__(self):
        self.sessions: dict[str, ClientSession] = {}

    async def add_server(self, name: str, params: StdioServerParameters):
        """添加并连接一个新的MCP Server"""
        read, write = await stdio_client(params).__aenter__()
        session = ClientSession(read, write)
        await session.__aenter__()
        await session.initialize()

        self.sessions[name] = session
        print(f"已连接到Server: {name}")

        # 列出该Server提供的能力
        tools = await session.list_tools()
        print(f"  工具: {[t.name for t in tools.tools]}")

        resources = await session.list_resources()
        print(f"  资源: {[r.uri for r in resources.resources]}")

    async def call_tool(self, server_name: str, tool_name: str, args: dict):
        """调用指定Server上的工具"""
        if server_name not in self.sessions:
            raise ValueError(f"Server未连接: {server_name}")

        session = self.sessions[server_name]
        result = await session.call_tool(tool_name, args)
        return result.content[0].text

    async def search_tools(self, query: str):
        """在所有Server中搜索工具"""
        results = []
        for name, session in self.sessions.items():
            tools = await session.list_tools()
            for tool in tools.tools:
                if query.lower() in tool.name.lower() or \
                   query.lower() in tool.description.lower():
                    results.append({
                        "server": name,
                        "tool": tool.name,
                        "description": tool.description
                    })
        return results

    async def disconnect_all(self):
        """断开所有Server连接"""
        for name, session in self.sessions.items():
            await session.close()
            print(f"已断开Server: {name}")
        self.sessions.clear()

async def main():
    host = MultiServerHost()

    # 连接文件系统Server
    await host.add_server("filesystem", StdioServerParameters(
        command="python",
        args=["filesystem_server.py"]
    ))

    # 连接数据库Server
    await host.add_server("database", StdioServerParameters(
        command="python",
        args=["database_server.py"]
    ))

    # 搜索所有包含"read"的工具
    read_tools = await host.search_tools("read")
    print(f"\n包含'read'的工具:")
    for t in read_tools:
        print(f"  [{t['server']}] {t['tool']}: {t['description']}")

    # 调用文件系统Server的read_file工具
    content = await host.call_tool(
        "filesystem", "read_file",
        {"path": "config.json"}
    )
    print(f"\n文件内容:\n{content}")

    await host.disconnect_all()

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

三、核心概念：四大原语

MCP定义了四种核心原语（Primitives），它们是Server向Client暴露能力的基本方式。理解这四种原语是掌握MCP的关键。

3.1 Resources：数据资源

Resources代表Server可以提供的数据内容，类似于REST API中的GET端点。Resources是只读的，由应用程序或用户控制是否包含在LLM上下文中。

Python

from mcp.server import Server
from mcp.types import Resource, ResourceContents

server = Server("my-server")

# 定义一个文件资源
@server.resource("file:///logs/{date}")
async def get_log_file(date: str) -> ResourceContents:
    """读取指定日期的日志文件"""
    file_path = f"/var/log/app/{date}.log"
    content = await read_file(file_path)
    return ResourceContents(
        uri=f"file:///logs/{date}",
        mimeType="text/plain",
        text=content
    )

# 定义一个动态资源
@server.resource("db://users/{user_id}")
async def get_user(user_id: str) -> ResourceContents:
    """获取用户信息"""
    user = await db.get_user(user_id)
    return ResourceContents(
        uri=f"db://users/{user_id}",
        mimeType="application/json",
        text=json.dumps(user)
    )

3.2 Tools：可执行工具

Tools代表Server可以执行的操作，由模型控制（LLM决定何时调用）。Tools类似于函数，接受输入参数并返回执行结果。与Resources不同，Tools是有副作用的。

Python

from mcp.server import Server
from mcp.types import Tool, ToolResult
from pydantic import BaseModel

server = Server("my-server")

# 定义工具的输入参数
class SendEmailArgs(BaseModel):
    to: str
    subject: str
    body: str

# 注册邮件发送工具
@server.tool()
async def send_email(args: SendEmailArgs) -> ToolResult:
    """发送邮件到指定地址"""
    await email_service.send(
        to=args.to,
        subject=args.subject,
        body=args.body
    )
    return ToolResult(
        content=f"邮件已发送至 {args.to}",
        isError=False
    )

# 注册数据库查询工具
@server.tool()
async def query_database(sql: str) -> ToolResult:
    """执行SQL查询"""
    try:
        results = await db.execute(sql)
        return ToolResult(
            content=json.dumps(results),
            isError=False
        )
    except Exception as e:
        return ToolResult(
            content=f"查询失败: {str(e)}",
            isError=True
        )

3.3 Prompts：提示模板

Prompts是Server提供的预定义提示模板，由用户控制（用户选择是否使用）。Prompts可以将上下文和指令封装为可复用的模板，简化复杂的交互流程。

Python

from mcp.server import Server
from mcp.types import Prompt, PromptMessage, Role

server = Server("my-server")

@server.prompt()
async def code_review_prompt(
    repository: str,
    pull_request: int
) -> list[PromptMessage]:
    """代码审查提示模板"""
    # 获取PR信息
    pr_info = await github.get_pr(repository, pull_request)
    diff = await github.get_diff(repository, pull_request)

    return [
        PromptMessage(
            role=Role.USER,
            content=f"请审查以下Pull Request:\n\n"
                    f"标题: {pr_info['title']}\n"
                    f"描述: {pr_info['body']}\n\n"
                    f"代码变更:\n{diff}"
        )
    ]

3.4 Sampling：模型采样

Sampling是MCP最独特的原语，它允许Server请求LLM进行文本生成。这使得Server可以利用AI能力完成复杂任务，同时Host保持对采样过程的完全控制。

四、协议规范：JSON-RPC 2.0

MCP基于JSON-RPC 2.0构建，这是一种轻量级的远程过程调用协议。JSON-RPC使用JSON作为数据格式，支持请求-响应和通知两种消息模式。

初始化握手

连接建立时，Client和Server通过初始化消息交换能力信息：

JSON

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

// Server → Client: 初始化响应
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "protocolVersion": "2024-11-05",
    "capabilities": {
      "resources": { "subscribe": true },
      "tools": { "listChanged": true },
      "prompts": {}
    },
    "serverInfo": {
      "name": "DatabaseServer",
      "version": "1.0.0"
    }
  }
}

能力发现

初始化完成后，Client可以通过列表请求发现Server提供的所有能力：

JSON

// Client → Server: 列出所有工具
{
  "jsonrpc": "2.0",
  "id": 2,
  "method": "tools/list",
  "params": {}
}

// Server → Client: 返回工具列表
{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "tools": [
      {
        "name": "query_database",
        "description": "执行SQL查询",
        "inputSchema": {
          "type": "object",
          "properties": {
            "sql": {
              "type": "string",
              "description": "SQL查询语句"
            }
          },
          "required": ["sql"]
        }
      },
      {
        "name": "send_email",
        "description": "发送邮件",
        "inputSchema": {
          "type": "object",
          "properties": {
            "to": { "type": "string" },
            "subject": { "type": "string" },
            "body": { "type": "string" }
          },
          "required": ["to", "subject", "body"]
        }
      }
    ]
  }
}

错误处理机制

MCP定义了一套标准的错误码体系，用于处理各种异常情况。正确的错误处理对于构建健壮的MCP应用至关重要。

JSON - 错误响应示例

// 参数验证错误
{
  "jsonrpc": "2.0",
  "id": 3,
  "error": {
    "code": -32602,
    "message": "Invalid parameters",
    "data": {
      "field": "path",
      "reason": "Path must be a non-empty string"
    }
  }
}

// 工具执行错误
{
  "jsonrpc": "2.0",
  "id": 4,
  "result": {
    "content": [{
      "type": "text",
      "text": "文件不存在: /nonexistent/file.txt"
    }],
    "isError": true
  }
}

取消与超时

在实际使用中，某些操作可能耗时较长或被用户主动取消。MCP提供了标准化的取消机制来处理这些场景。

JSON - 取消操作

// Client → Server: 取消进行中的请求
{
  "jsonrpc": "2.0",
  "method": "notifications/cancelled",
  "params": {
    "requestId": 5,
    "reason": "用户取消操作"
  }
}

超时是另一个重要的考虑因素。Client应该为每个请求设置合理的超时时间，避免无限等待。MCP规范建议使用30秒作为默认超时时间，对于长时间运行的操作（如文件上传），可以通过进度通知来延长超时。

五、传输机制

MCP的传输层负责消息的序列化和通信。协议本身与传输方式解耦，当前支持两种主要的传输机制：标准输入/输出（stdio）和HTTP Server-Sent Events（SSE）。

5.1 标准输入/输出（stdio）

stdio是最简单的传输方式，适用于本地进程间通信。Server作为子进程启动，通过stdin/stdout交换JSON消息。这种方式无需网络配置，安全性由操作系统进程隔离保证。

Python - stdio传输

import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server

server = Server("local-tools")

@server.tool()
async def read_file(path: str) -> str:
    """读取本地文件"""
    with open(path, 'r') as f:
        return f.read()

async def main():
    # 启动stdio传输
    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())

5.2 HTTP SSE（Server-Sent Events）

SSE传输适用于远程Server部署场景。Server作为HTTP服务运行，Client通过POST请求发送消息，Server通过SSE流返回响应和通知。这种方式支持多Client并发连接，适合生产环境部署。

Python - SSE传输

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

server = Server("remote-server")

@server.resource("config://app")
async def get_config() -> str:
    """获取应用配置"""
    return json.dumps({"version": "1.0", "debug": False})

# SSE传输实例
sse = SseServerTransport("/messages/")

async def handle_sse(request):
    """处理SSE连接"""
    async with sse.connect_sse(
        request.scope, request.receive, request._send
    ) as streams:
        await server.run(
            streams[0], streams[1],
            server.create_initialization_options()
        )

# 创建Starlette应用
app = Starlette(
    routes=[
        Route("/sse", endpoint=handle_sse),
        Mount("/messages/", app=sse.handle_post_message),
    ]
)

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

传输方式	适用场景	连接方式	部署方式
stdio	本地集成、CLI工具	进程stdin/stdout	作为子进程启动
SSE	远程服务、生产环境	HTTP POST + SSE	独立HTTP服务

六、服务端开发实战

本节通过完整示例展示如何开发一个实用的MCP Server。我们将构建一个文件系统MCP Server，它能够读写文件、列出目录并执行基本的文件操作。

6.1 Python实现

以下是使用Python MCP SDK实现的文件系统Server完整代码：

Python - 文件系统MCP Server

"""
文件系统MCP Server
提供文件读写、目录列表等功能
"""
import os
import json
import asyncio
from pathlib import Path
from typing import Optional
from mcp.server import Server
from mcp.types import (
    Tool, Resource, ResourceContents,
    ToolResult, Prompt, PromptMessage, Role
)
from pydantic import BaseModel

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

# 工具参数模型
class ReadFileArgs(BaseModel):
    path: str

class WriteFileArgs(BaseModel):
    path: str
    content: str

class ListDirArgs(BaseModel):
    path: str = "."

class DeleteFileArgs(BaseModel):
    path: str

# ==================== Tools ====================

@server.tool()
async def read_file(args: ReadFileArgs) -> ToolResult:
    """读取文件内容"""
    try:
        file_path = Path(args.path).resolve()
        if not file_path.exists():
            return ToolResult(
                content=f"文件不存在: {args.path}",
                isError=True
            )
        content = file_path.read_text(encoding='utf-8')
        return ToolResult(content=content)
    except Exception as e:
        return ToolResult(
            content=f"读取失败: {str(e)}",
            isError=True
        )

@server.tool()
async def write_file(args: WriteFileArgs) -> ToolResult:
    """写入文件内容"""
    try:
        file_path = Path(args.path).resolve()
        file_path.parent.mkdir(parents=True, exist_ok=True)
        file_path.write_text(args.content, encoding='utf-8')
        return ToolResult(
            content=f"文件已写入: {args.path} ({len(args.content)} 字节)"
        )
    except Exception as e:
        return ToolResult(
            content=f"写入失败: {str(e)}",
            isError=True
        )

@server.tool()
async def list_directory(args: ListDirArgs) -> ToolResult:
    """列出目录内容"""
    try:
        dir_path = Path(args.path).resolve()
        if not dir_path.is_dir():
            return ToolResult(
                content=f"目录不存在: {args.path}",
                isError=True
            )
        items = []
        for item in sorted(dir_path.iterdir()):
            item_type = "📁" if item.is_dir() else "📄"
            size = item.stat().st_size if item.is_file() else 0
            items.append(f"{item_type} {item.name} ({size} bytes)")

        result = f"目录: {dir_path}\n\n" + "\n".join(items)
        return ToolResult(content=result)
    except Exception as e:
        return ToolResult(
            content=f"列表失败: {str(e)}",
            isError=True
        )

@server.tool()
async def delete_file(args: DeleteFileArgs) -> ToolResult:
    """删除指定文件"""
    try:
        file_path = Path(args.path).resolve()
        if not file_path.exists():
            return ToolResult(
                content=f"文件不存在: {args.path}",
                isError=True
            )
        if file_path.is_dir():
            return ToolResult(
                content="不能删除目录，请使用rm -r",
                isError=True
            )
        file_path.unlink()
        return ToolResult(content=f"文件已删除: {args.path}")
    except Exception as e:
        return ToolResult(
            content=f"删除失败: {str(e)}",
            isError=True
        )

# ==================== Resources ====================

@server.resource("file://{path}")
async def get_file_resource(path: str) -> ResourceContents:
    """获取文件资源"""
    file_path = Path(path).resolve()
    content = file_path.read_text(encoding='utf-8')
    mime_type = "text/plain"
    if path.endswith('.json'):
        mime_type = "application/json"
    elif path.endswith('.md'):
        mime_type = "text/markdown"

    return ResourceContents(
        uri=f"file://{path}",
        mimeType=mime_type,
        text=content
    )

# ==================== Prompts ====================

@server.prompt()
async def summarize_file_prompt(path: str) -> list[PromptMessage]:
    """生成文件摘要的提示模板"""
    file_content = await read_file(ReadFileArgs(path=path))
    return [
        PromptMessage(
            role=Role.USER,
            content=f"请为以下文件内容生成简洁的摘要:\n\n```\n{file_content.content}\n```"
        )
    ]

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

async def main():
    from mcp.server.stdio import stdio_server
    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())

6.2 TypeScript实现

TypeScript版本的实现同样简洁，使用官方@modelcontextprotocol/sdk包：

TypeScript - 文件系统MCP Server

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
import * as fs from "fs/promises";
import * as path from "path";

// 创建MCP Server
const server = new McpServer({
  name: "filesystem-server",
  version: "1.0.0",
});

// 注册读取文件工具
server.tool(
  "read_file",
  "读取文件内容",
  { path: z.string().describe("文件路径") },
  async ({ path: filePath }) => {
    try {
      const content = await fs.readFile(filePath, "utf-8");
      return {
        content: [{ type: "text", text: content }],
      };
    } catch (error) {
      return {
        content: [{ type: "text", text: `读取失败: ${error.message}` }],
        isError: true,
      };
    }
  }
);

// 注册写入文件工具
server.tool(
  "write_file",
  "写入文件内容",
  {
    path: z.string().describe("文件路径"),
    content: z.string().describe("文件内容"),
  },
  async ({ path: filePath, content }) => {
    try {
      await fs.mkdir(path.dirname(filePath), { recursive: true });
      await fs.writeFile(filePath, content, "utf-8");
      return {
        content: [{ type: "text", text: `文件已写入: ${filePath}` }],
      };
    } catch (error) {
      return {
        content: [{ type: "text", text: `写入失败: ${error.message}` }],
        isError: true,
      };
    }
  }
);

// 注册列出目录工具
server.tool(
  "list_directory",
  "列出目录内容",
  { path: z.string().default(".").describe("目录路径") },
  async ({ path: dirPath }) => {
    try {
      const items = await fs.readdir(dirPath, { withFileTypes: true });
      const result = items
        .map((item) => {
          const icon = item.isDirectory() ? "📁" : "📄";
          return `${icon} ${item.name}`;
        })
        .join("\n");
      return {
        content: [{ type: "text", text: `目录: ${dirPath}\n\n${result}` }],
      };
    } catch (error) {
      return {
        content: [{ type: "text", text: `列表失败: ${error.message}` }],
        isError: true,
      };
    }
  }
);

// 注册资源
server.resource("config", "config://app", async (uri) => ({
  contents: [
    {
      uri: uri.href,
      mimeType: "application/json",
      text: JSON.stringify({ version: "1.0.0", debug: false }),
    },
  ],
}));

// 启动stdio传输
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("Filesystem MCP Server 运行中...");
}

main().catch(console.error);

七、客户端集成

MCP客户端负责连接Server并将其能力集成到AI应用中。本节展示如何在Python应用中集成MCP客户端。

Python - MCP客户端

"""
MCP客户端示例
连接MCP Server并调用其工具和资源
"""
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def main():
    # 配置Server连接参数
    server_params = StdioServerParameters(
        command="python",
        args=["filesystem_server.py"],
        env=None  # 可选环境变量
    )

    # 建立连接
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # 初始化连接
            await session.initialize()

            print("已连接到MCP Server!")

            # 列出所有可用工具
            tools = await session.list_tools()
            print(f"\n可用工具 ({len(tools.tools)}):")
            for tool in tools.tools:
                print(f"  - {tool.name}: {tool.description}")

            # 调用工具
            result = await session.call_tool(
                "list_directory",
                arguments={"path": "."}
            )
            print(f"\n目录内容:\n{result.content[0].text}")

            # 读取资源
            resource = await session.read_resource("config://app")
            print(f"\n配置信息:\n{resource.contents[0].text}")

            # 使用提示模板
            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())

在实际应用中，通常需要将MCP能力与LLM的Function Calling结合使用。以下是将MCP工具转换为LLM工具定义的示例：

Python - MCP工具转换为LLM工具

async def convert_mcp_tools_to_llm(session):
    """将MCP工具定义转换为LLM可使用的格式"""
    tools = await session.list_tools()

    llm_tools = []
    for tool in tools.tools:
        llm_tools.append({
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.inputSchema
            }
        })

    return llm_tools

async def handle_llm_tool_call(session, tool_name, arguments):
    """处理LLM的工具调用请求"""
    result = await session.call_tool(tool_name, arguments)
    return result.content[0].text

八、安全机制与最佳实践

安全是MCP协议设计的核心考量。由于MCP Server可以访问文件系统、执行代码、连接数据库等敏感操作，必须采取严格的安全措施来保护用户数据和系统安全。

8.1 安全架构原则

Python - 安全Server配置

from mcp.server import Server
from mcp.types import ToolResult
import logging

server = Server("secure-server")

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mcp-server")

# 工具权限控制
TOOL_PERMISSIONS = {
    "read_file": {"requires_approval": False},
    "write_file": {"requires_approval": True},
    "delete_file": {"requires_approval": True},
    "execute_query": {"requires_approval": True},
}

@server.tool()
async def secure_tool_wrapper(tool_name: str, **kwargs) -> ToolResult:
    """安全包装器：验证权限并记录操作"""
    permission = TOOL_PERMISSIONS.get(tool_name)

    if permission is None:
        return ToolResult(
            content=f"未知工具: {tool_name}",
            isError=True
        )

    # 记录操作日志
    logger.info(
        f"工具调用: {tool_name}, "
        f"参数: {kwargs}, "
        f"需要审批: {permission['requires_approval']}"
    )

    # 执行实际工具（此处应调用对应的工具函数）
    # result = await execute_tool(tool_name, **kwargs)

    return ToolResult(content="操作完成")

8.2 常见安全威胁

Python - 输入验证与消毒

from pydantic import BaseModel, validator
import re

class SafeQueryArgs(BaseModel):
    """安全的SQL查询参数"""
    query: str
    max_rows: int = 100

    @validator('query')
    def validate_query(cls, v):
        """验证SQL查询安全性"""
        # 禁止危险操作
        dangerous_keywords = [
            'DROP', 'DELETE', 'TRUNCATE', 'ALTER',
            'INSERT', 'UPDATE', 'EXEC', 'EXECUTE'
        ]
        upper_query = v.upper()
        for keyword in dangerous_keywords:
            if keyword in upper_query:
                raise ValueError(f"禁止执行危险操作: {keyword}")

        # 长度限制
        if len(v) > 1000:
            raise ValueError("查询长度超过限制")

        return v

    @validator('max_rows')
    def validate_max_rows(cls, v):
        """验证返回行数限制"""
        if v < 1 or v > 1000:
            raise ValueError("返回行数必须在1-1000之间")
        return v

8.3 日志与审计

完善的日志记录和审计机制是MCP Server安全运行的重要保障。通过记录所有关键操作，可以快速定位问题并满足合规要求。

Python - 审计日志实现

"""
MCP Server审计日志实现
记录所有工具调用和安全事件
"""
import logging
import json
from datetime import datetime
from typing import Any
from dataclasses import dataclass, asdict

# 配置审计日志
audit_logger = logging.getLogger("mcp.audit")
audit_logger.setLevel(logging.INFO)

# 创建JSON格式的审计日志处理器
class AuditFormatter(logging.Formatter):
    def format(self, record):
        audit_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": record.levelname,
            "message": record.getMessage(),
            "extra": getattr(record, 'extra', {})
        }
        return json.dumps(audit_entry, ensure_ascii=False)

handler = logging.StreamHandler()
handler.setFormatter(AuditFormatter())
audit_logger.addHandler(handler)

@dataclass
class AuditEvent:
    """审计事件数据结构"""
    event_type: str  # tool_call, auth_failure, permission_denied
    tool_name: str
    arguments: dict
    result: str
    duration_ms: float
    user_id: str = None
    error: str = None

def log_tool_call(event: AuditEvent):
    """记录工具调用审计日志"""
    audit_logger.info(
        f"Tool调用: {event.tool_name}",
        extra={"extra": asdict(event)}
    )

def log_security_event(event_type: str, details: dict):
    """记录安全事件审计日志"""
    audit_logger.warning(
        f"安全事件: {event_type}",
        extra={"extra": details}
    )

# 使用示例
async def audited_tool_call(tool_name: str, args: dict) -> Any:
    """带审计的工具调用包装器"""
    start_time = datetime.now()

    try:
        # 执行工具
        result = await execute_tool(tool_name, args)
        duration = (datetime.now() - start_time).total_seconds() * 1000

        # 记录成功调用
        event = AuditEvent(
            event_type="tool_call",
            tool_name=tool_name,
            arguments=args,
            result="success",
            duration_ms=duration
        )
        log_tool_call(event)

        return result

    except PermissionError as e:
        duration = (datetime.now() - start_time).total_seconds() * 1000
        # 记录权限拒绝
        event = AuditEvent(
            event_type="permission_denied",
            tool_name=tool_name,
            arguments=args,
            result="denied",
            duration_ms=duration,
            error=str(e)
        )
        log_tool_call(event)
        raise

    except Exception as e:
        duration = (datetime.now() - start_time).total_seconds() * 1000
        # 记录执行错误
        event = AuditEvent(
            event_type="tool_error",
            tool_name=tool_name,
            arguments=args,
            result="error",
            duration_ms=duration,
            error=str(e)
        )
        log_tool_call(event)
        raise

if keyword in upper_query: raise ValueError(f"禁止执行: {keyword}") # 只允许SELECT查询 if not upper_query.strip().startswith('SELECT'): raise ValueError("只允许SELECT查询") return v @validator('max_rows') def validate_max_rows(cls, v): """限制返回行数""" if v > 1000: return 1000 if v < 1: return 1 return v @server.tool() async def safe_query(args: SafeQueryArgs) -> ToolResult: """安全的数据库查询工具""" try: # 使用参数化查询防止SQL注入 results = await db.execute( args.query + f" LIMIT {args.max_rows}" ) return ToolResult(content=json.dumps(results)) except ValueError as e: return ToolResult( content=f"安全验证失败: {str(e)}", isError=True )

8.3 人类审批机制

对于高风险操作，MCP推荐实现人类审批（Human-in-the-Loop）机制。Host应该在执行敏感Tool前征求用户确认。

Python - 人类审批实现

"""
人类审批机制示例
在执行敏感操作前征求用户确认
"""
from dataclasses import dataclass
from enum import Enum
from typing import Optional, Callable
import asyncio

class ApprovalLevel(Enum):
    NONE = "none"           # 无需审批
    NOTIFICATION = "notification"  # 仅通知
    CONFIRMATION = "confirmation"  # 需要确认
    DIALOG = "dialog"       # 需要详细对话框

@dataclass
class ApprovalRequest:
    tool_name: str
    arguments: dict
    description: str
    risk_level: ApprovalLevel

class ApprovalManager:
    def __init__(self):
        self.approval_configs: dict[str, ApprovalLevel] = {}
        self.approval_history: list[dict] = []

    def configure_tool(self, tool_name: str, level: ApprovalLevel):
        """配置工具的审批级别"""
        self.approval_configs[tool_name] = level

    async def request_approval(
        self, request: ApprovalRequest
    ) -> bool:
        """请求用户审批"""
        level = self.approval_configs.get(
            request.tool_name,
            ApprovalLevel.CONFIRMATION
        )

        if level == ApprovalLevel.NONE:
            return True

        if level == ApprovalLevel.NOTIFICATION:
            print(f"[通知] 即将执行: {request.tool_name}")
            return True

        # 需要用户确认的操作
        print(f"\n{'='*50}")
        print(f"工具调用审批请求")
        print(f"{'='*50}")
        print(f"工具: {request.tool_name}")
        print(f"描述: {request.description}")
        print(f"参数: {request.arguments}")
        print(f"风险等级: {level.value}")
        print(f"{'='*50}")

        response = input("是否允许执行？(y/n): ").strip().lower()

        approved = response == 'y'

        # 记录审批历史
        self.approval_history.append({
            "tool": request.tool_name,
            "approved": approved,
            "arguments": request.arguments
        })

        return approved

# 使用示例
async def main():
    manager = ApprovalManager()

    # 配置高风险工具需要确认
    manager.configure_tool("delete_file", ApprovalLevel.CONFIRMATION)
    manager.configure_tool("execute_query", ApprovalLevel.DIALOG)
    manager.configure_tool("read_file", ApprovalLevel.NONE)

    # 请求删除文件审批
    request = ApprovalRequest(
        tool_name="delete_file",
        arguments={"path": "important.txt"},
        description="删除文件 important.txt",
        risk_level=ApprovalLevel.CONFIRMATION
    )

    approved = await manager.request_approval(request)
    if approved:
        print("用户已批准，执行删除操作")
    else:
        print("用户拒绝，操作已取消")

8.4 认证与授权

对于远程MCP Server（SSE传输），认证和授权是必不可少的安全机制。MCP协议本身定义了认证的框架，具体实现由Server和Host协商决定。

Python - OAuth 2.0集成示例

"""
MCP Server OAuth 2.0认证示例
展示如何在MCP Server中集成OAuth 2.0认证
"""
from fastapi import FastAPI, Depends, HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from authlib.integrations.starlette_client import OAuth
from starlette.config import Config
import jwt

app = FastAPI()
security = HTTPBearer()

# OAuth配置
oauth = OAuth()
oauth.register(
    name='mcp_provider',
    client_id='your-client-id',
    client_secret='your-client-secret',
    server_metadata_url='https://auth.example.com/.well-known/openid-configuration',
    client_kwargs={'scope': 'openid email profile mcp:tools'}
)

# JWT验证
def verify_token(credentials: HTTPAuthorizationCredentials = Security(security)):
    try:
        payload = jwt.decode(
            credentials.credentials,
            'your-secret-key',
            algorithms=['HS256']
        )
        return payload
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="Token已过期")
    except jwt.InvalidTokenError:
        raise HTTPException(status_code=401, detail="无效的Token")

# 需要认证的MCP端点
@app.get("/mcp/tools")
async def list_tools(user=Depends(verify_token)):
    """列出可用工具（需要认证）"""
    # 根据用户权限返回工具列表
    tools = get_tools_for_user(user['sub'])
    return {"tools": tools}

@app.post("/mcp/tools/{tool_name}/call")
async def call_tool(
    tool_name: str,
    arguments: dict,
    user=Depends(verify_token)
):
    """调用工具（需要认证+授权）"""
    # 检查用户是否有权调用该工具
    if not has_permission(user['sub'], tool_name):
        raise HTTPException(status_code=403, detail="权限不足")

    result = await execute_tool(tool_name, arguments)
    return result

# 登录端点
@app.get("/login")
async def login(request):
    """重定向到OAuth提供商"""
    redirect_uri = request.url_for('auth_callback')
    return await oauth.mcp_provider.authorize_redirect(request, redirect_uri)

@app.get("/auth/callback")
async def auth_callback(request):
    """OAuth回调处理"""
    token = await oauth.mcp_provider.authorize_access_token(request)
    # 生成JWT并返回给客户端
    jwt_token = create_jwt(token)
    return {"access_token": jwt_token, "token_type": "bearer"}

九、实战：构建一个完整的MCP Server

本节将综合前面所学，从零开始构建一个实用的天气查询MCP Server。该Server提供天气查询工具、城市资源和天气预报提示模板。

Python - 天气查询MCP Server（完整版）

"""
天气查询MCP Server
提供天气查询工具、城市资源和天气预报提示模板
"""
import json
import asyncio
import httpx
from datetime import datetime
from mcp.server import Server
from mcp.types import (
    Tool, Resource, ResourceContents,
    ToolResult, Prompt, PromptMessage, Role
)
from pydantic import BaseModel

server = Server("weather-server")

# 模拟天气数据（实际应用中应调用真实API）
WEATHER_DATA = {
    "北京": {"temp": 25, "condition": "晴", "humidity": 45, "wind": "北风3级"},
    "上海": {"temp": 28, "condition": "多云", "humidity": 70, "wind": "东南风2级"},
    "广州": {"temp": 32, "condition": "雷阵雨", "humidity": 85, "wind": "南风2级"},
    "深圳": {"temp": 31, "condition": "多云", "humidity": 80, "wind": "西南风3级"},
    "杭州": {"temp": 26, "condition": "晴转多云", "humidity": 60, "wind": "东风2级"},
    "成都": {"temp": 24, "condition": "阴", "humidity": 75, "wind": "微风"},
    "武汉": {"temp": 29, "condition": "晴", "humidity": 55, "wind": "北风2级"},
    "西安": {"temp": 27, "condition": "晴", "humidity": 40, "wind": "西北风3级"},
}

# ==================== 工具参数 ====================

class WeatherQueryArgs(BaseModel):
    city: str
    unit: str = "celsius"

class ForecastArgs(BaseModel):
    city: str
    days: int = 3

class CompareArgs(BaseModel):
    cities: list[str]

# ==================== Tools ====================

@server.tool()
async def query_weather(args: WeatherQueryArgs) -> ToolResult:
    """查询指定城市的当前天气"""
    city = args.city

    if city not in WEATHER_DATA:
        available = ", ".join(WEATHER_DATA.keys())
        return ToolResult(
            content=f"未找到 {city} 的天气数据。支持的城市: {available}",
            isError=True
        )

    data = WEATHER_DATA[city]
    temp = data["temp"]
    if args.unit == "fahrenheit":
        temp = temp * 9/5 + 32

    result = {
        "city": city,
        "temperature": f"{temp}°{'F' if args.unit == 'fahrenheit' else 'C'}",
        "condition": data["condition"],
        "humidity": f"{data['humidity']}%",
        "wind": data["wind"],
        "update_time": datetime.now().isoformat()
    }

    return ToolResult(content=json.dumps(result, ensure_ascii=False))

@server.tool()
async def get_forecast(args: ForecastArgs) -> ToolResult:
    """获取未来几天的天气预报"""
    if args.city not in WEATHER_DATA:
        return ToolResult(
            content=f"未找到 {args.city} 的天气数据",
            isError=True
        )

    base = WEATHER_DATA[args.city]
    forecasts = []
    for i in range(args.days):
        forecast = {
            "date": f"第{i+1}天",
            "city": args.city,
            "temp_high": base["temp"] + i,
            "temp_low": base["temp"] - 2 + i,
            "condition": base["condition"],
            "humidity": f"{base['humidity']}%"
        }
        forecasts.append(forecast)

    return ToolResult(
        content=json.dumps(forecasts, ensure_ascii=False, indent=2)
    )

@server.tool()
async def compare_cities(args: CompareArgs) -> ToolResult:
    """比较多个城市的天气"""
    results = []
    for city in args.cities:
        if city in WEATHER_DATA:
            data = WEATHER_DATA[city]
            results.append({
                "city": city,
                "temp": f"{data['temp']}°C",
                "condition": data["condition"]
            })
        else:
            results.append({
                "city": city,
                "error": "数据不可用"
            })

    return ToolResult(
        content=json.dumps(results, ensure_ascii=False, indent=2)
    )

# ==================== Resources ====================

@server.resource("weather://cities")
async def get_supported_cities() -> ResourceContents:
    """获取支持的城市列表"""
    cities = list(WEATHER_DATA.keys())
    return ResourceContents(
        uri="weather://cities",
        mimeType="application/json",
        text=json.dumps(cities, ensure_ascii=False)
    )

@server.resource("weather://{city}/current")
async def get_city_weather_resource(city: str) -> ResourceContents:
    """获取城市当前天气资源"""
    if city not in WEATHER_DATA:
        raise ValueError(f"城市不存在: {city}")

    data = WEATHER_DATA[city]
    return ResourceContents(
        uri=f"weather://{city}/current",
        mimeType="application/json",
        text=json.dumps({
            "city": city,
            "data": data,
            "timestamp": datetime.now().isoformat()
        }, ensure_ascii=False)
    )

# ==================== Prompts ====================

@server.prompt()
async def weather_advisor_prompt(
    city: str,
    activity: str
) -> list[PromptMessage]:
    """天气出行建议提示模板"""
    weather_data = WEATHER_DATA.get(city, {})
    prompt_text = f"""请根据以下天气信息，为用户的{activity}活动提供出行建议：

城市: {city}
温度: {weather_data.get('temp', '未知')}°C
天气: {weather_data.get('condition', '未知')}
湿度: {weather_data.get('humidity', '未知')}%
风力: {weather_data.get('wind', '未知')}

请给出：
1. 是否适合进行该活动
2. 穿衣建议
3. 注意事项
4. 替代方案（如果不适合）"""

    return [PromptMessage(role=Role.USER, content=prompt_text)]

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

async def main():
    from mcp.server.stdio import stdio_server
    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())

TypeScript MCP Server示例

TypeScript是MCP的另一种主流开发语言。以下是使用官方TypeScript SDK构建MCP Server的完整示例：

TypeScript - MCP Server基础示例

import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

// 创建MCP Server实例
const server = new McpServer({
  name: "weather-server",
  version: "1.0.0",
});

// 定义工具 - 使用zod进行参数验证
server.tool(
  "query_weather",
  "查询指定城市的当前天气",
  {
    city: z.string().describe("城市名称"),
    unit: z.enum(["celsius", "fahrenheit"]).default("celsius").describe("温度单位"),
  },
  async ({ city, unit }) => {
    // 模拟天气数据
    const weatherData: Record<string, any> = {
      "北京": { temp: 25, condition: "晴", humidity: 45 },
      "上海": { temp: 28, condition: "多云", humidity: 70 },
      "广州": { temp: 32, condition: "雷阵雨", humidity: 85 },
    };

    const data = weatherData[city];
    if (!data) {
      return {
        content: [{ type: "text", text: `未找到 ${city} 的天气数据` }],
        isError: true,
      };
    }

    let temp = data.temp;
    if (unit === "fahrenheit") {
      temp = temp * 9/5 + 32;
    }

    return {
      content: [{
        type: "text",
        text: JSON.stringify({
          city,
          temperature: `${temp}°${unit === "fahrenheit" ? "F" : "C"}`,
          condition: data.condition,
          humidity: `${data.humidity}%`,
        }),
      }],
    };
  }
);

// 定义资源
server.resource(
  "config",
  "config://app",
  async (uri) => ({
    contents: [{
      uri: uri.href,
      mimeType: "application/json",
      text: JSON.stringify({
        name: "天气查询服务",
        version: "1.0.0",
        supportedCities: ["北京", "上海", "广州"],
      }),
    }],
  })
);

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

main().catch(console.error);

错误处理与重试模式

健壮的MCP Server需要完善的错误处理机制。以下是常见的错误处理模式和最佳实践：

Python - MCP错误处理模式

import asyncio
import logging
from functools import wraps
from typing import Callable, Any

logger = logging.getLogger("mcp-server")

# ==================== 重试装饰器 ====================

def with_retry(
    max_retries: int = 3,
    delay: float = 1.0,
    backoff: float = 2.0,
    exceptions: tuple = (Exception,)
):
    """带指数退避的重试装饰器"""
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        async def wrapper(*args, **kwargs) -> Any:
            last_exception = None
            current_delay = delay

            for attempt in range(max_retries + 1):
                try:
                    return await func(*args, **kwargs)
                except exceptions as e:
                    last_exception = e
                    if attempt < max_retries:
                        logger.warning(
                            f"尝试 {attempt + 1}/{max_retries} 失败: {e}, "
                            f"{current_delay}秒后重试..."
                        )
                        await asyncio.sleep(current_delay)
                        current_delay *= backoff
                    else:
                        logger.error(f"所有重试均失败: {e}")

            raise last_exception
        return wrapper
    return decorator

# ==================== 工具错误处理 ====================

class MCPErrorHandler:
    """MCP错误处理器"""

    @staticmethod
    def create_error_result(message: str, error_code: str = "UNKNOWN_ERROR"):
        """创建标准错误结果"""
        return {
            "content": [{
                "type": "text",
                "text": f"[{error_code}] {message}"
            }],
            "isError": True,
            "_meta": {"errorCode": error_code}
        }

    @staticmethod
    def handle_validation_error(e: Exception):
        """处理参数验证错误"""
        return MCPErrorHandler.create_error_result(
            f"参数验证失败: {str(e)}",
            error_code="VALIDATION_ERROR"
        )

    @staticmethod
    def handle_not_found(resource: str, identifier: str):
        """处理资源未找到"""
        return MCPErrorHandler.create_error_result(
            f"未找到{resource}: {identifier}",
            error_code="NOT_FOUND"
        )

    @staticmethod
    def handle_permission_denied(operation: str):
        """处理权限不足"""
        return MCPErrorHandler.create_error_result(
            f"权限不足，无法执行操作: {operation}",
            error_code="PERMISSION_DENIED"
        )

    @staticmethod
    def handle_rate_limit(retry_after: int = 60):
        """处理速率限制"""
        return {
            "content": [{
                "type": "text",
                "text": f"请求过于频繁，请在{retry_after}秒后重试"
            }],
            "isError": True,
            "_meta": {
                "errorCode": "RATE_LIMITED",
                "retryAfter": retry_after
            }
        }

# ==================== 使用示例 ====================

@server.tool()
@with_retry(max_retries=3, delay=1.0, exceptions=(ConnectionError,))
async def query_weather_with_retry(args: WeatherQueryArgs) -> ToolResult:
    """带重试机制的天气查询工具"""
    try:
        # 模拟可能失败的API调用
        data = await fetch_weather_api(args.city)
        return ToolResult(content=json.dumps(data))
    except ConnectionError as e:
        logger.error(f"天气API连接失败: {e}")
        raise  # 交给重试装饰器处理
    except TimeoutError as e:
        return MCPErrorHandler.create_error_result(
            "天气API请求超时",
            error_code="TIMEOUT"
        )
    except Exception as e:
        logger.exception("未知错误")
        return MCPErrorHandler.create_error_result(
            f"查询失败: {str(e)}",
            error_code="INTERNAL_ERROR"
        )

十、MCP生态系统与未来展望

MCP协议自发布以来，已经形成了一个活跃的生态系统。从官方Server到社区贡献，从工具库到最佳实践，MCP正在成为AI工具集成的事实标准。

社区贡献

除了官方Server，MCP社区也贡献了大量高质量的Server和工具库。这些社区项目覆盖了更多使用场景，推动了MCP生态的繁荣。

未来发展方向