MCP六问

Me:
MCP六问：
MCP Server 是什么
Tool、Resource、Prompt 分别是什么
一个 Agent 如何发现和调用 MCP 工具
如何自己封装一个 MCP Server
如何把数据库/API/内部系统包装成 MCP 能力
MCP Server 的权限、安全边界如何设计
本地 MCP 和远程 MCP 有什么区别

请给出详细的、针对计算机科班出身者的回答

DeepSeek:

下面从协议设计、分布式系统、安全模型和实现角度，系统回答这六个问题。

1. MCP Server 是什么

Model Context Protocol (MCP) 是 Anthropic 于 2024 年底提出的一套开放协议，旨在标准化 LLM 应用（Host/Client）与外部工具、数据源之间的交互方式。类比：

HTTP 标准化了浏览器与 Web 服务器之间的通信
LSP (Language Server Protocol) 标准化了编辑器与语言服务之间的通信
MCP 标准化了 AI 应用与工具/资源之间的通信

MCP Server 是实现该协议的服务端进程，它：

暴露一组能力（Tools / Resources / Prompts），通过 JSON-RPC 2.0 协议描述和调用
可以被 MCP Client（如 Claude Desktop、IDE 插件、自定义 Agent 框架）发现并连接
负责执行实际的操作：查询数据库、调用 API、读写文件、执行业务逻辑等

从架构角度看，MCP Server 是一个能力适配层：把异构的后端系统（数据库、内部微服务、文件系统、第三方 SaaS）统一包装成 LLM 可理解、可调用的标准接口。

┌─────────────┐      JSON-RPC 2.0      ┌─────────────┐     ┌──────────┐
│  MCP Client │ ◄────────────────────► │  MCP Server │ ──► │ Database │
│  (Host/LLM) │    stdio / HTTP/SSE    │             │     │ API      │
└─────────────┘                        └─────────────┘     │ File Sys │
                                                           └──────────┘

2. Tool、Resource、Prompt 分别是什么

这是 MCP 协议定义的三种原语（primitives），是 Server 向 Client 暴露能力的三种抽象：

2.1 Tool（工具）

本质：可执行的操作，有输入参数和返回结果，类似于函数签名（function call）
语义：tool = name + description + JSON Schema (input) + 执行函数
特点：
有副作用（可以修改外部状态）
LLM 通过 tools/call 主动调用
执行结果返回给 LLM 作为上下文继续推理
举例：query_database(sql)、send_email(to, subject, body)、create_jira_ticket(...)

// 工具描述（Client 侧看到的 schema）
{
  "name": "get_weather",
  "description": "获取指定城市的实时天气",
  "inputSchema": {
    "type": "object",
    "properties": {
      "city": { "type": "string", "description": "城市名" }
    },
    "required": ["city"]
  }
}

2.2 Resource（资源）

本质：可读取的数据/文件，有 URI 标识，类似于 REST 中的资源或文件系统路径
语义：resource = URI + MIME type + 内容
特点：
只读（协议上不保证，但设计意图是数据源而非操作）
由 Host/LLM 通过 resources/read 拉取内容
可以是静态的（配置文件）、半静态的（数据库某张表的快照）、动态的（实时查询结果）
举例：postgres://orders/schema、file:///workspace/README.md、api://users/current

2.3 Prompt（提示模板）

本质：预设的提示词模板，包含结构化参数，可以组合生成最终发给 LLM 的 prompt
语义：prompt = name + arguments + 模板渲染逻辑
特点：
纯内容生成，无副作用
由 Host 通过 prompts/get 获取渲染后的消息序列
用于复用专家级 prompt、强制特定格式输出、注入系统指令
举例：code_review_guidelines、sql_analysis_template

2.4 三者对比总结

维度	Tool	Resource	Prompt
是否有副作用	是	否	否
调用方式	`tools/call`	`resources/read`	`prompts/get`
返回内容	任意 JSON/文本	带 MIME 的内容	消息序列
类比	函数/RPC	文件/GET 请求	模板/宏
LLM 角色	主动调用执行	被动读取上下文	被动获取提示

3. 一个 Agent 如何发现和调用 MCP 工具

完整的流程如下：

3.1 连接建立（Transport）

MCP 支持两种传输层：

stdio：Client 以子进程方式启动 Server，通过 stdin/stdout 通信（本地优先）
HTTP + SSE：远程 Server，Client 通过 HTTP POST + Server-Sent Events 通信

连接建立后，Client 发送 initialize 握手请求，协商协议版本和能力。

3.2 能力发现（Discovery）

Client 调用 tools/list（或 resources/list、prompts/list），获取：

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/list",
  "params": {}
}

Server 返回：

{
  "tools": [
    {
      "name": "query_database",
      "description": "执行 SQL 查询，返回结果集",
      "inputSchema": { /* JSON Schema */ }
    }
  ]
}

3.3 Agent 端集成

Agent 框架（如 LangChain、自研框架）将工具描述注入 LLM 的 system prompt 或 function calling 参数：

你是一个数据分析助手。你可以调用以下工具：

- query_database(sql: string): 执行 SQL 查询
- get_file_content(path: string): 读取文件内容

LLM 根据用户意图，输出 function call：

{ "name": "query_database", "arguments": { "sql": "SELECT * FROM orders WHERE date > '2025-01-01'" } }

3.4 工具调用（Invocation）

Agent 框架拦截 LLM 的 function call 输出，向 MCP Server 发送：

{
  "method": "tools/call",
  "params": {
    "name": "query_database",
    "arguments": { "sql": "SELECT * FROM orders WHERE date > '2025-01-01'" }
  }
}

Server 执行实际查询，返回：

{
  "content": [{ "type": "text", "text": "[{\"order_id\": 1, ...}]" }],
  "isError": false
}

Agent 将结果作为新的对话轮次追加，LLM 继续推理。这是一个 ReAct 循环：推理 → 调用 → 观察 → 推理，直到 LLM 产出最终回答。

3.5 完整时序

LLM                    Agent Framework              MCP Server
 │                           │                           │
 │  用户问题                  │                           │
 │◄──────────────────────────│                           │
 │                           │  initialize               │
 │                           │──────────────────────────►│
 │                           │  tools/list                │
 │                           │──────────────────────────►│
 │                           │◄──────────────────────────│
 │                           │                           │
 │  注入工具 schema           │                           │
 │◄──────────────────────────│                           │
 │                           │                           │
 │  function call 输出       │                           │
 │──────────────────────────►│                           │
 │                           │  tools/call                │
 │                           │──────────────────────────►│
 │                           │                           │──执行
 │                           │◄──────────────────────────│
 │                           │                           │
 │  工具结果作为上下文        │                           │
 │◄──────────────────────────│                           │
 │                           │                           │
 │  最终回答                 │                           │
 │──────────────────────────►│                           │

4. 如何自己封装一个 MCP Server

以 Python 为例（官方 SDK mcp），最简实现：

4.1 安装

pip install mcp

4.2 最小 Server

# server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
import mcp.types as types

app = Server("my-first-server")

@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="add",
            description="两个数相加",
            inputSchema={
                "type": "object",
                "properties": {
                    "a": {"type": "number"},
                    "b": {"type": "number"}
                },
                "required": ["a", "b"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    if name == "add":
        result = arguments["a"] + arguments["b"]
        return [types.TextContent(type="text", text=str(result))]
    raise ValueError(f"Unknown tool: {name}")

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

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

4.3 核心设计要点

关注点	设计决策
传输层	本地用 stdio（零网络开销），远程用 HTTP/SSE
Schema 设计	工具描述要足够详细（LLM 靠描述决定何时调用），参数用严格 JSON Schema
错误处理	返回 `isError: true` + 错误信息，让 LLM 知道失败原因并能自我修正
并发	Server 需处理并发调用，数据库连接池、锁等
日志	stderr 输出日志（stdio 模式 stdout 被协议占用）
幂等性	写操作工具应尽量设计为幂等，LLM 可能重复调用

5. 如何把数据库/API/内部系统包装成 MCP 能力

5.1 数据库 → MCP

核心策略：工具粒度 = 业务查询粒度，而非原始 SQL 粒度

@app.list_tools()
async def list_tools():
    return [
        types.Tool(
            name="get_recent_orders",
            description="获取最近 N 天的订单列表，按时间倒序",
            inputSchema={
                "type": "object",
                "properties": {
                    "days": {"type": "integer", "default": 7},
                    "limit": {"type": "integer", "default": 50}
                }
            }
        ),
        types.Tool(
            name="get_order_detail",
            description="根据订单 ID 获取订单详细信息",
            inputSchema={
                "type": "object",
                "properties": {
                    "order_id": {"type": "string"}
                },
                "required": ["order_id"]
            }
        ),
        # 可能还需要一个受限的 raw SQL 工具，用于复杂临时查询
        types.Tool(
            name="run_select_sql",
            description="执行只读 SQL 查询（仅 SELECT）",
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {"type": "string", "description": "只允许 SELECT 语句"}
                },
                "required": ["sql"]
            }
        )
    ]

实现要点：
- 使用连接池（如 asyncpg.create_pool）
- 只读工具和写工具分开，写工具需要额外权限控制
- 查询结果序列化为 JSON 文本返回给 LLM
- 大结果集需要截断/分页（LLM 上下文窗口有限）
- 考虑用 Resource 暴露 schema（db://schema/tables），让 LLM 先了解数据结构

5.2 外部 API → MCP

import httpx

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == "search_github_repos":
        async with httpx.AsyncClient() as client:
            resp = await client.get(
                "https://api.github.com/search/repositories",
                params={"q": arguments["query"]},
                headers={"Authorization": f"Bearer {API_TOKEN}"}
            )
            data = resp.json()
            # 只提取 LLM 关心的字段，减少 token
            simplified = [
                {"name": r["full_name"], "stars": r["stargazers_count"], "desc": r["description"]}
                for r in data.get("items", [])[:10]
            ]
            return [types.TextContent(type="text", text=json.dumps(simplified))]

要点：
- 在 Server 层做响应瘦身，只返回 LLM 需要的字段
- API 密钥管理：环境变量注入，不硬编码
- 处理限流、超时、重试
- 将复杂 API 组合拆分为多个细粒度工具

5.3 内部系统 → MCP

内部系统（如 Jira、K8s、CMDB、订单系统）：

@app.list_tools()
async def list_tools():
    return [
        types.Tool(
            name="create_incident_ticket",
            description="在 ITSM 系统中创建故障工单",
            inputSchema={
                "type": "object",
                "properties": {
                    "title": {"type": "string"},
                    "severity": {"type": "string", "enum": ["P1", "P2", "P3"]},
                    "description": {"type": "string"},
                    "assignee": {"type": "string"}
                },
                "required": ["title", "severity", "description"]
            }
        ),
        types.Tool(
            name="get_k8s_pod_status",
            description="获取 Kubernetes 命名空间下 Pod 的状态",
            inputSchema={
                "type": "object",
                "properties": {
                    "namespace": {"type": "string"},
                    "label_selector": {"type": "string"}
                },
                "required": ["namespace"]
            }
        )
    ]

要点：
- 鉴权代理：MCP Server 作为内部系统的统一认证入口，持有服务账号 token
- 审计日志：所有工具调用记录操作者、时间、参数（合规要求）
- 降级策略：内部系统不可用时返回可理解的错误信息
- 批量操作限制：限制一次可操作的资源数量

5.4 通用架构模式

                    ┌──────────────────────────────┐
                    │         MCP Server           │
                    │                              │
 LLM ◄──JSON-RPC──► │  Tool Registry & Dispatcher  │
                    │                              │
                    │  ┌──────────┐ ┌──────────┐  │
                    │  │ DB Adapter│ │API Adapter│ │
                    │  └────┬─────┘ └────┬─────┘  │
                    └───────┼────────────┼────────┘
                            │            │
                      ┌─────▼───┐   ┌────▼─────┐
                      │Postgres │   │ REST API │
                      └─────────┘   └──────────┘

6. MCP Server 的权限、安全边界如何设计

这是生产环境中最关键的问题。LLM 是一个不可完全信任的执行体，它可能被 prompt injection 攻击，产生意外调用。

6.1 权限模型

分层权限设计：

┌─────────────────────────────────────────┐
│ 第 0 层：连接层权限                      │
│ - 谁可以连接这个 MCP Server？            │
│ - 认证方式：API Key / OAuth / mTLS       │
├─────────────────────────────────────────┤
│ 第 1 层：工具级权限                      │
│ - 连接者可以调用哪些工具？               │
│ - 只读工具默认开放，写工具需白名单       │
├─────────────────────────────────────────┤
│ 第 2 层：参数级权限                      │
│ - 工具参数的取值范围限制                 │
│ - SQL 白名单、资源 ID 前缀匹配           │
├─────────────────────────────────────────┤
│ 第 3 层：数据级权限                      │
│ - 行级安全（RLS）、列级脱敏             │
│ - 敏感字段 masking                      │
└─────────────────────────────────────────┘

6.2 关键安全措施

（1）工具分类与默认拒绝

TOOL_POLICIES = {
    "query_database": {"risk": "read", "default": "allow"},
    "get_file": {"risk": "read", "default": "allow", "path_whitelist": ["/workspace", "/data"]},
    "update_record": {"risk": "write", "default": "deny", "require_approval": True},
    "delete_record": {"risk": "dangerous", "default": "deny", "require_human_approval": True},
    "execute_shell": {"risk": "dangerous", "default": "deny"},
}

（2）SQL 注入防护

async def call_tool(name, arguments):
    if name == "run_select_sql":
        sql = arguments["sql"]
        # 只允许 SELECT
        if not sql.strip().upper().startswith("SELECT"):
            return error("只允许 SELECT 查询")
        # 禁止危险关键字
        for keyword in ["INTO OUTFILE", "LOAD_FILE", "SLEEP(", "BENCHMARK("]:
            if keyword.lower() in sql.lower():
                return error(f"禁止使用 {keyword}")
        # 使用参数化查询或受限执行用户
        async with pool.acquire() as conn:
            result = await conn.fetch(sql)  # 使用最低权限数据库用户

（3）人机审批（Human-in-the-Loop）

对高风险操作，MCP Server 可以返回一个”待审批”状态，由人类确认后执行：

# 方案 A：同步阻塞等待审批（通过回调或轮询）
# 方案 B：先返回 request_id，人类通过另一个通道批准后执行
@app.call_tool()
async def call_tool(name, arguments):
    if name == "delete_production_data":
        approval_id = await create_approval_request(name, arguments)
        return [types.TextContent(
            type="text",
            text=f"该操作需要人工审批。审批 ID: {approval_id}。请等待审批通过后重试。"
        )]

（4）Prompt Injection 防御

工具描述中明确边界：该工具只能查询数据，不能执行任何写操作
最小权限原则：每个 Agent 实例只挂载完成任务所需的最少工具
上下文隔离：用户数据、工具结果与系统指令之间用明确分隔符
输出校验：Server 端对 LLM 返回结果做 schema 验证

（5）审计与监控

# 每次工具调用记录结构化审计日志
audit_log = {
    "timestamp": "...",
    "tool": "query_database",
    "arguments": {"sql": "SELECT ..."},
    "caller": "agent-01",
    "session_id": "...",
    "result_summary": "returned 10 rows",
    "duration_ms": 45,
    "error": None
}

6.3 安全边界总结

威胁	防御措施
Prompt Injection 导致恶意工具调用	最小工具集挂载、高危操作审批、参数白名单
SQL 注入	只允许 SELECT、关键字过滤、低权限 DB 用户
数据泄露	列级脱敏、结果截断、敏感字段过滤
越权访问	每层独立鉴权、数据级 RBAC/RLS
资源耗尽	查询超时、结果行数限制、并发限流
不可追溯	全量审计日志、调用链路追踪

7. 本地 MCP 和远程 MCP 的区别

7.1 架构对比

本地 MCP（stdio）：

┌──────────────────┐        ┌──────────────────┐
│   MCP Client     │        │   MCP Server     │
│   (Host 进程)    │  stdio │   (子进程)       │
│                  │◄──────►│                  │
│  Claude Desktop  │ pipe   │  my-server.py    │
│  / IDE / Agent   │        │  （本地运行）     │
└──────────────────┘        └──────────────────┘

Client 通过 spawn() 启动 Server 子进程
通信通过 stdin/stdout，消息以换行符分隔的 JSON-RPC
Server 生命周期与 Client 绑定

远程 MCP（HTTP + SSE）：

┌──────────────────┐     HTTPS     ┌──────────────────┐
│   MCP Client     │◄─────────────►│   MCP Server     │
│                  │   JSON-RPC    │   （远程部署）    │
│  浏览器 / Agent  │  + SSE 推送   │  云服务器 / K8s   │
└──────────────────┘               └──────────────────┘

Server 是一个常驻的 HTTP 服务
Client 通过 POST 发送 JSON-RPC 请求
Server 通过 SSE（Server-Sent Events）推送响应/通知
支持多 Client 并发连接

7.2 详细差异

维度	本地 MCP（stdio）	远程 MCP（HTTP/SSE）
部署方式	与 Client 同机，子进程	独立部署，网络可达
通信协议	stdin/stdout 管道	HTTP POST + SSE
生命周期	Client 启动时 spawn，退出时 kill	常驻服务，独立于 Client
并发模型	单 Client 单 Server 实例	多 Client 共享 Server
认证	进程级信任（本地用户）	需要 API Key / OAuth / mTLS
网络要求	无	需要网络连通性
延迟	极低（进程间通信）	较高（网络往返）
状态管理	可共享 Client 的内存	无状态 or 服务端会话
可扩展性	单机	水平扩展（多副本）
适用场景	个人工具、本地文件、开发调试	团队共享、生产环境、企业服务
版本更新	随 Client 或本地安装更新	服务端统一更新，Client 无感
安全边界	继承本地用户权限	独立的安全域，需要身份认证
调试	可直接本地调试	需要日志收集、监控

7.3 如何选择

需要访问本地资源（文件、localhost 服务）？
├── 是 → 本地 MCP（stdio）
└── 否 → 需要团队共享？
         ├── 是 → 远程 MCP（HTTP/SSE）
         └── 否 → 需要频繁更新、集中管理？
                  ├── 是 → 远程 MCP
                  └── 否 → 本地 MCP 更简单

7.4 混用场景

实际生产中，常见模式是：

本地开发：stdio 模式，快速迭代
生产部署：HTTP/SSE 模式，部署在 K8s，配置 ServiceAccount、Ingress TLS
网关模式：多个远程 MCP Server 通过一个 MCP Gateway 聚合，Client 只连网关

总结

MCP 本质上是把 Function Calling 从各 LLM 厂商的私有协议中抽象出来，成为跨模型、跨框架的开放标准。对计算机科班出身者来说，可以把它理解为：

MCP = OpenAPI/Swagger + JSON-RPC + 服务发现 + 权限模型，专门为 LLM Agent 场景优化

核心设计哲学：
1. 描述优于代码：工具 schema 要写得像 API 文档一样清晰（LLM 靠描述来理解）
2. 能力最小化：暴露尽可能少、尽可能窄的工具，降低 LLM 误用风险
3. 分层安全：连接、工具、参数、数据四层权限控制
4. 可观测性：所有调用必须可审计、可追溯、可回放