尧图网站建设 
1 项目背景业务场景「云帆科技」的 Agent 工作流已经能处理制度检索和报销计算。但业务部门提出了越来越多的外部集成需求财务部想在工作流中直接查询 SQL 数据库获取员工的最新报销额度行政部想调用内部工单系统 API 查看设备维修进度HR 想对接企业微信/飞书的消息通道实现机器人主动推送。最让开发团队头疼的是集成方式不统一——有的外部系统是 REST API有的是 gRPC有的是直接连数据库有的只有 Python SDK。运维小李希望能有一个统一的工具接入标准不用为每个外部系统单独写适配代码。就在这时Anthropic 发布了 MCPModel Context Protocol——一个让 LLM Agent 标准化发现和调用外部工具的开放协议。RAGFlow 也跟进了 MCP 支持。痛点没有统一工具集成框架时的困境每个工具都是一次性开发调 SQL 写一套代码调工单 API 又写一套调邮件服务再写一套——代码碎片化且不可复用。工具发现靠文档Agent 不知道我能用什么工具需要开发者手动注册、手动描述——随着工具增多维护成本线性增长。工具调用的安全检查各自为战有的工具有参数校验有的没有有的有超时控制有的没有——安全漏洞无处不在。工具结果格式不统一有的返回 JSON有的返回纯文本有的返回二进制——LLM 无法统一理解。统一工具框架解决的核心问题 Before无 MCP: LLM Agent → 开发者硬编码每个工具的调用方式 成本: 10个工具 10套独立代码 10套安全配置 After用 MCP: LLM Agent → MCP Client → MCP Server (统一发现调用安全) 成本: 工具只需实现 MCP Server 接口Agent 自动发现2 项目设计小胖在 RAGFlow 的工具配置页面翻来翻去“大师我想让 Agent 回答’查询我的报销额度’时自动调 SQL 查数据库还想让它帮我在工单系统里建一个报修单。但每个系统接口都不一样我难道要学 10 种 API”大师“以前确实是这样的——每个外部系统都要写专门的适配代码。但现在有了 MCPModel Context Protocol它相当于给所有工具定了统一的’插座标准’——不管你背后的系统是 SQL 数据库、REST API、还是 gRPC 服务只要实现了 MCP Server 接口Agent 就能用统一的方式发现你、调用你。”小胖“MCP 是什么和 Function Calling 有什么区别”大师“Function Calling 是 LLM 层面的——大模型知道了’有这个函数可以调’。MCP 是架构层面的——定义了工具怎么被发现、怎么被调用、怎么处理错误、怎么传输数据。两者的关系和 OSI 七层模型类似——Function Calling 在应用层告诉 LLM 有什么能力MCP 在传输和表示层定义标准化的调用协议。”Function Calling vs MCP OpenAI Function Calling: └── 只定义了 LLM ↔ 开发者之间的协议 开发者定义 function schema → LLM 决定调用哪个 → 开发者执行 → 返回结果给 LLM 局限工具的发现、调用、安全都由开发者自行管理 MCP (Model Context Protocol): └── 定义了 Client ↔ Server 之间的完整协议 - 工具发现: tools/list → 返回所有可用工具及参数 schema - 工具调用: tools/call → 传入参数返回结果 - 资源访问: resources/read → 读取文件、数据库等 - 提示模板: prompts/get → 获取预置 Prompt 优势Client 无需提前知道有什么工具运行时动态发现技术映射Function Calling 餐厅菜单列出有什么菜MCP 餐厅的后厨标准操作流程怎么下单、怎么做菜、怎么上菜、怎么处理退菜——标准化了整个流程而不仅仅是菜单。小胖“那 RAGFlow 里怎么接入 MCP内置工具和 MCP 工具有什么关系”大师“RAGFlow 的工具体系分为两层”RAGFlow 工具体系 第一层内置工具agent/tools/ ├── RetrievalTool: 在数据集中检索 ├── WebSearchTool: 互联网搜索 ├── SQLTool: 执行 SQL 查询 ├── EmailTool: 发送邮件 ├── GitHubTool: 操作 GitHub 仓库 ├── FinanceTool: 获取金融数据 ├── TranslationTool: 翻译 └── WeatherTool: 天气查询 第二层MCP 工具agent/plugin/ ├── MCPClient: 与 MCP Server 通信的客户端 ├── 自动发现: 连接 MCP Server → 获取工具列表 └── 标准化调用: 统一 tools/call 接口小白“那如果我想接入公司自己的工单系统内部 REST API该怎么做”大师“三个步骤无论你是用内置方式还是 MCP 方式”# 方式1自定义内置 Tool继承 ToolBasefromagent.tools.baseimportToolBaseclassTicketSystemTool(ToolBase):公司内部工单系统查询工具nameticket_systemdescription查询工单系统中的工单状态、创建工单、更新工单# 定义参数 schemaOpenAI function calling 格式parameters{type:object,properties:{action:{type:string,enum:[query,create,update],description:操作类型},ticket_id:{type:string,description:工单IDquery/update时需要},title:{type:string,description:工单标题create时需要},},required:[action]}asyncdef_call(self,action,ticket_idNone,titleNone,**kwargs):执行工具调用importhttpx headers{Authorization:fBearer{self.api_key}}ifactionquery:asyncwithhttpx.AsyncClient()asclient:respawaitclient.get(f{self.base_url}/tickets/{ticket_id},headersheaders,timeout30)returnresp.json()elifactioncreate:asyncwithhttpx.AsyncClient()asclient:respawaitclient.post(f{self.base_url}/tickets,headersheaders,json{title:title},timeout30)returnresp.json()elifactionupdate:# ... 更新逻辑pass# 方式2实现 MCP Server更标准化的方式# 工单系统 MCP Server 核心代码frommcp.serverimportServer,NotificationOptionsfrommcp.server.modelsimportInitializationCapabilities serverServer(ticket-system-mcp)server.list_tools()asyncdefhandle_list_tools():MCP 协议列出所有可用工具return[Tool(namequery_ticket,description查询工单状态,inputSchema{type:object,properties:{ticket_id:{type:string}}}),Tool(namecreate_ticket,description创建工单,inputSchema{type:object,properties:{title:{type:string}}}),]server.call_tool()asyncdefhandle_call_tool(name:str,arguments:dict):MCP 协议调用指定工具ifnamequery_ticket:resultawaitquery_ticket_internal(arguments[ticket_id])return[TextContent(typetext,textjson.dumps(result))]elifnamecreate_ticket:resultawaitcreate_ticket_internal(arguments[title])return[TextContent(typetext,textjson.dumps(result))]技术映射自定义 Tool 自己造一个带特定接口的插座MCP Server 用国标插座标准任何符合国标的插头都能插进来。小胖“那这些工具在 Canvas 里怎么用”大师“在 Canvas 中添加一个Tool组件选择你想要调用的工具。Agent 会在执行时自动决策——根据用户的问题判断要不要调工具、调哪个工具、用什么参数——无需手动在 Canvas 上连线。这就是自主 Agent模式Agent 自己规划→执行→观察→再规划直到完成任务。”小白“安全方面呢工具调用的权限、审计、限制怎么搞”大师“工具体系的安全四要素”工具调用安全四要素 1. 参数校验Parameter Validation 每个工具定义的 parameters schema 就是第一道防线 RAGFlow 在调用前校验参数类型、必填项、枚举值 2. 权限隔离Permission Isolation 不同用户/租户可用的工具集合不同 工单系统普通员工只能 query管理员可以 create/update 3. 超时控制Timeout Control 每个工具调用有独立的超时时间 查询类工具5-10 秒创建类工具15-30 秒 4. 审计日志Audit Log 每次工具调用记录谁、什么时候、调了什么工具、传了什么参数、返回了什么 用于安全审计和调试回溯3 项目实战环境准备目标实现一个 Agent先检索制度再调 SQL 查员工报销额度最后生成审批建议。前提RAGFlow 已部署SQL 数据库可访问含员工信息表Canvas 可用。分步实现步骤1配置工具体系目标在 RAGFlow 系统设置中启用并配置 SQL Tool 和自定义邮件 Tool。# 1. 配置 SQL Tool内置curl-XPOST http://localhost/api/v1/tools/configure\-HAuthorization: Bearer$TOKEN\-HContent-Type: application/json\-d{ tool_name: sql_query, config: { db_type: mysql, host: mysql, port: 3306, database: yunfan_hr, username: readonly_user, password: ReadOnly2024, max_rows: 100, query_timeout: 10 } }# 2. 配置 Email Tool内置curl-XPOST http://localhost/api/v1/tools/configure\-HAuthorization: Bearer$TOKEN\-HContent-Type: application/json\-d{ tool_name: send_email, config: { smtp_host: smtp.yunfan.com, smtp_port: 587, username: ragflow-botyunfan.com, password: BotEmail2024, from_name: 云帆报销助手 } }步骤2连接 MCP Server目标启动一个 MCP Server 并注册到 RAGFlow。# 启动工单系统 MCP Server示例# 方式通过 npx 或 python 启动npx-ymodelcontextprotocol/server-ticket-system\--api-url https://ticket.yunfan.com/api\--api-key$TICKET_API_KEY\--port3001# 在 RAGFlow 中注册该 MCP Servercurl-XPOST http://localhost/api/v1/tools/mcp/register\-HAuthorization: Bearer$TOKEN\-HContent-Type: application/json\-d{ name: 工单系统, transport: streamable_http, url: http://ticket-mcp:3001, description: 公司内部工单系统支持查询、创建、更新工单 }坑点MCP 支持两种传输方式——stdio进程间通信和streamable_http网络通信。Docker 容器之间用streamable_http。步骤3创建工具增强型 Agent Canvas目标搭建一个完整的报销咨询→额度查询→审批建议→邮件发送工作流。# step3_agent_with_tools.py# 创建 Agent 型 Canvas让 LLM 自主决策调哪些工具canvasrag.create_canvas(name智能报销审批助手)# Begin 入口begincanvas.add_component(Begin,idbegin)# Agent 组件自主规划工具调用agentcanvas.add_component(Agent,idagent_main,config{model:gpt-4o,temperature:0.1,max_iterations:10,# 最多调用 10 次工具available_tools:[retrieval,# RAG 检索sql_query,# SQL 查询send_email,# 发送邮件ticket_system,# 工单系统来自 MCP],system_prompt:你是一个智能报销审批助手。处理逻辑 1. 首先在知识库中检索相关报销制度 2. 如果用户提到具体员工用SQL查询该员工的报销额度信息 3. 综合制度标准和员工额度给出报销建议 4. 如果需要人工审批创建工单并发送邮件通知 重要涉及数据库查询时务必只查询、不修改。 涉及发送邮件时务必先确认收件人邮箱。,})# End 出口endcanvas.add_component(End,idend)canvas.connect(begin,agent)canvas.connect(agent,end)canvas.publish()步骤4工具调用链追踪目标观察 Agent 在一次复杂任务中调用了哪些工具及其顺序。# step4_trace_tool_calls.pyimportjson# 测试问题test_input查询张三的报销额度帮他算一下出差北京5天能报销多少然后发邮件通知他。sessioncanvas.test_run(inputs{query:test_input})print( Agent 工具调用链 )fori,stepinenumerate(session.execution_log):ifstep.component_idagent_main:print(f\n--- Agent 迭代{step.iteration}---)ifstep.action_typetool_call:print(f 工具调用:{step.tool_name})print(f 参数:{json.dumps(step.tool_args,ensure_asciiFalse)})print(f 结果:{str(step.tool_result)[:200]}...)elifstep.action_typellm_thought:print(f LLM 思考:{step.thought[:200]}...)elifstep.action_typefinal_answer:print(f 最终回答:{step.output[:300]}...)预期输出 Agent 工具调用链 --- Agent 迭代 1 --- LLM 思考: 用户想查询张三的报销额度并计算出差费用。我需要先检索报销制度再查数据库。 工具调用: retrieval 参数: {query: 出差住宿费标准 北京} 结果: 根据差旅费管理办法一线城市北京住宿标准为500元/晚... --- Agent 迭代 2 --- 工具调用: sql_query 参数: {query: SELECT name, reimbursement_limit, used_amount FROM employees WHERE name LIKE %张三%} 结果: {rows: [{name: 张三, reimbursement_limit: 10000, used_amount: 3000}]} --- Agent 迭代 3 --- LLM 思考: 张三剩余额度7000元。北京出差5天住宿费5×5002500元。25007000额度足够。 工具调用: send_email 参数: {to: zhangsanyunfan.com, subject: 出差报销计算结果, body: 张三你好你的出差北京5天预计报销2500元...} --- Agent 迭代 4 --- LLM 思考: 任务完成。已检索制度、查询额度、发送通知。准备最终回答。 最终回答: 张三您本次出差北京的预估报销如下住宿费 2500元5晚×500元/晚交通补助等详见邮件。您的年度报销额度剩余7000元本次报销后额度充足。详细计算结果已发送至您的邮箱 zhangsanyunfan.com。步骤5工具调用的安全与监控目标配置工具的审计日志和调用限制。# step5_tool_audit.py - 审计与监控classToolAuditLogger:工具调用审计日志deflog_call(self,user_id,tenant_id,tool_name,args,result,duration_ms,success):记录每次工具调用importjson,datetime record{timestamp:datetime.datetime.now().isoformat(),user_id:user_id,tenant_id:tenant_id,tool_name:tool_name,args:args,result_preview:str(result)[:200]ifresultelseNone,duration_ms:duration_ms,success:success,}# 写入审计日志表# AuditLog.create(**record)print(f[AUDIT]{tool_name}by{user_id}:{OKifsuccesselseFAILED}({duration_ms}ms))# 工具调用频率限制classToolRateLimiter:防止工具被滥用def__init__(self):self.limits{sql_query:{per_minute:30,per_user_per_minute:5},send_email:{per_minute:10,per_user_per_minute:2},ticket_system:{per_minute:20,per_user_per_minute:3},}self.countersdefaultdict(lambda:defaultdict(int))defcheck_and_increment(self,tool_name,user_id):检查是否超过限流阈值未超过则记录一次调用limitsself.limits.get(tool_name,{per_minute:10})current_minuteint(time.time()/60)# 全局限制global_keyf{tool_name}:{current_minute}user_keyf{tool_name}:{user_id}:{current_minute}ifself.counters[global_key][count]limits.get(per_minute,10):raiseException(f工具{tool_name}全局限流已达上限)ifself.counters[user_key][count]limits.get(per_user_per_minute,3):raiseException(f用户{user_id}对工具{tool_name}的调用频率已达上限)self.counters[global_key][count]1self.counters[user_key][count]1returnTrue测试验证# test_tool_integration.pyclassTestToolIntegration:deftest_sql_tool_returns_data(self):验证 SQL 工具可正常查询resulttool_registry.call(sql_query,{query:SELECT COUNT(*) AS cnt FROM employees})assertresult[rows][0][cnt]0deftest_agent_chooses_correct_tool(self):验证 Agent 为不同问题选择正确工具# 制度类问题 → 应选 retrievalresponseagent_chat(年假有几天)tool_callsextract_tool_calls(response)assertany(retrievalintcfortcintool_calls)# 数据查询类 → 应选 sql_queryresponseagent_chat(张三的报销额度是多少)tool_callsextract_tool_calls(response)assertany(sql_queryintcfortcintool_calls)deftest_tool_rate_limiting(self):验证限流生效limiterToolRateLimiter()# 连续调用 6 次超过用户每分钟限制 5 次for_inrange(5):limiter.check_and_increment(sql_query,test_user)withpytest.raises(Exception,match频率已达上限):limiter.check_and_increment(sql_query,test_user)完整代码清单路径说明agent/tools/base.pyToolBase 基类参数校验、超时、审计agent/tools/sql.pySQL 查询工具agent/tools/email.py邮件发送工具agent/tools/retrieval.py检索工具agent/plugin/MCP 客户端与集成agent/canvas.pyCanvas 中 Tool 组件的执行逻辑4 项目总结优点 缺点维度RAGFlow 工具MCPLangChain ToolsOpenAI Function Calling自建工具框架工具标准化★★★ MCP 协议★★☆ Tool 基类★★★ Function schema★☆☆ 需自行定义运行时发现★★★ MCP 自动发现★★☆ 手动注册★☆☆ 需代码注册★☆☆ 需开发RAG 深度集成★★★ 原生检索工具★★★ 可组合★☆☆ 需自行实现★★☆ 需集成Canvas 可视化★★★ 拖拽式编排★☆☆ 代码编排★☆☆ 无★☆☆ 需开发安全管控★★☆ 基本审计限流★★☆ 回调控制★☆☆ 依赖平台★★★ 完全可控学习曲线★★☆ 中等★★☆ 中等★★★ 较简单★☆☆ 高适用场景数据驱动的业务 Agent检索制度 SQL 查数据 外部 API 调用——如本章的报销审批。企业内部系统集成通过 MCP 标准化接入工单系统、OA、CRM、ERP 等。多工具协同Agent 自主选择最优工具组合完成复杂任务。低代码扩展非开发人员通过 MCP Server 模板快速接入新系统。工具集市按部门/租户发布可用工具建成企业工具库。不适用场景极简单的 RAG 问答不需要工具调用纯检索即可。无外部系统依赖的封闭环境如果 Agent 只访问内部知识库不需要工具体系。注意事项SQL Tool 的只读限制强烈建议 SQL 工具连接只读副本或限制只能执行 SELECT 语句。绝不能给 Agent DELETE/DROP 权限。MCP Server 的传输安全streamable_http方式如果部署在公网必须开启 TLS 鉴权。stdio方式更安全但仅限于同一台机器。工具描述的精确性工具的description和参数description直接影响 LLM 的决策——写得模糊 LLM 就会选错工具。审计日志的存储与保留工具调用涉及敏感的数据库查询和邮件发送审计日志建议保留至少 90 天。Agent 无限循环如果 LLM 反复调同一个工具且每次都说再试一次可能陷入死循环——max_iterations是兜底。常见踩坑经验故障现象根因解决方法LLM 调了错误的工具工具名/描述与用户意图的语义不匹配优化工具的 description用更多同义词SQL 工具返回数据太大导致超时SELECT * 返回数万行设置max_rows100Agent Prompt 中要求加 LIMITMCP Server 断连后 Agent 一直重试无连接健康检查Agent 不知道工具不可用MCP Client 侧加 health check不可用工具从列表移除工具参数校验老是失败LLM 输出的参数格式不完全符合 schema宽容解析——对 LLM 输出的参数做兼容性处理如 “北京” → [“北京”]审计日志被恶意用户刷爆无日志采样和限流对单个用户的审计日志做限流采样每用户每分钟最多 10 条思考题公司的数据库中有 200 张表。如果让 Agent 直接看到全部 200 张表LLM 会难以选择正确的表和字段。请设计一个分层工具注册方案——第一层只有 10 个业务视图工具查询员工信息、查询报销记录、查询部门预算…第二层的具体 SQL 由视图工具内部生成Agent 永远看不到底层表结构。如果一条用户消息触发 Agent 调用了 5 个工具检索→SQL→计算→邮件→工单其中第 4 步发邮件时失败了SMTP 服务器宕机。请设计一个Saga 补偿方案——将已执行的操作回滚或补偿如撤销工单、标记计算结果为无效、通知用户部分操作失败。答案提示见第26章末尾或附录 D。延伸阅读与资源10倍开发者的 Dify 魔法书从零构建全栈 AI 应用后端工程师转型AI第一课-Ollama 与私有化大模型实战大型语言模型(LLM) vLLM 高性能推理落地实战Agent开发之LlamaIndex 实战修炼与源码进阶大语言模型Transformers 实战修炼与源码剖析 第25章：RAGFlow 工具调用、MCP 与外部系统集成
1 项目背景业务场景「云帆科技」的 Agent 工作流已经能处理制度检索和报销计算。但业务部门提出了越来越多的外部集成需求财务部想在工作流中直接查询 SQL 数据库获取员工的最新报销额度行政部想调用内部工单系统 API 查看设备维修进度HR 想对接企业微信/飞书的消息通道实现机器人主动推送。最让开发团队头疼的是集成方式不统一——有的外部系统是 REST API有的是 gRPC有的是直接连数据库有的只有 Python SDK。运维小李希望能有一个统一的工具接入标准不用为每个外部系统单独写适配代码。就在这时Anthropic 发布了 MCPModel Context Protocol——一个让 LLM Agent 标准化发现和调用外部工具的开放协议。RAGFlow 也跟进了 MCP 支持。痛点没有统一工具集成框架时的困境每个工具都是一次性开发调 SQL 写一套代码调工单 API 又写一套调邮件服务再写一套——代码碎片化且不可复用。工具发现靠文档Agent 不知道我能用什么工具需要开发者手动注册、手动描述——随着工具增多维护成本线性增长。工具调用的安全检查各自为战有的工具有参数校验有的没有有的有超时控制有的没有——安全漏洞无处不在。工具结果格式不统一有的返回 JSON有的返回纯文本有的返回二进制——LLM 无法统一理解。统一工具框架解决的核心问题 Before无 MCP: LLM Agent → 开发者硬编码每个工具的调用方式 成本: 10个工具 10套独立代码 10套安全配置 After用 MCP: LLM Agent → MCP Client → MCP Server (统一发现调用安全) 成本: 工具只需实现 MCP Server 接口Agent 自动发现2 项目设计小胖在 RAGFlow 的工具配置页面翻来翻去“大师我想让 Agent 回答’查询我的报销额度’时自动调 SQL 查数据库还想让它帮我在工单系统里建一个报修单。但每个系统接口都不一样我难道要学 10 种 API”大师“以前确实是这样的——每个外部系统都要写专门的适配代码。但现在有了 MCPModel Context Protocol它相当于给所有工具定了统一的’插座标准’——不管你背后的系统是 SQL 数据库、REST API、还是 gRPC 服务只要实现了 MCP Server 接口Agent 就能用统一的方式发现你、调用你。”小胖“MCP 是什么和 Function Calling 有什么区别”大师“Function Calling 是 LLM 层面的——大模型知道了’有这个函数可以调’。MCP 是架构层面的——定义了工具怎么被发现、怎么被调用、怎么处理错误、怎么传输数据。两者的关系和 OSI 七层模型类似——Function Calling 在应用层告诉 LLM 有什么能力MCP 在传输和表示层定义标准化的调用协议。”Function Calling vs MCP OpenAI Function Calling: └── 只定义了 LLM ↔ 开发者之间的协议 开发者定义 function schema → LLM 决定调用哪个 → 开发者执行 → 返回结果给 LLM 局限工具的发现、调用、安全都由开发者自行管理 MCP (Model Context Protocol): └── 定义了 Client ↔ Server 之间的完整协议 - 工具发现: tools/list → 返回所有可用工具及参数 schema - 工具调用: tools/call → 传入参数返回结果 - 资源访问: resources/read → 读取文件、数据库等 - 提示模板: prompts/get → 获取预置 Prompt 优势Client 无需提前知道有什么工具运行时动态发现技术映射Function Calling 餐厅菜单列出有什么菜MCP 餐厅的后厨标准操作流程怎么下单、怎么做菜、怎么上菜、怎么处理退菜——标准化了整个流程而不仅仅是菜单。小胖“那 RAGFlow 里怎么接入 MCP内置工具和 MCP 工具有什么关系”大师“RAGFlow 的工具体系分为两层”RAGFlow 工具体系 第一层内置工具agent/tools/ ├── RetrievalTool: 在数据集中检索 ├── WebSearchTool: 互联网搜索 ├── SQLTool: 执行 SQL 查询 ├── EmailTool: 发送邮件 ├── GitHubTool: 操作 GitHub 仓库 ├── FinanceTool: 获取金融数据 ├── TranslationTool: 翻译 └── WeatherTool: 天气查询 第二层MCP 工具agent/plugin/ ├── MCPClient: 与 MCP Server 通信的客户端 ├── 自动发现: 连接 MCP Server → 获取工具列表 └── 标准化调用: 统一 tools/call 接口小白“那如果我想接入公司自己的工单系统内部 REST API该怎么做”大师“三个步骤无论你是用内置方式还是 MCP 方式”# 方式1自定义内置 Tool继承 ToolBasefromagent.tools.baseimportToolBaseclassTicketSystemTool(ToolBase):公司内部工单系统查询工具nameticket_systemdescription查询工单系统中的工单状态、创建工单、更新工单# 定义参数 schemaOpenAI function calling 格式parameters{type:object,properties:{action:{type:string,enum:[query,create,update],description:操作类型},ticket_id:{type:string,description:工单IDquery/update时需要},title:{type:string,description:工单标题create时需要},},required:[action]}asyncdef_call(self,action,ticket_idNone,titleNone,**kwargs):执行工具调用importhttpx headers{Authorization:fBearer{self.api_key}}ifactionquery:asyncwithhttpx.AsyncClient()asclient:respawaitclient.get(f{self.base_url}/tickets/{ticket_id},headersheaders,timeout30)returnresp.json()elifactioncreate:asyncwithhttpx.AsyncClient()asclient:respawaitclient.post(f{self.base_url}/tickets,headersheaders,json{title:title},timeout30)returnresp.json()elifactionupdate:# ... 更新逻辑pass# 方式2实现 MCP Server更标准化的方式# 工单系统 MCP Server 核心代码frommcp.serverimportServer,NotificationOptionsfrommcp.server.modelsimportInitializationCapabilities serverServer(ticket-system-mcp)server.list_tools()asyncdefhandle_list_tools():MCP 协议列出所有可用工具return[Tool(namequery_ticket,description查询工单状态,inputSchema{type:object,properties:{ticket_id:{type:string}}}),Tool(namecreate_ticket,description创建工单,inputSchema{type:object,properties:{title:{type:string}}}),]server.call_tool()asyncdefhandle_call_tool(name:str,arguments:dict):MCP 协议调用指定工具ifnamequery_ticket:resultawaitquery_ticket_internal(arguments[ticket_id])return[TextContent(typetext,textjson.dumps(result))]elifnamecreate_ticket:resultawaitcreate_ticket_internal(arguments[title])return[TextContent(typetext,textjson.dumps(result))]技术映射自定义 Tool 自己造一个带特定接口的插座MCP Server 用国标插座标准任何符合国标的插头都能插进来。小胖“那这些工具在 Canvas 里怎么用”大师“在 Canvas 中添加一个Tool组件选择你想要调用的工具。Agent 会在执行时自动决策——根据用户的问题判断要不要调工具、调哪个工具、用什么参数——无需手动在 Canvas 上连线。这就是自主 Agent模式Agent 自己规划→执行→观察→再规划直到完成任务。”小白“安全方面呢工具调用的权限、审计、限制怎么搞”大师“工具体系的安全四要素”工具调用安全四要素 1. 参数校验Parameter Validation 每个工具定义的 parameters schema 就是第一道防线 RAGFlow 在调用前校验参数类型、必填项、枚举值 2. 权限隔离Permission Isolation 不同用户/租户可用的工具集合不同 工单系统普通员工只能 query管理员可以 create/update 3. 超时控制Timeout Control 每个工具调用有独立的超时时间 查询类工具5-10 秒创建类工具15-30 秒 4. 审计日志Audit Log 每次工具调用记录谁、什么时候、调了什么工具、传了什么参数、返回了什么 用于安全审计和调试回溯3 项目实战环境准备目标实现一个 Agent先检索制度再调 SQL 查员工报销额度最后生成审批建议。前提RAGFlow 已部署SQL 数据库可访问含员工信息表Canvas 可用。分步实现步骤1配置工具体系目标在 RAGFlow 系统设置中启用并配置 SQL Tool 和自定义邮件 Tool。# 1. 配置 SQL Tool内置curl-XPOST http://localhost/api/v1/tools/configure\-HAuthorization: Bearer$TOKEN\-HContent-Type: application/json\-d{ tool_name: sql_query, config: { db_type: mysql, host: mysql, port: 3306, database: yunfan_hr, username: readonly_user, password: ReadOnly2024, max_rows: 100, query_timeout: 10 } }# 2. 配置 Email Tool内置curl-XPOST http://localhost/api/v1/tools/configure\-HAuthorization: Bearer$TOKEN\-HContent-Type: application/json\-d{ tool_name: send_email, config: { smtp_host: smtp.yunfan.com, smtp_port: 587, username: ragflow-botyunfan.com, password: BotEmail2024, from_name: 云帆报销助手 } }步骤2连接 MCP Server目标启动一个 MCP Server 并注册到 RAGFlow。# 启动工单系统 MCP Server示例# 方式通过 npx 或 python 启动npx-ymodelcontextprotocol/server-ticket-system\--api-url https://ticket.yunfan.com/api\--api-key$TICKET_API_KEY\--port3001# 在 RAGFlow 中注册该 MCP Servercurl-XPOST http://localhost/api/v1/tools/mcp/register\-HAuthorization: Bearer$TOKEN\-HContent-Type: application/json\-d{ name: 工单系统, transport: streamable_http, url: http://ticket-mcp:3001, description: 公司内部工单系统支持查询、创建、更新工单 }坑点MCP 支持两种传输方式——stdio进程间通信和streamable_http网络通信。Docker 容器之间用streamable_http。步骤3创建工具增强型 Agent Canvas目标搭建一个完整的报销咨询→额度查询→审批建议→邮件发送工作流。# step3_agent_with_tools.py# 创建 Agent 型 Canvas让 LLM 自主决策调哪些工具canvasrag.create_canvas(name智能报销审批助手)# Begin 入口begincanvas.add_component(Begin,idbegin)# Agent 组件自主规划工具调用agentcanvas.add_component(Agent,idagent_main,config{model:gpt-4o,temperature:0.1,max_iterations:10,# 最多调用 10 次工具available_tools:[retrieval,# RAG 检索sql_query,# SQL 查询send_email,# 发送邮件ticket_system,# 工单系统来自 MCP],system_prompt:你是一个智能报销审批助手。处理逻辑 1. 首先在知识库中检索相关报销制度 2. 如果用户提到具体员工用SQL查询该员工的报销额度信息 3. 综合制度标准和员工额度给出报销建议 4. 如果需要人工审批创建工单并发送邮件通知 重要涉及数据库查询时务必只查询、不修改。 涉及发送邮件时务必先确认收件人邮箱。,})# End 出口endcanvas.add_component(End,idend)canvas.connect(begin,agent)canvas.connect(agent,end)canvas.publish()步骤4工具调用链追踪目标观察 Agent 在一次复杂任务中调用了哪些工具及其顺序。# step4_trace_tool_calls.pyimportjson# 测试问题test_input查询张三的报销额度帮他算一下出差北京5天能报销多少然后发邮件通知他。sessioncanvas.test_run(inputs{query:test_input})print( Agent 工具调用链 )fori,stepinenumerate(session.execution_log):ifstep.component_idagent_main:print(f\n--- Agent 迭代{step.iteration}---)ifstep.action_typetool_call:print(f 工具调用:{step.tool_name})print(f 参数:{json.dumps(step.tool_args,ensure_asciiFalse)})print(f 结果:{str(step.tool_result)[:200]}...)elifstep.action_typellm_thought:print(f LLM 思考:{step.thought[:200]}...)elifstep.action_typefinal_answer:print(f 最终回答:{step.output[:300]}...)预期输出 Agent 工具调用链 --- Agent 迭代 1 --- LLM 思考: 用户想查询张三的报销额度并计算出差费用。我需要先检索报销制度再查数据库。 工具调用: retrieval 参数: {query: 出差住宿费标准 北京} 结果: 根据差旅费管理办法一线城市北京住宿标准为500元/晚... --- Agent 迭代 2 --- 工具调用: sql_query 参数: {query: SELECT name, reimbursement_limit, used_amount FROM employees WHERE name LIKE %张三%} 结果: {rows: [{name: 张三, reimbursement_limit: 10000, used_amount: 3000}]} --- Agent 迭代 3 --- LLM 思考: 张三剩余额度7000元。北京出差5天住宿费5×5002500元。25007000额度足够。 工具调用: send_email 参数: {to: zhangsanyunfan.com, subject: 出差报销计算结果, body: 张三你好你的出差北京5天预计报销2500元...} --- Agent 迭代 4 --- LLM 思考: 任务完成。已检索制度、查询额度、发送通知。准备最终回答。 最终回答: 张三您本次出差北京的预估报销如下住宿费 2500元5晚×500元/晚交通补助等详见邮件。您的年度报销额度剩余7000元本次报销后额度充足。详细计算结果已发送至您的邮箱 zhangsanyunfan.com。步骤5工具调用的安全与监控目标配置工具的审计日志和调用限制。# step5_tool_audit.py - 审计与监控classToolAuditLogger:工具调用审计日志deflog_call(self,user_id,tenant_id,tool_name,args,result,duration_ms,success):记录每次工具调用importjson,datetime record{timestamp:datetime.datetime.now().isoformat(),user_id:user_id,tenant_id:tenant_id,tool_name:tool_name,args:args,result_preview:str(result)[:200]ifresultelseNone,duration_ms:duration_ms,success:success,}# 写入审计日志表# AuditLog.create(**record)print(f[AUDIT]{tool_name}by{user_id}:{OKifsuccesselseFAILED}({duration_ms}ms))# 工具调用频率限制classToolRateLimiter:防止工具被滥用def__init__(self):self.limits{sql_query:{per_minute:30,per_user_per_minute:5},send_email:{per_minute:10,per_user_per_minute:2},ticket_system:{per_minute:20,per_user_per_minute:3},}self.countersdefaultdict(lambda:defaultdict(int))defcheck_and_increment(self,tool_name,user_id):检查是否超过限流阈值未超过则记录一次调用limitsself.limits.get(tool_name,{per_minute:10})current_minuteint(time.time()/60)# 全局限制global_keyf{tool_name}:{current_minute}user_keyf{tool_name}:{user_id}:{current_minute}ifself.counters[global_key][count]limits.get(per_minute,10):raiseException(f工具{tool_name}全局限流已达上限)ifself.counters[user_key][count]limits.get(per_user_per_minute,3):raiseException(f用户{user_id}对工具{tool_name}的调用频率已达上限)self.counters[global_key][count]1self.counters[user_key][count]1returnTrue测试验证# test_tool_integration.pyclassTestToolIntegration:deftest_sql_tool_returns_data(self):验证 SQL 工具可正常查询resulttool_registry.call(sql_query,{query:SELECT COUNT(*) AS cnt FROM employees})assertresult[rows][0][cnt]0deftest_agent_chooses_correct_tool(self):验证 Agent 为不同问题选择正确工具# 制度类问题 → 应选 retrievalresponseagent_chat(年假有几天)tool_callsextract_tool_calls(response)assertany(retrievalintcfortcintool_calls)# 数据查询类 → 应选 sql_queryresponseagent_chat(张三的报销额度是多少)tool_callsextract_tool_calls(response)assertany(sql_queryintcfortcintool_calls)deftest_tool_rate_limiting(self):验证限流生效limiterToolRateLimiter()# 连续调用 6 次超过用户每分钟限制 5 次for_inrange(5):limiter.check_and_increment(sql_query,test_user)withpytest.raises(Exception,match频率已达上限):limiter.check_and_increment(sql_query,test_user)完整代码清单路径说明agent/tools/base.pyToolBase 基类参数校验、超时、审计agent/tools/sql.pySQL 查询工具agent/tools/email.py邮件发送工具agent/tools/retrieval.py检索工具agent/plugin/MCP 客户端与集成agent/canvas.pyCanvas 中 Tool 组件的执行逻辑4 项目总结优点 缺点维度RAGFlow 工具MCPLangChain ToolsOpenAI Function Calling自建工具框架工具标准化★★★ MCP 协议★★☆ Tool 基类★★★ Function schema★☆☆ 需自行定义运行时发现★★★ MCP 自动发现★★☆ 手动注册★☆☆ 需代码注册★☆☆ 需开发RAG 深度集成★★★ 原生检索工具★★★ 可组合★☆☆ 需自行实现★★☆ 需集成Canvas 可视化★★★ 拖拽式编排★☆☆ 代码编排★☆☆ 无★☆☆ 需开发安全管控★★☆ 基本审计限流★★☆ 回调控制★☆☆ 依赖平台★★★ 完全可控学习曲线★★☆ 中等★★☆ 中等★★★ 较简单★☆☆ 高适用场景数据驱动的业务 Agent检索制度 SQL 查数据 外部 API 调用——如本章的报销审批。企业内部系统集成通过 MCP 标准化接入工单系统、OA、CRM、ERP 等。多工具协同Agent 自主选择最优工具组合完成复杂任务。低代码扩展非开发人员通过 MCP Server 模板快速接入新系统。工具集市按部门/租户发布可用工具建成企业工具库。不适用场景极简单的 RAG 问答不需要工具调用纯检索即可。无外部系统依赖的封闭环境如果 Agent 只访问内部知识库不需要工具体系。注意事项SQL Tool 的只读限制强烈建议 SQL 工具连接只读副本或限制只能执行 SELECT 语句。绝不能给 Agent DELETE/DROP 权限。MCP Server 的传输安全streamable_http方式如果部署在公网必须开启 TLS 鉴权。stdio方式更安全但仅限于同一台机器。工具描述的精确性工具的description和参数description直接影响 LLM 的决策——写得模糊 LLM 就会选错工具。审计日志的存储与保留工具调用涉及敏感的数据库查询和邮件发送审计日志建议保留至少 90 天。Agent 无限循环如果 LLM 反复调同一个工具且每次都说再试一次可能陷入死循环——max_iterations是兜底。常见踩坑经验故障现象根因解决方法LLM 调了错误的工具工具名/描述与用户意图的语义不匹配优化工具的 description用更多同义词SQL 工具返回数据太大导致超时SELECT * 返回数万行设置max_rows100Agent Prompt 中要求加 LIMITMCP Server 断连后 Agent 一直重试无连接健康检查Agent 不知道工具不可用MCP Client 侧加 health check不可用工具从列表移除工具参数校验老是失败LLM 输出的参数格式不完全符合 schema宽容解析——对 LLM 输出的参数做兼容性处理如 “北京” → [“北京”]审计日志被恶意用户刷爆无日志采样和限流对单个用户的审计日志做限流采样每用户每分钟最多 10 条思考题公司的数据库中有 200 张表。如果让 Agent 直接看到全部 200 张表LLM 会难以选择正确的表和字段。请设计一个分层工具注册方案——第一层只有 10 个业务视图工具查询员工信息、查询报销记录、查询部门预算…第二层的具体 SQL 由视图工具内部生成Agent 永远看不到底层表结构。如果一条用户消息触发 Agent 调用了 5 个工具检索→SQL→计算→邮件→工单其中第 4 步发邮件时失败了SMTP 服务器宕机。请设计一个Saga 补偿方案——将已执行的操作回滚或补偿如撤销工单、标记计算结果为无效、通知用户部分操作失败。答案提示见第26章末尾或附录 D。延伸阅读与资源10倍开发者的 Dify 魔法书从零构建全栈 AI 应用后端工程师转型AI第一课-Ollama 与私有化大模型实战大型语言模型(LLM) vLLM 高性能推理落地实战Agent开发之LlamaIndex 实战修炼与源码进阶大语言模型Transformers 实战修炼与源码剖析