开发者实战:用MCP+云函数搭建企业知识库Agent
客服主管常见的请求是:把售后文档、产品手册和工单记录做成一个问答入口,但销售部不能看到售后内部备注,外包客服不能看到价格策略。普通 RAG 容易漏掉权限:先把所有片段召回再让模型挑,片段已经进了上下文,越权就发生了。这套做法把检索和权限放在 MCP Server 里,Agent 只负责编排,云函数负责鉴权、审计和弹性。
环境准备
截至 2026 年,MCP 有 Python 和 TypeScript 两套官方 SDK。Python 包名 mcp,TypeScript 包名 @modelcontextprotocol/sdk。企业内网如果不能用公网模型,选通义千问、腾讯混元或本地 vLLM;能调外网再考虑 OpenAI/Claude。本地开发用 Docker Compose 起 Milvus 或 Chroma,生产换 DashVector/腾讯云向量数据库。
| 组件 | 可选 | 说明 |
|---|---|---|
| MCP Server | Python mcp / TS SDK | 暴露 tools,不生成答案 |
| 云函数 | 阿里云 FC 3.0 / 腾讯云 SCF | HTTP 触发,按量计费 |
| 向量库 | Milvus、Chroma、DashVector、腾讯云向量数据库 | 生产要支持 metadata filter |
| 模型 | 通义千问、腾讯混元、vLLM、OpenAI、Claude | 按合规和预算选 |
| 身份 | 企业微信、钉钉、飞书 SSO + JWT | 把 tenant_id、dept_ids 写进 token |
| 可观测 | OpenTelemetry + SLS/CLS | 日志先脱敏再落盘 |
- Python 3.11 + uv
- Node 20,仅在 MCP Server 用 TypeScript 时需要
- Docker
- 阿里云 FC 3.0 或腾讯云 SCF 账号
- 向量库实例,记下 endpoint 和 API key
- OIDC/JWT 签发端,企业微信/钉钉/飞书 SSO 都行
MCP Server开发
MCP Server 只做三件事:列出工具、执行检索、返回结构化片段。别把大模型塞进去,否则审计困难,换模型还要改 Server。传输选 Streamable HTTP。云函数对外是 HTTP,stdio 适合本地 IDE。MCP 的 tools/list 和 tools/call 都要带 Authorization 头。
from mcp.server.fastmcp import FastMCP
mcp = FastMCP('kb-server')
@mcp.tool()
async def search_knowledge(query: str, top_k: int = 5):
claims = get_jwt_claims()
if not claims:
raise PermissionError('missing token')
if top_k > 10:
top_k = 10
docs = vector_search(
query=query,
top_k=top_k,
filter={
'tenant_id': claims['tenant_id'],
'dept_id': {'$in': claims['dept_ids']},
'acl_tags': {'$in': claims['roles']},
},
)
return [
{
'doc_id': d['id'],
'title': d['title'],
'snippet': d['text'][:800],
'score': d['score'],
}
for d in docs
]
tenant_id 从 token 来,不让模型传。这是关键。模型只能决定搜什么词、要几条,不能决定看哪个租户的数据。
向量库与权限接入
文档入库时 metadata 必须完整。缺 tenant_id 的片段在生产环境直接拒绝入库。
| 字段 | 示例 | 用途 |
|---|---|---|
| tenant_id | t_1001 | 租户隔离 |
| dept_id | dept_after_sales | 部门过滤 |
| acl_tags | ['internal','price'] | 角色/标签过滤 |
| doc_version | 2026-03-01 | 避免旧版本 |
| source_url | 内部 wiki 地址 | 审计溯源 |
权限校验至少两层:API 网关验签,MCP Server 解析 JWT 再查一次。网关挡住外部调用,MCP Server 挡住内部误调。文档里出现“忽略系统指令”时,不要把它拼进系统提示。把检索结果放在 user 消息或 tool 消息里,并声明只作资料。
Function Calling编排
Agent 云函数接收问题,从 MCP Server 拉 tools,转成模型 function schema。模型返回 tool_calls 后,Agent 调 MCP,再把结果回传。
tools = await mcp_client.list_tools()
messages = [{'role': 'user', 'content': question}]
for _ in range(5):
resp = model.chat(messages, tools=tools)
if not resp.tool_calls:
return resp.content
for call in resp.tool_calls:
if call.name != 'search_knowledge':
continue
args = call.arguments
args.pop('tenant_id', None)
result = await mcp_client.call_tool(call.name, args)
messages.append({
'role': 'tool',
'tool_call_id': call.id,
'content': json.dumps(result, ensure_ascii=False),
})
return '工具调用次数超限'
限制循环 5 次,top_k 最大 10,query 最长 500 字。腾讯云 SCF 事件函数最大执行超时 900 秒,Agent 循环别拖太久;阿里云 FC 可配更长超时,但也要设 60 秒业务超时。超时后返回已查到的片段,不要无限重试。
可观测性与安全审计
日志先脱敏再落盘。记 trace_id、user_id、tenant_id、tool_name、query_hash、doc_ids、latency_ms、status。不要记用户问题原文和文档全文。
| 事件 | 必须字段 | 告警 |
|---|---|---|
| 检索 | trace_id,user_id,tenant_id,query_hash,doc_ids | 单用户 1 分钟 > 30 次 |
| 越权 | user_id, tenant_id, 请求 filter | 立即告警 |
| 工具失败 | tool_name, error_code, latency_ms | 5 分钟失败率 > 5% |
| 输出拦截 | user_id, rule_id | 记审计 |
- API 网关开 JWT 校验,请求体限制 1MB
- 云函数角色只读指定向量库和 KMS 密钥
- 向量库 filter 在服务端执行
- 用户输入和文档片段做提示注入检测
- 输出过滤手机号、身份证、API key
- 每月做一次越权检索测试
部署到阿里云或腾讯云
| 环节 | 阿里云 | 腾讯云 |
|---|---|---|
| 函数 | FC 3.0,容器镜像 | SCF,事件函数/容器 |
| 网关 | API 网关 | API 网关 |
| 向量库 | DashVector/百炼知识库 | 腾讯云向量数据库 |
| 日志 | SLS | CLS |
| 权限 | RAM | CAM |
| 密钥 | KMS | KMS |
| 超时注意 | FC 可配到 86400 秒,按控制台为准 | SCF 事件函数最大 900 秒 |
| 计费 | 按量,查官网 | 按量,查官网 |
- 构建 MCP Server 镜像,推 ACR/TCR
- 创建云函数,配 VPC、环境变量、函数角色
- 配 API 网关触发,开启 JWT 鉴权
- 绑定向量库 endpoint、KMS 密钥
- 调 tools/list 和 tools/call,确认权限过滤生效
- 压测 10 并发,看冷启动和超时
下一步:拿 20 篇内部文档,建一个测试向量集合,只写 tenant_id 和 dept_id。用 A 部门 token 搜 B 部门关键词,确认返回空列表,同时审计日志出现越权拦截。检查项:如果你的 MCP Server 日志里出现完整用户问题,先改日志脱敏,再上线。