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:预置提示模板,比如「生成客户回访摘要」。
| 传输 | 场景 | 限制 |
|---|---|---|
| stdio | 本地工具、IDE | 不能跨网络,Server 随 Host 启动 |
| SSE | 远程推送 | 长连接容易被网关断开,2026 年新项目优先 streamable HTTP |
| streamable HTTP | 远程 Server、云部署 | 需要自己做鉴权、限流、审计 |
官方文档在 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_id | string | 审计与引用 |
| title | string | 展示来源 |
| snippet | string | 给模型阅读 |
| source_url | string | 可点击原文 |
| score | float | 判断是否要重排 |
接 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 授权,会过期 |
飞书 Server 可以同时暴露资源与工具:把群聊记录作为 resource,把发消息作为 tool。群聊记录涉及隐私,只在用户 token 有对应群权限时返回。
企业权限:OAuth 2.1 + 租户隔离 + 最小 scope
企业落地的权限模型不要靠提示词约束。提示词能被绕过,token 校验不能。MCP Server 作为资源服务器,至少校验:
- Authorization: Bearer 是否存在且未过期
- token 的 audience 是否指向当前 MCP Server
- token 的 scope 是否包含当前工具所需权限
- tenant_id 是否与请求数据一致
- user_id 是否在工具允许的组内
- 写操作是否有单独审批标记
department == user.department。这些规则放在 Server 的鉴权中间件,不放在工具描述里。
OAuth 2.1 推荐授权码 + PKCE。MCP Client 启动时做动态客户端注册,或由企业统一发 client_id。内部系统也可以走 mTLS + JWT,但都要有短期 token 和吊销机制。
RAG、工具调用与审计串起来
一个客服 Agent 的完整链路:
审计日志至少记这些字段:
| 字段 | 示例 | 说明 |
|---|---|---|
| trace_id | 7f3a... | 串联一次会话 |
| user_id | u_1024 | 谁发起 |
| tenant_id | t_123 | 哪个租户 |
| tool_name | crm_get_tickets | 调了什么 |
| params_digest | sha256:... | 参数摘要,不记敏感明文 |
| result_count | 3 | 返回行数 |
| data_scope | dept=售后 | 数据范围 |
| latency_ms | 184 | 耗时 |
| status | ok/denied/error | 结果 |
部署:阿里云和腾讯云的最小示例
不用一上来就上 K8s。一个只读知识库 Server,用 ECS/CVM + Docker + HTTPS 负载均衡就够。等并发上来再换 ACK/TKE。
| 项目 | 阿里云 | 腾讯云 |
|---|---|---|
| 虚拟机 | ECS 突发性能实例 t6 | 轻量应用服务器或 CVM |
| 容器 | ACK 或直接 Docker | TKE 或直接 Docker |
| 负载均衡 | SLB | CLB |
| 日志 | SLS | CLS |
| 向量库 | RDS PostgreSQL + pgvector,或 DashVector | TencentDB for PostgreSQL + pgvector,或 VectorDB |
| 密钥 | KMS | KMS |
| 函数计算 | 函数计算 FC | SCF |
- 创建 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,安全组只开 443 和 SSH。
- 安装 Docker,部署同一个镜像。
- 挂 CLB HTTPS,后端 8080。
- 日志投递 CLS。
- 如果用 SCF,注意它无状态、有执行时长上限,SSE 长连接不适合;短请求的 streamable HTTP 可以试,但要查 2026 年配额。
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 限流 | 大量 429 | Server 侧令牌桶 + 重试上限 |
| JSON-RPC 错误码乱用 | Client 无法判断重试 | 按协议返回 error code,业务错误放 data |
下一步
选一个只读场景,例如「查产品文档」。用 FastMCP 包一个 search_knowledge,接上测试租户的 OAuth,把审计日志打到现有日志平台。跑通后再加 CRM 和飞书。不要先做写操作。
免责声明:本文示例用于本地验证,云厂商价格、配额和 API 字段以 2026 年官网为准。生产环境需要安全评审、数据分级和合规检查。官方文档:https://modelcontextprotocol.io/ ,飞书开放平台:https://open.feishu.cn/ 。