MCP协议入门与实践
2026/8/3 13:14:16 网站建设 项目流程

摘要:在大语言模型(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.logpostgres://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,手把手构建一个包含ToolsResourcesPrompts以及长任务进度通知的全面 MCP Server。

4.1 环境准备

确保你已安装 Python 3.10+,推荐使用现代包管理工具uvpip

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. 启动入口 ====================

ifname== "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_usageexecute_batch_task工具!当你在对话框中询问“帮我算一下 /tmp 目录占用了多少空间”时,Claude 会自动发起 MCP 工具调用。

五、 实战演练:构建自定义 MCP Client(客户端通信实现)

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

下面演示如何使用 PythonmcpSDK 编写一个程序化的客户端,自动连接上面的 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 等)。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询