开发者实战:用MCP+云函数搭建企业知识库Agent

开发者教程MCP云函数企业知识库Agent向量数据库Function Calling安全审计2026-10-05

开发者实战:用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 ServerPython mcp / TS SDK暴露 tools,不生成答案
云函数阿里云 FC 3.0 / 腾讯云 SCFHTTP 触发,按量计费
向量库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_idt_1001租户隔离
dept_iddept_after_sales部门过滤
acl_tags['internal','price']角色/标签过滤
doc_version2026-03-01避免旧版本
source_url内部 wiki 地址审计溯源
检索过滤要在向量库服务端执行。Chroma、Milvus、DashVector、腾讯云向量数据库都支持 metadata filter,语法不同。不要先拉全量再在 Python 里过滤。用户是 A 部门,查询条件就带 `tenant_id = 当前租户` 和 `dept_id IN 用户部门`,让不匹配的片段根本不返回。

权限校验至少两层:API 网关验签,MCP Server 解析 JWT 再查一次。网关挡住外部调用,MCP Server 挡住内部误调。文档里出现“忽略系统指令”时,不要把它拼进系统提示。把检索结果放在 user 消息或 tool 消息里,并声明只作资料。

Function Calling编排

Agent 云函数接收问题,从 MCP Server 拉 tools,转成模型 function schema。模型返回 tool_calls 后,Agent 调 MCP,再把结果回传。

flowchart LR U[用户] --> GW[API 网关] GW --> A[Agent 云函数] A --> M[模型] M -->|tool_calls| A A --> S[MCP Server] S --> V[向量库] V --> S S --> A A --> M M --> A A --> U
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_ms5 分钟失败率 > 5%
输出拦截user_id, rule_id记审计
OpenTelemetry 接云函数。阿里云导出到 SLS,腾讯云导出到 CLS。保留天数按合规要求,常见 180 天。
  • API 网关开 JWT 校验,请求体限制 1MB
  • 云函数角色只读指定向量库和 KMS 密钥
  • 向量库 filter 在服务端执行
  • 用户输入和文档片段做提示注入检测
  • 输出过滤手机号、身份证、API key
  • 每月做一次越权检索测试

部署到阿里云或腾讯云

环节阿里云腾讯云
函数FC 3.0,容器镜像SCF,事件函数/容器
网关API 网关API 网关
向量库DashVector/百炼知识库腾讯云向量数据库
日志SLSCLS
权限RAMCAM
密钥KMSKMS
超时注意FC 可配到 86400 秒,按控制台为准SCF 事件函数最大 900 秒
计费按量,查官网按量,查官网
  • 构建 MCP Server 镜像,推 ACR/TCR
  • 创建云函数,配 VPC、环境变量、函数角色
  • 配 API 网关触发,开启 JWT 鉴权
  • 绑定向量库 endpoint、KMS 密钥
  • 调 tools/list 和 tools/call,确认权限过滤生效
  • 压测 10 并发,看冷启动和超时
MCP Server 如果用 SSE 长连接,云函数会被超时打断。优先 Streamable HTTP,或把 MCP Server 放容器服务/ACK/TKE,云函数只跑 Agent 编排。费用按调用次数、资源使用和公网流量算。2026 年各厂商免费额度和单价可能调整,上线前看计费页。

下一步:拿 20 篇内部文档,建一个测试向量集合,只写 tenant_id 和 dept_id。用 A 部门 token 搜 B 部门关键词,确认返回空列表,同时审计日志出现越权拦截。检查项:如果你的 MCP Server 日志里出现完整用户问题,先改日志脱敏,再上线。

PREMIUM

需要完整版教程?

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

购买完整版 ¥29.90