摘要：在大语言模型（LLM）与 AI Agent（智能体）全面落地的今天，如何让 AI 安全、高效、标准化地连接外部数据库、API、本地文件与协同工具，成为构建生产级 AI 应用的核心瓶颈。

Model Context Protocol (MCP，模型上下文协议) 应运而生。作为 AI 领域的“USB-C 开放接口标准”，MCP 彻底解耦了 AI 客户端（Host/Agent）与外部数据源（Tools/Resources），将传统的 N × M 网状接入难题 降维为 N + M 标准化对接。

本文将从底层原理出发，深入拆解 MCP 的架构设计、三大原语（Tools、Resources、Prompts）、传输层协议与通信生命周期，并结合 Python 手把手带你实现一个生产级 MCP Server 与自定义 Agent Client，最后总结企业级部署的安全防御与治理最佳实践。

前言：AI 连接万物的“USB-C 时刻”

在大模型爆发的早期，开发者为了给 AI 助手增加“外部能力”，经历了几个阶段：

  1. 纯 Prompt 工程：将上下文写死在提示词中（受限于 Token 窗口与静态时效）。

  2. 硬编码 Function Calling：为每一个模型硬写工具调用逻辑（绑定特定 API 与数据格式）。

  3. 私有插件系统：每个 AI IDE（如 Cursor）或对话客户端（如 Claude Desktop）都有一套自己的插件开发标准。

这直接导致了严重的 N × M 接入困境：

如果有 5 个 AI 客户端（Claude Desktop, Cursor, VS Code Extension, 自研 Agent 系统, Windsurf）和 5 个外部系统（PostgreSQL, GitHub, Slack, Jira, 本地文件系统），开发者需要编写 5 × 5 = 25 个适配器。每当工具或客户端更新，所有连接器都需要重写。

【传统硬编码：N × M 复杂度】

AI 客户端 A ────┬────> PostgreSQL 连接器
AI 客户端 B ────┼────> GitHub 连接器
AI 客户端 C ────┼────> Slack 连接器
AI 客户端 D ────┴────> Jira 连接器

MCP（Model Context Protocol）的出现彻底改变了这一格局。正如 USB-C 接口统一了外设硬件标准 一样，MCP 统一了 LLM 与外部上下文连接的协议接口：

【MCP 架构：N + M 标准化复杂度】

AI 客户端 A ┐                               ┌ 数据库 MCP Server
AI 客户端 B ├───────>  [ MCP 协议 ]  ───────┼ GitHub MCP Server
AI 客户端 C ┘                               └ Slack MCP Server

客户端只需要实现一个 MCP Client，服务器只需实现一个 MCP Server，接入复杂度立即降至 N + M！

一、 什么是 Model Context Protocol (MCP)？

1.1 核心定义

Model Context Protocol (MCP) 是由 Anthropic 于 2024 年底公开发布并在全行业快速推广的开放通信标准。

它允许 AI 应用程序（如 AI IDE、桌面助手、Agent 平台）通过统一且安全的方式，发现并调用运行在本地或远程服务器上的数据资源（Resources）、可执行工具（Tools） 和 提示词模版（Prompts）。

1.2 MCP 的分层解耦架构

MCP 采用了清晰的客户端-服务器（Client-Server）架构，其中包含四个关键角色：

┌─────────────────────────────────────────────────────────┐
│                      MCP Host                           │
│  ┌──────────────────┐            ┌───────────────────┐  │
│  │   LLM 智能引擎    │            │    MCP Client     │  │
│  └────────┬─────────┘            └─────────┬─────────┘  │
└───────────┼────────────────────────────────┼────────────┘
            │                                │
            │  (推理与抉择)                   │  (JSON-RPC 通信)
            ▼                                ▼
┌─────────────────────────────────────────────────────────┐
│                      MCP Server                         │
│  ┌───────────────────────────────────────────────────┐  │
│  │  三要素暴露：Tools / Resources / Prompts          │  │
│  └────────────────────────┬──────────────────────────┘  │
│                           │                             │
│                           ▼                             │
│             底座服务 (Database, API, Files)             │
└─────────────────────────────────────────────────────────┘

  1. MCP Host：包含 LLM 的宿主应用程序（如 Claude Desktop、Cursor、自研 Agent）。它掌控着用户交互界面与大模型推理主循环。

  2. MCP Client：运行于 MCP Host 内部的客户端模块。它负责管理与各个 MCP Server 的连接、进行能力协商并转换数据格式。

  3. MCP Server：独立的上下文提供程序。它通过标准协议暴露数据与能力，不直接参与 LLM 的训练或推理。

  4. LLM（大语言模型）：负责理解用户意图，生成对 MCP Tools 的调用指令或对 MCP Resources 进行归纳总结。

二、 MCP 架构设计与三大核心原语

MCP 将外部系统提供给 AI 的能力高度抽象为三大核心原语（Primitives）：Tools（工具）、Resources（资源） 和 Prompts（提示词）。

2.1 三大核心原语对比

原语名称 核心性质 是否产生副作用 抽象类比 典型应用场景
Tools（工具） 可执行函数/动作 是（可写入/修改） 操作系统函数/REST API 提交 Git Commit、发送邮件、执行 SQL 写操作
Resources（资源） 只读数据与上下文 否（只读读取） 文件系统 URI / GET 接口 深度读取本地文件、查询数据库日志、读取配置
Prompts（提示词） 参数化文本模版 否（结构化生成） 工作流宏/快捷命令 快速触发重构代码模版、自动化 Weekly 报告生成

2.2 详细原语机制剖析

1. Tools（工具）：模型的“手和脚”

  • 定义：由服务器暴露给模型的模型可控函数（Model-controlled Functions）。

  • 规范：每个 Tool 必须拥有唯一的名称（name）、清晰的描述（description）以及符合 JSON Schema 规范的参数声明。

  • 安全性：协议建议所有带副作用的 Tool 调用都应支持人工确认（Human-in-the-Loop, HITL）机制。

2. Resources（资源）：模型的“眼睛”

  • 定义：由 URI 唯一标识的数据上下文（例如 file:///logs/app.log 或 postgres://db/users）。

  • 类型：

    • 静态资源：固定 URI 指向的静态文件或配置。

    • 动态模版资源（Resource Templates）：带参数的 URI 模版，例如 github://{owner}/{repo}/issues。

  • 事件通知：Server 可以在资源内容发生变更时，向 Client 发送 notifications/resources/updated 通知，提示 Client 刷新上下文。

3. Prompts（提示词模版）：经验的“复用器”

  • 定义：预先定好的标准化提示词片段或对话上下文。

  • 作用：让用户在客户端界面方便地选择预设好的高级指令，并将上下文资源与参数自动填充至对话框中。

2.3 传输层协议（Transports）

MCP 协议与具体传输介质解耦，主要支持以下几种底层的通信方式：

                             ┌───────────────┐
                             │  MCP 消息层   │
                             │ (JSON-RPC 2.0)│
                             └───────┬───────┘
                                     │
                 ┌───────────────────┴───────────────────┐
                 ▼                                       ▼
     Stdio Transport (本地)                 SSE Transport (远程)
┌─────────────────────────┐             ┌─────────────────────────┐
│ 子进程 stdin / stdout   │             │ HTTP Server-Sent Events │
│ 低延迟、零网络开销       │             │ 支持分布式、云端多租户   │
└─────────────────────────┘             └─────────────────────────┘

  1. Stdio Transport（标准输入输出）：

    • 运行机制：MCP Host 通过命令行启动 MCP Server 子进程，通过管道（stdin/stdout）进行二进制/文本双向通信。

    • 场景：适合本地工具（如本地文件管理、Git 操作、本地 SQLite 数据库）。极高吞吐、零网络暴露风险。

  2. SSE Transport（Server-Sent Events over HTTP）：

    • 运行机制：客户端发起 GET 请求建立 SSE 订阅通道获取服务端长连接事件，后续客户端请求通过 HTTP POST 发送到服务端指定的 Endpoint。

    • 场景：适合远程分布式服务（如企业级知识库、 SaaS API、跨网络数据库服务）。

三、 协议通信机制与生命周期深挖

MCP 全程基于 JSON-RPC 2.0 规范进行异步双向消息传递。

3.1 核心通信生命周期序列图

整个通信流程包含连接握手与初始化、能力发现、工具执行/资源读取 三个阶段：

[ MCP Client ]                                          [ MCP Server ]
      │                                                       │
      │ ───────────────── 1. initialize ───────────────────>  │
      │    (协议版本, Client capabilities, ClientInfo)        │
      │                                                       │
      │ <──────────────── 2. response ─────────────────────── │
      │    (协议版本, Server capabilities, ServerInfo)        │
      │                                                       │
      │ ───────────────── 3. initialized ───────────────────> │
      │    (通知 Server 初始化建立完成)                        │
      │                                                       │
      ├───────────────────────────────────────────────────────┤
      │                     能力发现与调用                     │
      ├───────────────────────────────────────────────────────┤
      │                                                       │
      │ ──────────────── 4. tools/list ─────────────────────> │
      │ <─────────────── 5. tools response ────────────────── │
      │    (返回可调用的 Tool JSON Schema 列表)                │
      │                                                       │
      │ ──────────────── 6. tools/call ─────────────────────> │
      │    (name: "calculate_tax", arguments: {amount: 100})  │
      │                                                       │
      │ <─────────────── 7. progress report (可选) ─────────── │
      │                                                       │
      │ <─────────────── 8. call response ─────────────────── │
      │    (content: [{type: "text", text: "结果为: 20"}])    │

3.2 真实 JSON-RPC 消息报文体解析

为了清晰理解底层原理，我们来看看真实的传输报文：

1. 初始化握手请求（Client -> Server）
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "initialize",
  "params": {
    "protocolVersion": "2024-11-05",
    "capabilities": {
      "roots": { "listChanged": true },
      "sampling": {}
    },
    "clientInfo": {
      "name": "CustomAgentApp",
      "version": "1.0.0"
    }
  }
}
2. 工具列表响应（Server -> Client）
{
  "jsonrpc": "2.0",
  "id": 2,
  "result": {
    "tools": [
      {
        "name": "query_inventory",
        "description": "查询仓库商品实时库存与价格",
        "inputSchema": {
          "type": "object",
          "properties": {
            "sku_id": {
              "type": "string",
              "description": "商品 SKU 编号"
            }
          },
          "required": ["sku_id"]
        }
      }
    ]
  }
}
3. 执行工具请求（Client -> Server）
{
  "jsonrpc": "2.0",
  "id": 3,
  "method": "tools/call",
  "params": {
    "name": "query_inventory",
    "arguments": {
      "sku_id": "SKU-99821"
    }
  }
}

四、 实战演练：从零构建你的第一个 Python MCP Server

接下来，我们使用官方的 Python MCP SDK (mcp) 和高阶封装框架 FastMCP，手把手构建一个包含 Tools、Resources、Prompts 以及长任务进度通知的全面 MCP Server。

4.1 环境准备

确保你已安装 Python 3.10+，推荐使用现代包管理工具 uv 或 pip：

pip install mcp httpx

4.2 完整 MCP Server 实现（server.py）

编写如下代码，创建一个名为 DevOps-Assistant 的 MCP 服务：

import asyncio
from datetime import datetime, timezone
from typing import Annotated
from mcp.server.fastmcp import FastMCP, Context
from mcp.server.session import ServerSession

# 1. 创建 FastMCP 服务实例
mcp = FastMCP(
    name="DevOps-Assistant-Server",
    dependencies=["httpx"]
)

# ==================== A. 核心原语 1：Tools（工具） ====================

@mcp.tool()
def calculate_disk_usage(path: str) -> str:
    """计算指定目录的估算磁盘空间占用（单位 MB）。"""
    import os
    try:
        total_size = 0
        for dirpath, dirnames, filenames in os.walk(path):
            for f in filenames:
                fp = os.path.join(dirpath, f)
                if not os.path.islink(fp):
                    total_size += os.path.getsize(fp)
        size_mb = round(total_size / (1024 * 1024), 2)
        return f"目录 '{path}' 当前占用空间: {size_mb} MB"
    except Exception as e:
        return f"查询出错: {str(e)}"

@mcp.tool()
async def execute_batch_task(
    total_steps: int,
    ctx: Annotated[Context[ServerSession, None], "上下文注入"]
) -> str:
    """模拟一个长时间运行的批量运维任务，演示流式进度汇报（Progress Reporting）。"""
    for step in range(1, total_steps + 1):
        await asyncio.sleep(0.3)  # 模拟任务耗时
        # 向客户端上报进度通知
        await ctx.report_progress(
            progress=step,
            total=total_steps,
            message=f"正在处理第 {step}/{total_steps} 个节点的配置同步..."
        )
    return f"成功完成全部 {total_steps} 个节点的配置同步任务！"

# ==================== B. 核心原语 2：Resources（资源） ====================

@mcp.resource("system://metrics/{hostname}")
def get_system_metrics(hostname: str) -> str:
    """动态资源：读取特定主机名的系统实时监控指标。"""
    now = datetime.now(timezone.utc).isoformat()
    return f"""
    [主机监控指标]
    主机名: {hostname}
    时间戳: {now}
    CPU 使用率: 42.5%
    内存空闲率: 61.2%
    服务状态: Healthy
    """

@mcp.resource("config://app-settings")
def get_static_config() -> str:
    """静态资源：获取应用配置信息。"""
    return '{"env": "production", "debug": false, "max_connections": 500}'

# ==================== C. 核心原语 3：Prompts（提示词） ====================

@mcp.prompt()
def code_review_prompt(language: str, code_snippet: str) -> str:
    """生成专业的代码审查指令模版。"""
    return f"""你是一位 senior {language} 架构师。请针对以下代码片段进行严格的 Code Review。
评估维度：
1. 是否存在内存泄露或未捕获的异常？
2. 时间/空间复杂度是否可优化？
3. 给出优雅的重构版本。

待审查代码：
```{language}
{code_snippet}

"""

==================== D. 启动入口 ====================

if name == "main":

# 使用 Stdio 标准输入输出模式运行 (本地 CLI/IDE 接入最佳选择)

mcp.run(transport="stdio")


---

### 4.3 将 MCP Server 接入 Claude Desktop / Cursor

要让现有的 AI 客户端（如 Claude Desktop）调用你的 MCP Server，只需在其配置文件中添加你的脚本运行命令。

#### 打开配置文件：
* **Mac**: `~/Library/Application Support/Claude/claude_desktop_config.json`
* **Windows**: `%APPDATA%\Claude\claude_desktop_config.json`

#### 写入如下配置：
```json
{
  "mcpServers": {
    "my-devops-tools": {
      "command": "python",
      "args": [
        "/绝对路径/到你的脚本/server.py"
      ]
    }
  }
}

重新启动 Claude Desktop，你将在对话框右下角看到一个“锤子”图标，包含了你刚刚定义的 calculate_disk_usage 和 execute_batch_task 工具！当你在对话框中询问“帮我算一下 /tmp 目录占用了多少空间”时，Claude 会自动发起 MCP 工具调用。

五、 实战演练：构建自定义 MCP Client（客户端通信实现）

如果你正在开发自研的 Agent 框架或大模型应用系统，你需要在代码中集成 MCP Client 模块。

下面演示如何使用 Python mcp SDK 编写一个程序化的客户端，自动连接上面的 Server，列出工具，并结合大模型（例如 DeepSeek / OpenAI API）完成自动化 Loop。

5.1 Python 客户端完整代码（client.py）

import asyncio
import os
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from openai import OpenAI

# 1. 设置 Server 的连接参数 (Stdio 模式)
server_params = StdioServerParameters(
    command="python",
    args=["server.py"], # 确保指向你的 MCP Server 脚本
    env=None
)

async def run_mcp_agent():
    # 2. 建立 Stdio 进程管道连接
    async with stdio_client(server_params) as (read_stream, write_stream):
        # 3. 初始化 Client 动态 Session
        async with ClientSession(read_stream, write_stream) as session:
            # 协议握手
            await session.initialize()
            print("✔ 成功与 MCP Server 建立协议连接并完成初始化握手！\n")

            # A. 动态获取 Server 暴露的所有工具 (Tools List)
            tools_response = await session.list_tools()
            available_tools = tools_response.tools
            print(f"✔ 发现 Server 提供的工具列表: {[t.name for t in available_tools]}")

            # B. 动态获取 Server 暴露的资源 (Resource Read)
            resource_data = await session.read_resource("system://metrics/prod-db-node1")
            print(f"\n[读取 MCP Resource 真实内容]:\n{resource_data.contents[0].text}")

            # C. 调用工具 (Call Tool)
            print("\n正在调用 execute_batch_task 工具，接收实时进度...")
            
            # 进度回调函数
            async def on_progress(progress: float, total: float | None, message: str | None):
                pct = round((progress / (total or 1)) * 100, 1)
                print(f" 进度通知: [{pct}%] - {message}")

            # 执行带有进度追踪的工具
            tool_result = await session.call_tool(
                "execute_batch_task",
                arguments={"total_steps": 5},
                on_progress=on_progress # 注册回调
            )

            print(f"\n[工具最终返回结果]:\n{tool_result.content[0].text}")

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

六、 生产环境中的 MCP 架构演进与安全防线

在将 MCP 架构部署到企业生产环境时，安全与治理是绝对不能忽视的核心。

6.1 核心安全风险矩阵

风险类型 漏洞原理 生产级解决方案
间接提示词注入 外部网页/数据库内包含恶意 Prompt，在读入 Resource 时诱导 LLM 越权执行写工具 对读入的 Context 进行安全过滤；限制 Tools 执行越权破坏操作
未授权工具执行 LLM 产生幻觉误触发数据删除/转账等危险 Tool 强制在敏感 Tools 前增加 Human-in-the-Loop（人工确认） 拦截层
SSRF / 内部网络越权 远程 SSE Server 被利用扫描企业内网 严禁 Server 运行在特权 Pod/机器上，使用网络隔离与 OAuth2 鉴权

6.2 人工确认（Human-in-the-Loop, HITL）架构

对于涉及数据库写操作、部署上线、资金划转的极度危险 Tool，必须在 MCP Client 侧实现弹窗提醒与审批拦截机制：

[ LLM 生成 Tool Call 指令 ]
            │
            ▼
┌───────────────────────┐
│   MCP Client 安全网关  │
└───────────┬───────────┘
            │
    (是否包含危险属性?)
     ├── 否 ──> [ 直接发送给 MCP Server 执行 ]
     │
     └── 是 ──> 触发 HITL 确认 ──> [ UI 弹窗询问用户: "确认执行删除操作吗？" ]
                                      ├── 用户批准 ──> 发送给 MCP Server
                                      └── 用户拒绝 ──> 向 LLM 返回 "User denied action"

七、 总结与未来展望

Model Context Protocol（MCP）的快速崛起，标志着 AI Agent 开发范式从“粗暴硬编码”全面迈向“标准协议化”。

7.1 生态演进现状

截至目前，包括 Anthropic Claude、Cursor、Windsurf、Zed、Continue.dev、Databricks 在内的主流 AI 产品和企业级服务，已全线支持 MCP 协议接入。社区也涌现了数千个开箱即用的 MCP Server（涵盖 GitHub、PostgreSQL、Puppeteer、Slack、Notion 等）。

# 人工智能
Logo
EazyDevelop社区

一站式 AI 云服务平台

更多推荐

cover

办公效率提升方案，OpenClaw 本地智能体搭建攻略（含安装包）

avatar EazyDevelop社区
cover

2026论文工具真实排行[特殊字符]去掉噱头！只留能用的

avatar EazyDevelop社区
cover

百度小程序开发品牌有哪些？2026主流搭建平台你更喜欢哪个？

avatar EazyDevelop社区