MCP从入门到企业落地:AI Agent工具链实战

开发者教程MCPAI AgentRAG飞书企业权限审计阿里云腾讯云2026-09-22

MCP从入门到企业落地:AI Agent工具链实战

客服主管丢来一个问题:「这家客户上次投诉是什么,我们承诺了什么?」Agent 要回答,得同时查 CRM 的客户记录、飞书群里的沟通记录、内部知识库的售后政策。每个系统的鉴权、分页、字段都不一样。MCP 的作用是把这些系统包装成统一协议下的 Server,Agent 作为 Host 通过 Client 调用。下面按能跑起来的顺序写。

MCP 的三块:Host、Client、Server

Host 是你用的 Agent 客户端,比如 IDE 插件、自研客服工作台、ChatGPT 类桌面端。Client 在 Host 内部,一个 Client 连一个 Server。Server 暴露三类能力:

  • resources:只读数据,比如文档、配置、数据库记录。
  • tools:可执行函数,比如查 CRM、发飞书消息。
  • prompts:预置提示模板,比如「生成客户回访摘要」。
协议层用 JSON-RPC 2.0。传输方式常见三种:

传输场景限制
stdio本地工具、IDE不能跨网络,Server 随 Host 启动
SSE远程推送长连接容易被网关断开,2026 年新项目优先 streamable HTTP
streamable HTTP远程 Server、云部署需要自己做鉴权、限流、审计
flowchart LR U[用户] --> H[Host Agent] H --> C1[MCP Client] H --> C2[MCP Client] C1 --> S1[知识库 Server] C2 --> S2[CRM Server] S2 --> CRM[(CRM API)] S1 --> KB[(向量库)] H --> FS[飞书 Server] FS --> FSAPI[飞书开放平台]

官方文档在 https://modelcontextprotocol.io/ ,代码仓库在 https://github.com/modelcontextprotocol 。云厂商和 SDK 更新快,遇到字段对不上先查官方文档。

写一个最小 Server:只读知识库查询

用 Python SDK 的 FastMCP 写一个工具。工具描述要短,参数要显式,不要把 SQL 直接暴露给模型。

from mcp.server.fastmcp import FastMCP

mcp = FastMCP('kb-server')

@mcp.tool() def search_knowledge(query: str, tenant_id: str, top_k: int = 5) -> list[dict]: '''按租户检索知识库,返回标题、片段、文档ID和来源链接。''' rows = kb.search(query, tenant_id=tenant_id, top_k=top_k) return [ { 'chunk_id': r.chunk_id, 'title': r.title, 'snippet': r.snippet, 'source_url': r.source_url, 'score': r.score, } for r in rows ]

if __name__ == '__main__': mcp.run(transport='streamable-http')

这里 tenant_id 必须由 Server 从 token 里取,不能交给模型填。模型能看到的参数越少,越不容易越权。

接知识库:RAG 在 Server 侧做,权限过滤在检索前

知识库接入常见错误是让 Agent 先拿全量文档再自己筛。企业数据量大,密级复杂,这样做既慢又容易泄漏。把 RAG 放在 MCP Server 内部:

  • 用户提问进入 Server。
  • 从 OAuth token 解析 user_id、tenant_id、部门、密级。
  • 向量检索时加 metadata filter,例如 tenant_id = 't_123' and acl_group in ('售后','客服')。
  • 返回 top_k 片段,附带 chunk_id 和 source_url。
  • Agent 回答时引用来源,审计日志记录 chunk_id。
工具返回结构建议固定:

字段类型用途
chunk_idstring审计与引用
titlestring展示来源
snippetstring给模型阅读
source_urlstring可点击原文
scorefloat判断是否要重排
如果知识库有版本,把 `doc_version` 也带上。用户问「2025 年政策」时,模型不会把 2024 年旧文档混进来。

接 CRM:工具要窄,身份要透传

CRM 一般有 REST API 和 OAuth 2.0。MCP Server 不要封装成 crm_query(sql),而是拆成窄工具:

  • crm_get_account(account_id)
  • crm_list_opportunities(account_id, status)
  • crm_get_tickets(account_id, since)
  • crm_create_note(account_id, content) 这类写操作要单独审批
窄工具的好处:参数少,模型不容易编;审计能看清调了什么;权限可以按工具粒度控制。

身份透传的做法:MCP Client 把用户 access token 放在 HTTP Header,Server 校验后,用同一个用户身份调 CRM API。不要用管理员账号。管理员账号一旦被提示注入利用,能拉全公司客户。

写操作要加三道:幂等键、二次确认、权限检查。CRM API 通常有限流,Server 侧加 Redis 令牌桶,返回 429 时告诉 Agent「稍后重试」,不要让模型无限重试。

接飞书:机器人、云文档和事件回调

飞书开放平台文档在 https://open.feishu.cn/ 。接入时区分两种 token:

token用途风险
tenant_access_token应用身份,发机器人消息、读云文档权限大,不能给 Agent
user_access_token用户身份,按用户权限读文档需要 OAuth 授权,会过期
事件订阅要配 Encrypt Key 和 Verification Token。回调 URL 必须验签,失败会返回 403。机器人发消息用 `im/v1/messages`,云文档搜索用 `drive/v1/files/search`。Scope 按最小权限申请,例如只申请 `docx:document:readonly` 和 `im:message:send_as_bot`。

飞书 Server 可以同时暴露资源与工具:把群聊记录作为 resource,把发消息作为 tool。群聊记录涉及隐私,只在用户 token 有对应群权限时返回。

企业权限:OAuth 2.1 + 租户隔离 + 最小 scope

企业落地的权限模型不要靠提示词约束。提示词能被绕过,token 校验不能。MCP Server 作为资源服务器,至少校验:

  • Authorization: Bearer 是否存在且未过期
  • token 的 audience 是否指向当前 MCP Server
  • token 的 scope 是否包含当前工具所需权限
  • tenant_id 是否与请求数据一致
  • user_id 是否在工具允许的组内
  • 写操作是否有单独审批标记
RBAC 管角色,ABAC 管属性。例如「售后主管」角色可以看客户手机号,但只能看自己团队的客户。ABAC 条件写成:department == user.department。这些规则放在 Server 的鉴权中间件,不放在工具描述里。

OAuth 2.1 推荐授权码 + PKCE。MCP Client 启动时做动态客户端注册,或由企业统一发 client_id。内部系统也可以走 mTLS + JWT,但都要有短期 token 和吊销机制。

RAG、工具调用与审计串起来

一个客服 Agent 的完整链路:

sequenceDiagram participant U as 用户 participant H as Host Agent participant C as MCP Client participant S as MCP Server participant K as 知识库 participant R as CRM U->>H: 这家客户上次投诉是什么 H->>C: 调用 crm_get_tickets C->>S: JSON-RPC + user token S->>S: 校验 tenant/部门/scope S->>R: 用用户身份查 CRM R-->>S: 工单列表 S-->>C: 返回工具结果 H->>C: 调用 search_knowledge C->>S: 带 tenant_id 和 query S->>K: metadata filter + 向量检索 K-->>S: 片段和来源 S-->>C: 返回引用片段 H-->>U: 汇总回答 + 来源

审计日志至少记这些字段:

字段示例说明
trace_id7f3a...串联一次会话
user_idu_1024谁发起
tenant_idt_123哪个租户
tool_namecrm_get_tickets调了什么
params_digestsha256:...参数摘要,不记敏感明文
result_count3返回行数
data_scopedept=售后数据范围
latency_ms184耗时
statusok/denied/error结果
日志脱敏:手机号、邮箱、身份证、token 都不进日志。需要排查时,用 trace_id 到原始系统拉取,并走审批。

部署:阿里云和腾讯云的最小示例

不用一上来就上 K8s。一个只读知识库 Server,用 ECS/CVM + Docker + HTTPS 负载均衡就够。等并发上来再换 ACK/TKE。

项目阿里云腾讯云
虚拟机ECS 突发性能实例 t6轻量应用服务器或 CVM
容器ACK 或直接 DockerTKE 或直接 Docker
负载均衡SLBCLB
日志SLSCLS
向量库RDS PostgreSQL + pgvector,或 DashVectorTencentDB for PostgreSQL + pgvector,或 VectorDB
密钥KMSKMS
函数计算函数计算 FCSCF
阿里云 ECS 快速验证:
  • 创建 ECS,安全组只开 443 和你的 SSH 端口。
  • 安装 Docker,拉取 MCP Server 代码。
  • 在 .env 写 OAUTH_ISSUER、CRM_BASE_URL、KB_ENDPOINT,密钥放 KMS 或环境变量。
  • docker build -t mcp-server:0.1 .
  • docker run -d --name mcp-server -p 8080:8080 --env-file .env mcp-server:0.1
  • 挂 SLB HTTPS,证书用阿里云 SSL,后端 8080。
  • 日志投递 SLS,开启请求级别 trace。
腾讯云 CVM 快速验证:
  • 创建 CVM,安全组只开 443 和 SSH。
  • 安装 Docker,部署同一个镜像。
  • 挂 CLB HTTPS,后端 8080。
  • 日志投递 CLS。
  • 如果用 SCF,注意它无状态、有执行时长上限,SSE 长连接不适合;短请求的 streamable HTTP 可以试,但要查 2026 年配额。
Dockerfile 可以很短:
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8080
CMD ['python', 'server.py']

生产还要加:非 root 用户运行、只读文件系统、出网白名单、请求体大小限制、每租户限流。

常见坑

坑现象处理
SSE 被网关断Agent 调到一半无响应换 streamable HTTP,或调大空闲超时
stdio 当远程用云上 Server 连不上stdio 只用于本地;远程走 HTTP
工具太多模型选错工具、上下文爆按业务域拆分 Server,每个 Server 少于 20 个工具
工具描述写小作文模型忽略关键参数描述只写用途、参数、返回,不写营销话术
用管理员 token数据越权每个请求透传用户 token,Server 校验 scope
审计只记 request出事查不到结果记 trace_id、tool_name、result_count、data_scope
知识库无 metadata检索到别的租户文档入库时写 tenant_id、acl_group、密级
飞书验签失败回调 403检查 Encrypt Key、Verification Token、URL 编码
CRM 限流大量 429Server 侧令牌桶 + 重试上限
JSON-RPC 错误码乱用Client 无法判断重试按协议返回 error code,业务错误放 data

下一步

选一个只读场景,例如「查产品文档」。用 FastMCP 包一个 search_knowledge,接上测试租户的 OAuth,把审计日志打到现有日志平台。跑通后再加 CRM 和飞书。不要先做写操作。

免责声明:本文示例用于本地验证,云厂商价格、配额和 API 字段以 2026 年官网为准。生产环境需要安全评审、数据分级和合规检查。官方文档:https://modelcontextprotocol.io/ ,飞书开放平台:https://open.feishu.cn/ 。

PREMIUM

需要完整版教程?

包含详细步骤、视频演示、提示词模板和可下载资料包。微信支付即时获取。

购买完整版 ¥29.90