简介一套聚焦金融科技场景的DifyMCP智能理财助手实战资料面向具备Python基础、希望掌握AI智能体开发的开发者。内容围绕基于Model as Copilot协议构建可执行金融操作的智能体完整覆盖金融助手核心能力架构、MCP智能体协议配置、Dify工作流构建、微信消息路由服务及RAG知识库集成并给出关键代码片段。资料为单个PDF文档仅248KB却在有限篇幅内梳理了实时行情数据获取、资产组合分析引擎、微信模板消息生成等核心实现以及系统部署与高并发压测结果便于快速通读与按步骤实践。读者可从中理解从用户意图识别、行情查询到策略生成、微信端自动推送的全链路自动化设计逻辑同时获得可复用的代码思路与部署排错参考适合金融科技、量化投资等交叉领域的开发者深入学习。该资料已有262人次学习具有不错的实战参考价值。1. 金融科技场景下先把这套推送系统的真问题说透把大模型接进理财助手最难的往往不是让模型读懂行情而是让策略跑完一圈之后还能稳定、合规、准时地推到微信上。很多团队做完行情分析和组合建议却卡在“结果出得来、消息发不出去、发了没人看”这三道坎上。这套基于 Dify MCP 的微信端自动化组合策略推送系统核心要解决的正是“策略计算 工具调用 消息触达”的完整闭环Dify 负责把策略逻辑编排成可调试的工作流MCP 负责把行情、信号、推送这些外部能力统一成标准接口最后通过微信端把组合策略推送给用户。适合手里已经有一定策略研究基础、想快速落地一个自动化推送原型的开发者或小团队。2. 选型与架构为什么用 Dify 编排 MCP 扩展而不是直接写代码2.1 Dify 在金融策略场景里到底帮了什么忙我最早接手类似需求时第一反应是把策略逻辑直接写成 Python 服务然后用定时任务触发、调微信 API 推送。但很快就发现一个问题策略条件经常要改行情源要换推送文案要调每次改动都要走“改代码 → 重新部署 → 测试”的流程一天能迭代两次就不错了。而且团队成员不全是后端工程师产品同学想看一眼“当前策略到底怎么跑的”也看不到。换成 Dify 之后最大的变化是策略逻辑的“可视化编排”。你可以在工作流里拖出“数据获取 → 信号计算 → 组合权重 → 文案生成 → 推送触达”这几个节点每个节点的输入输出一目了然。用户问“为什么今天推了这只基金”可以直接在工作流运行记录里看到每一步的中间结果不用猜。Dify 的另一个价值是 Prompt 和参数的版本管理。策略文案、风险提示措辞、推送频率限制这些内容都作为节点配置存在改动时只影响当前节点不会牵扯到其他代码。对金融场景来说这种“可审计”的变更记录很重要——监管抽查时你得说清楚某天的推送文案是哪一版策略生成出来的。2.2 MCP 不是“API 聚合器”是工具治理层MCP 容易被人理解成一堆 API 的封装但实际它在架构里的定位更高一层。策略系统要对接行情源、数据库、微信 API、回测服务如果每个都自己写 HTTP 调用Dify 工作流里就会塞满十几个自定义节点维护成本直线上升。MCP 的思路是把每个外部能力定义成一个标准化的“工具”工具的参数、返回结构、错误码都遵循同一套协议Dify 只需要通过 MCP 客户端去发现和调用这些工具。举个例子行情数据既可以来自某免费接口也可以换成付费数据源只要 MCP 工具内部把来源差异消化掉对外暴露的仍然是get_market_data(symbol, period)这个统一接口Dify 工作流完全不需要改。这就是“工具治理”的意义——把变化隔离在 MCP 服务内部让上层编排保持稳定。MCP 对金融场景还有一个很实际的好处工具调用记录可以做审计。谁在什么时间调用了行情工具、传了什么参数、返回了什么结果MCP 服务端都能留痕。合规审计要数据时直接导出一份调用日志比翻业务代码简洁得多。2.3 整体架构数据流、策略流、消息流三层分离我采用的架构分三层每层职责单一也方便单独部署和扩展。第一层是数据与工具层也就是 MCP 服务端。这里跑着行情获取、指标计算、信号判定、微信消息发送这几个工具。每个工具独立实现内部处理超时、重试和异常对外只暴露 MCP 协议接口。第二层是编排层也就是 Dify 工作流。它负责调度先后顺序先拿行情再算信号然后生成组合权重最后调用推送工具。这一层不做复杂计算只做流程控制。第三层是触达层。微信 API 调用本身放在 MCP 工具里但消息模板的管理、发送频率控制、失败补偿逻辑放在工作流节点中方便运营调整。这样的分层让两端都松耦合换微信端渠道时只改 MCP 工具实现Dify 工作流不动改策略逻辑时只动工作流节点MCP 工具不动。3. 从零搭建策略工作流在 Dify 里把行情、信号、组合权重和推送串起来3.1 第一步创建空白工作流并定义输入输出在 Dify 中新建一个“工作流”类型的应用命名为“组合策略每日推送”。先不急着拖节点先把输入和输出定义清楚。输入字段我一般保留三个user_id接收推送的用户标识portfolio_type稳健型 / 平衡型 / 进取型trigger_source预留字段用于区分是定时触发还是用户主动触发输出字段则根据推送需求定义为一个 JSON 结构体包含message_title、message_content、risk_tips和send_status。定义好输入输出后后续每个节点都能引用这些字段不会出现“跑到一半发现字段名对不上”的问题。这里的技巧是输入输出一定要用“窄接口”。不要试图把整个用户画像都传给工作流而是只传推送所需的必要信息其他数据通过 MCP 工具在节点内部拉取。这样工作流的可复用性更高换一个客户群体时不需要重构。3.2 第二步用 MCP 工具节点把行情和信号接入工作流Dify 的工作流节点支持调用 MCP 工具。配置方法是在节点类型里选择“MCP 工具”填入 MCP 服务器的地址和工具名称然后在上游节点或输入字段中配置参数映射。这里用一个示例描述 MCP 服务器如何定义get_market_data和generate_strategy_signal这两个工具。下面是一个基于 Python MCP 开发包的简化实现from mcp.server import Server from mcp.server.stdio import run_server app Server(finance_tools) app.tool() def get_market_data(symbol: str, period: str daily) - dict: 获取指定投资标的的行情数据返回最近 N 个周期的 OHLC 数据。 # 实际项目中这里会请求行情源并做超时和重试处理 data { symbol: symbol, period: period, close: [1.02, 1.05, 1.04, 1.08], ma5: 1.02, ma20: 0.98, timestamp: 2025-06-10T15:00:0008:00 } return data app.tool() def generate_strategy_signal(market_data: dict, portfolio_type: str) - dict: 基于行情数据计算买入/卖出/持有信号返回信号强度和仓位置信度。 close market_data[close] latest close[-1] ma5 market_data[ma5] signal hold strength 0.5 if latest ma5 * 1.02 and portfolio_type 进取型: signal buy strength 0.8 return { signal: signal, strength: strength, reason: f最新收盘价较5日均线偏离{ (latest/ma5 - 1) * 100:.2f}% } if __name__ __main__: run_server(app)这段代码的关键在于工具函数用装饰器注册输入输出都是可序列化的dictDify 工作流调用后会直接拿到结构化的 JSON不需要再做额外解析。注意portfolio_type这个参数必须由工作流从输入字段透传进来否则工具内部无法判断策略风格。实际配置 MCP 工具节点时需要在 Dify 中填写 MCP 服务器地址。Dify 在运行工作流时会在后端发起连接所以要确保 MCP 服务的网络可达。如果是本地开发用127.0.0.1加端口即可如果是部署在云上建议给 MCP 服务配上内网地址避免公网暴露。调用顺序上先调用get_market_data拿到行情再把返回结果作为generate_strategy_signal的market_data参数。在 Dify 节点配置里下拉框选择“引用上游节点输出”就能自动映射不需要手写代码。3.3 第三步用 LLM 节点生成组合策略和推送文案信号计算出来之后接下来要做的是把“信号”翻译成用户看得懂的“策略建议”。这一步我用 Dify 的 LLM 节点让它根据信号结果生成仓位建议和推送文案。LLM 节点的 Prompt 设计有讲究。我不会让它自由发挥而是给出非常明确的结构和约束。参考 Prompt 如下你是一位合规的金融理财助手。请根据以下策略信号生成组合策略建议 信号结果{{signal_result}} 用户风险偏好{{portfolio_type}} 输出格式要求 1. 先用一句中文总结当前市场状态 2. 再给出建议仓位比例股/债/现金合计必须等于100% 3. 最后附上一句风险提示提示用户“策略基于历史数据不构成投资建议” 不要输出任何收益承诺不要提到具体个股推荐。Prompt 中出现的{{signal_result}}是 Dify 的变量引用语法会在运行时替换为上游节点的输出。这里有两个必须控制的点第一是仓位比例合计 100%我在 Prompt 里写死后后续可以在工作流代码节点里再做数值校验第二是合规措辞“不构成投资建议”这句必须让模型无脑带出来所以把它直接写进 Prompt 要求的第三条。LLM 节点生成的结构化建议会传递到后续推送工具节点。我还会在节点输出结果中要求返回suggestion和risk_tips两个字段这样推送到微信时可以分开渲染。3.4 第四步定时触发与服务 API 配置策略推送需要每天固定时间运行不可能手动点“运行”按钮。Dify 工作流支持配置“定时触发”在应用设置里选择 Cron 表达式例如0 17 * * 1-5表示工作日下午五点整触发。定时触发时工作流的输入字段user_id需要一个默认值。常见做法是先配置一个“管理员用户”作为默认接收人等验证通过后再通过服务 API 动态传入不同用户的标识。Dify 的服务 API 可以让你从外部系统调用工作流。用 Python 请求示例import requests url https://your-dify-instance/v1/workflows/run headers { Authorization: Bearer app-xxxxxxxxxxxxx, Content-Type: application/json } payload { inputs: { user_id: user_001, portfolio_type: 平衡型 }, response_mode: blocking } resp requests.post(url, jsonpayload, headersheaders) print(resp.json())代码中response_mode我一般用blocking这样可以直接拿到工作流最终输出结果方便判断推送是否成功。如果策略链路很长、耗时超过 30 秒建议改用streaming模式按节点逐段接收运行状态避免等待超时。Bearer后面的 token 在 Dify 的“API 访问”页面生成注意这不是模型供应商的 Key而是这个工作流应用自己的访问凭证。4. 微信端推送实现把 Dify 工作流接到微信 API 的完整链路4.1 微信端推送的三条路线怎么选微信生态里能“自动化推送”的通道不止一个我实际对比过的有三条路线企业微信应用消息、微信公众号模板消息、微信客服消息。三者的差异主要在接口权限、消息格式和触发条件上。企业微信应用消息是最适合自建系统做自动化推送的。它的接口可以直接给指定用户发文本卡片不需要用户先关注、不需要模板审核只要拿到用户的userid就能推送。公众号模板消息推送更偏向运营场景需要申请模板、模板内容要经过类目审核适合给关注了公众号的粉丝做定期服务。微信客服消息则适合“用户先找过来、你再回复”的被动场景不适合主动推送。我的建议是业务验证期优先走企业微信应用消息因为权限链路短、调试快等验证了推送带来的真实价值后再考虑公众号模板消息做更大范围触达。这篇采用企业微信应用消息作为主通道。4.2 企业微信应用消息的最小实现创建一个企业微信自建应用拿到corpid、agentid和secret后推送分两步获取access_token然后调用message/send接口。最小实现如下import requests CORP_ID your_corp_id AGENT_ID your_agent_id SECRET your_app_secret def get_access_token(): url https://qyapi.weixin.qq.com/cgi-bin/gettoken params {corpid: CORP_ID, corpsecret: SECRET} resp requests.get(url, paramsparams, timeout5).json() if resp.get(errcode) ! 0: raise RuntimeError(f获取access_token失败: {resp}) return resp[access_token] def send_wecom_text(token, user_id, content): url https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token token payload { touser: user_id, msgtype: text, agentid: int(AGENT_ID), text: {content: content} } resp requests.post(url, jsonpayload, timeout5).json() if resp.get(errcode) ! 0: raise RuntimeError(f消息发送失败: {resp}) return resp这里最容易出问题的是agentid类型。企业微信接口要求它是整型但很多配置项里存的是字符串直接传字符串会报invalid agentid。我在代码里用int()显式转换避免类型不匹配的坑。另外touser支持多个用户用竖线分隔例如user1|user2。但一条消息最多 100 人超过就需要分批发。策略推送通常是一对一内容所以这里保持单用户发送。4.3 把推送工具封装成 MCP 工具并接入 Dify企业微信的发送 API 不适合直接在 Dify 的 HTTP 节点里裸调因为 token 缓存、错误重试、频率控制这些逻辑放工作流里会很笨重。正确做法是把它封装成一个 MCP 工具push_wecom_message然后在 Dify 工作流最后一步调用。app.tool() def push_wecom_message(user_id: str, title: str, content: str, risk_tips: str) - dict: 通过企业微信应用消息推送理财策略报告。 token get_access_token() # 内部会做缓存避免每次请求都获取token message f【{title}】\n{content}\n\n{risk_tips} result send_wecom_text(token, user_id, message) return { send_status: result.get(errcode) 0, errcode: result.get(errcode), errmsg: result.get(errmsg) }封装的好处是Dify 工作流里只需要一个节点就能完成推送状态返回统一是结构化 JSON。send_status会被后续的“条件分支”节点读取如果发送失败可以走重试分支或记录日志。Dify 配置该节点时输入参数title、content、risk_tips都引用上游 LLM 节点的输出字段。推送时间一般耗时在几秒到十几秒工作流节点超时时间建议设置成 30 秒。4.4 消息模板字段与频率限制规避企业微信对应用消息的频率限制是“每应用每分钟最多 200 条”。策略推送一般每天只有一次所以限频本身不会触发但有一种情况会同一个用户信息在测试环境重复推送多次而且每次推送都触发了微信 API 的 token 刷新容易把调用量放大。我的习惯是生产环境的access_token必须做缓存有效期 7200 秒。线程安全的写法是token_cache {token: None, expire_at: 0} def get_access_token(): import time now time.time() if token_cache[token] and now token_cache[expire_at] - 60: return token_cache[token] # 重新获取并写入缓存 token_cache[token] _fetch_token() token_cache[expire_at] now 7200 return token_cache[token]做了一层 60 秒的提前过期保护避免 token 刚好在边界时失效。经验是尽量别在推送高峰期的最后几秒续期否则错过一次推送用户就会在微信上问“今天的策略去哪了”。5. 避坑指南智能理财助手上线前最容易踩的六个坑5.1 策略信号分钟级变化推送却是日级频率结果对不上现象行情信号工作流跑了多次用户收到的推送内容里策略信号是中午的推送时间却是下午五点用户疑问“这建议还成立吗”。原因定时触发只负责启动工作流但行情数据获取节点拿到的是运行时刻的最新价而不是策略决策时点的收盘价。不同节点取数时间不一导致信号和消息对不上。解决在get_market_data工具中显式传入as_of_date参数强制拉取特定交易日的日线收盘数据而工作流的定时时间设定在收盘后确保数据源已经更新。5.2 MCP 工具超时导致整个工作流卡死现象某次行情源响应很慢MCP 工具自己没设超时Dify 工作流一直停在“调用工具”状态直到整个流程超时失败。原因MCP 服务端函数没有设置网络请求超时同时 Dify 节点的超时时间设置过长掩盖了上游慢接口的问题。解决MCP 工具内部所有网络调用都显式加timeout同时在 Dify 节点配置中把超时时间压到 15 秒。一旦超时就走“失败-降级”分支推送一条“策略暂时不可用”的提示而不是让用户干等。5.3 文案里的“建议”踩到金融合规边界现象LLM 生成文案里出现“该组合预计年化收益 12%”“这只基金值得重仓”这类表述被安全审查打回。原因Prompt 约束不到位模型把“策略建议”理解成了“收益保证”。金融场景里任何收益预测和个股推荐都需要极强的前提限定营销话术不适合直接进推送。解决在 Prompt 中追加硬性负向限制例如“禁止出现收益预测”“禁止出现具体产品推荐”“必须包含风险提示”同时在 LLM 节点输出后加一个代码节点做关键词过滤命中“收益”“增持”“卖出”等高风险词时拦截消息并替换为模板话术。5.4 时区与交易日历错乱导致推送日期不准现象部署在云服务器上默认时区是 UTC在 0 点多运行时算出来的“今日策略”推送日期却是前一天的。原因Python 服务里datetime.now()返回的是服务器时区时间而不是北京时间同时没有判断当天是否交易日节假日也照常推送。解决所有时间相关逻辑统一用zoneinfo.ZoneInfo(Asia/Shanghai)获取当前时间再接入交易日历数据源非交易日直接跳过推送工作流返回{skipped: true}不产生微信消息。5.5 微信 API 返回错误码但工作流仍显示成功现象消息发送失败但 Dify 工作流运行记录里显示成功日志也没报错。原因MCP 工具内部没有检查企业微信返回的errcode直接把 HTTP 200 当成了成功返回。企业微信业务失败时 HTTP 状态码仍然是 200只有errcode是非零。解决push_wecom_message内部严格判断errcode 0否则抛异常。Dify 节点捕获异常后进入“失败分支”把错误信息写入审计日志并触发补偿流程。5.6 隔了好久改 Prompt历史策略推送记录对不上现象运营想对比“改版前后两种策略文案的效果”但旧文案已经无法复现因为 Dify 里只保留了最新配置。原因没有在工作流运行前给当前 Prompt 和参数打版本快照。Dify 的运行记录能看中间结果但 LLM 节点的 Prompt 是动态读取的改动后历史运行的输入配置变了。解决在 MCP 工具中加一个save_strategy_version工具每次工作流启动时把当前 Prompt、模型参数、策略类型存入数据库。审计时用version_id关联历史推送记录保证可追溯。6. 进阶加一层灰度验证让推送系统先证明自己再全量发系统能跑通“行情 → 信号 → 文案 → 推送”之后最值得做的一件事就是灰度验证。我见过太多项目链路通了就全量上线结果第一次推送就出问题影响一堆用户。做法分三步。第一步是“个人内测”工作流输入user_id固定填自己的企业微信 ID推送内容和线上完全一致重点是验证消息到达率、显示格式和文案观感。第二步是“影子推送”策略照常运行但同时把推送结果写一份到本地日志不实际发给用户等积累一周后拿这些影子数据跟真实市场走势做对比看策略建议的胜率和回撤是否符合预期。第三步才是“小流量灰度”按用户 ID 哈希取模先放 5% 的用户观察两三天。灰度比例控制我一般放在 Dify 工作流入口的“条件分支”节点里实现。输入新增一个gray_ratio参数判断逻辑是hash(user_id) % 100 gray_ratio才走完整推送链路否则只记录日志不推送。这样灰度比例从 5% 调到 10%、50% 只需要改一个参数不用改代码。灰度期间重点盯两个指标一个是推送送达率也就是send_status为 true 的比例另一个是用户点击或取消关注的反馈行为。如果送达率低于 95%先查 MCP 工具的错误码分布和 token 缓存是否失效。别急着查策略逻辑——大部分推送问题出在触达链路而不是策略模型。我自己的教训是上线后最容易翻车的是“策略挺好消息没发出去”。后来养成一个习惯每天推送完成后在工作流末尾加一个“健康检查”节点把当天的push_success_count、failed_user_list汇总推到自己的企业微信上几秒钟就能看出来系统是不是正常。你如果也在做类似的智能理财助手推送系统建议把这个习惯保留下来能省下不少半夜排查的时间。希望帮到你。本文还有配套的精品资源点击获取 
长河君 · 2026/10/17 15:12:23 · 阅读 1.2k Dify + MCP 构建微信端金融策略自动推送系统实战
简介一套聚焦金融科技场景的DifyMCP智能理财助手实战资料面向具备Python基础、希望掌握AI智能体开发的开发者。内容围绕基于Model as Copilot协议构建可执行金融操作的智能体完整覆盖金融助手核心能力架构、MCP智能体协议配置、Dify工作流构建、微信消息路由服务及RAG知识库集成并给出关键代码片段。资料为单个PDF文档仅248KB却在有限篇幅内梳理了实时行情数据获取、资产组合分析引擎、微信模板消息生成等核心实现以及系统部署与高并发压测结果便于快速通读与按步骤实践。读者可从中理解从用户意图识别、行情查询到策略生成、微信端自动推送的全链路自动化设计逻辑同时获得可复用的代码思路与部署排错参考适合金融科技、量化投资等交叉领域的开发者深入学习。该资料已有262人次学习具有不错的实战参考价值。1. 金融科技场景下先把这套推送系统的真问题说透把大模型接进理财助手最难的往往不是让模型读懂行情而是让策略跑完一圈之后还能稳定、合规、准时地推到微信上。很多团队做完行情分析和组合建议却卡在“结果出得来、消息发不出去、发了没人看”这三道坎上。这套基于 Dify MCP 的微信端自动化组合策略推送系统核心要解决的正是“策略计算 工具调用 消息触达”的完整闭环Dify 负责把策略逻辑编排成可调试的工作流MCP 负责把行情、信号、推送这些外部能力统一成标准接口最后通过微信端把组合策略推送给用户。适合手里已经有一定策略研究基础、想快速落地一个自动化推送原型的开发者或小团队。2. 选型与架构为什么用 Dify 编排 MCP 扩展而不是直接写代码2.1 Dify 在金融策略场景里到底帮了什么忙我最早接手类似需求时第一反应是把策略逻辑直接写成 Python 服务然后用定时任务触发、调微信 API 推送。但很快就发现一个问题策略条件经常要改行情源要换推送文案要调每次改动都要走“改代码 → 重新部署 → 测试”的流程一天能迭代两次就不错了。而且团队成员不全是后端工程师产品同学想看一眼“当前策略到底怎么跑的”也看不到。换成 Dify 之后最大的变化是策略逻辑的“可视化编排”。你可以在工作流里拖出“数据获取 → 信号计算 → 组合权重 → 文案生成 → 推送触达”这几个节点每个节点的输入输出一目了然。用户问“为什么今天推了这只基金”可以直接在工作流运行记录里看到每一步的中间结果不用猜。Dify 的另一个价值是 Prompt 和参数的版本管理。策略文案、风险提示措辞、推送频率限制这些内容都作为节点配置存在改动时只影响当前节点不会牵扯到其他代码。对金融场景来说这种“可审计”的变更记录很重要——监管抽查时你得说清楚某天的推送文案是哪一版策略生成出来的。2.2 MCP 不是“API 聚合器”是工具治理层MCP 容易被人理解成一堆 API 的封装但实际它在架构里的定位更高一层。策略系统要对接行情源、数据库、微信 API、回测服务如果每个都自己写 HTTP 调用Dify 工作流里就会塞满十几个自定义节点维护成本直线上升。MCP 的思路是把每个外部能力定义成一个标准化的“工具”工具的参数、返回结构、错误码都遵循同一套协议Dify 只需要通过 MCP 客户端去发现和调用这些工具。举个例子行情数据既可以来自某免费接口也可以换成付费数据源只要 MCP 工具内部把来源差异消化掉对外暴露的仍然是get_market_data(symbol, period)这个统一接口Dify 工作流完全不需要改。这就是“工具治理”的意义——把变化隔离在 MCP 服务内部让上层编排保持稳定。MCP 对金融场景还有一个很实际的好处工具调用记录可以做审计。谁在什么时间调用了行情工具、传了什么参数、返回了什么结果MCP 服务端都能留痕。合规审计要数据时直接导出一份调用日志比翻业务代码简洁得多。2.3 整体架构数据流、策略流、消息流三层分离我采用的架构分三层每层职责单一也方便单独部署和扩展。第一层是数据与工具层也就是 MCP 服务端。这里跑着行情获取、指标计算、信号判定、微信消息发送这几个工具。每个工具独立实现内部处理超时、重试和异常对外只暴露 MCP 协议接口。第二层是编排层也就是 Dify 工作流。它负责调度先后顺序先拿行情再算信号然后生成组合权重最后调用推送工具。这一层不做复杂计算只做流程控制。第三层是触达层。微信 API 调用本身放在 MCP 工具里但消息模板的管理、发送频率控制、失败补偿逻辑放在工作流节点中方便运营调整。这样的分层让两端都松耦合换微信端渠道时只改 MCP 工具实现Dify 工作流不动改策略逻辑时只动工作流节点MCP 工具不动。3. 从零搭建策略工作流在 Dify 里把行情、信号、组合权重和推送串起来3.1 第一步创建空白工作流并定义输入输出在 Dify 中新建一个“工作流”类型的应用命名为“组合策略每日推送”。先不急着拖节点先把输入和输出定义清楚。输入字段我一般保留三个user_id接收推送的用户标识portfolio_type稳健型 / 平衡型 / 进取型trigger_source预留字段用于区分是定时触发还是用户主动触发输出字段则根据推送需求定义为一个 JSON 结构体包含message_title、message_content、risk_tips和send_status。定义好输入输出后后续每个节点都能引用这些字段不会出现“跑到一半发现字段名对不上”的问题。这里的技巧是输入输出一定要用“窄接口”。不要试图把整个用户画像都传给工作流而是只传推送所需的必要信息其他数据通过 MCP 工具在节点内部拉取。这样工作流的可复用性更高换一个客户群体时不需要重构。3.2 第二步用 MCP 工具节点把行情和信号接入工作流Dify 的工作流节点支持调用 MCP 工具。配置方法是在节点类型里选择“MCP 工具”填入 MCP 服务器的地址和工具名称然后在上游节点或输入字段中配置参数映射。这里用一个示例描述 MCP 服务器如何定义get_market_data和generate_strategy_signal这两个工具。下面是一个基于 Python MCP 开发包的简化实现from mcp.server import Server from mcp.server.stdio import run_server app Server(finance_tools) app.tool() def get_market_data(symbol: str, period: str daily) - dict: 获取指定投资标的的行情数据返回最近 N 个周期的 OHLC 数据。 # 实际项目中这里会请求行情源并做超时和重试处理 data { symbol: symbol, period: period, close: [1.02, 1.05, 1.04, 1.08], ma5: 1.02, ma20: 0.98, timestamp: 2025-06-10T15:00:0008:00 } return data app.tool() def generate_strategy_signal(market_data: dict, portfolio_type: str) - dict: 基于行情数据计算买入/卖出/持有信号返回信号强度和仓位置信度。 close market_data[close] latest close[-1] ma5 market_data[ma5] signal hold strength 0.5 if latest ma5 * 1.02 and portfolio_type 进取型: signal buy strength 0.8 return { signal: signal, strength: strength, reason: f最新收盘价较5日均线偏离{ (latest/ma5 - 1) * 100:.2f}% } if __name__ __main__: run_server(app)这段代码的关键在于工具函数用装饰器注册输入输出都是可序列化的dictDify 工作流调用后会直接拿到结构化的 JSON不需要再做额外解析。注意portfolio_type这个参数必须由工作流从输入字段透传进来否则工具内部无法判断策略风格。实际配置 MCP 工具节点时需要在 Dify 中填写 MCP 服务器地址。Dify 在运行工作流时会在后端发起连接所以要确保 MCP 服务的网络可达。如果是本地开发用127.0.0.1加端口即可如果是部署在云上建议给 MCP 服务配上内网地址避免公网暴露。调用顺序上先调用get_market_data拿到行情再把返回结果作为generate_strategy_signal的market_data参数。在 Dify 节点配置里下拉框选择“引用上游节点输出”就能自动映射不需要手写代码。3.3 第三步用 LLM 节点生成组合策略和推送文案信号计算出来之后接下来要做的是把“信号”翻译成用户看得懂的“策略建议”。这一步我用 Dify 的 LLM 节点让它根据信号结果生成仓位建议和推送文案。LLM 节点的 Prompt 设计有讲究。我不会让它自由发挥而是给出非常明确的结构和约束。参考 Prompt 如下你是一位合规的金融理财助手。请根据以下策略信号生成组合策略建议 信号结果{{signal_result}} 用户风险偏好{{portfolio_type}} 输出格式要求 1. 先用一句中文总结当前市场状态 2. 再给出建议仓位比例股/债/现金合计必须等于100% 3. 最后附上一句风险提示提示用户“策略基于历史数据不构成投资建议” 不要输出任何收益承诺不要提到具体个股推荐。Prompt 中出现的{{signal_result}}是 Dify 的变量引用语法会在运行时替换为上游节点的输出。这里有两个必须控制的点第一是仓位比例合计 100%我在 Prompt 里写死后后续可以在工作流代码节点里再做数值校验第二是合规措辞“不构成投资建议”这句必须让模型无脑带出来所以把它直接写进 Prompt 要求的第三条。LLM 节点生成的结构化建议会传递到后续推送工具节点。我还会在节点输出结果中要求返回suggestion和risk_tips两个字段这样推送到微信时可以分开渲染。3.4 第四步定时触发与服务 API 配置策略推送需要每天固定时间运行不可能手动点“运行”按钮。Dify 工作流支持配置“定时触发”在应用设置里选择 Cron 表达式例如0 17 * * 1-5表示工作日下午五点整触发。定时触发时工作流的输入字段user_id需要一个默认值。常见做法是先配置一个“管理员用户”作为默认接收人等验证通过后再通过服务 API 动态传入不同用户的标识。Dify 的服务 API 可以让你从外部系统调用工作流。用 Python 请求示例import requests url https://your-dify-instance/v1/workflows/run headers { Authorization: Bearer app-xxxxxxxxxxxxx, Content-Type: application/json } payload { inputs: { user_id: user_001, portfolio_type: 平衡型 }, response_mode: blocking } resp requests.post(url, jsonpayload, headersheaders) print(resp.json())代码中response_mode我一般用blocking这样可以直接拿到工作流最终输出结果方便判断推送是否成功。如果策略链路很长、耗时超过 30 秒建议改用streaming模式按节点逐段接收运行状态避免等待超时。Bearer后面的 token 在 Dify 的“API 访问”页面生成注意这不是模型供应商的 Key而是这个工作流应用自己的访问凭证。4. 微信端推送实现把 Dify 工作流接到微信 API 的完整链路4.1 微信端推送的三条路线怎么选微信生态里能“自动化推送”的通道不止一个我实际对比过的有三条路线企业微信应用消息、微信公众号模板消息、微信客服消息。三者的差异主要在接口权限、消息格式和触发条件上。企业微信应用消息是最适合自建系统做自动化推送的。它的接口可以直接给指定用户发文本卡片不需要用户先关注、不需要模板审核只要拿到用户的userid就能推送。公众号模板消息推送更偏向运营场景需要申请模板、模板内容要经过类目审核适合给关注了公众号的粉丝做定期服务。微信客服消息则适合“用户先找过来、你再回复”的被动场景不适合主动推送。我的建议是业务验证期优先走企业微信应用消息因为权限链路短、调试快等验证了推送带来的真实价值后再考虑公众号模板消息做更大范围触达。这篇采用企业微信应用消息作为主通道。4.2 企业微信应用消息的最小实现创建一个企业微信自建应用拿到corpid、agentid和secret后推送分两步获取access_token然后调用message/send接口。最小实现如下import requests CORP_ID your_corp_id AGENT_ID your_agent_id SECRET your_app_secret def get_access_token(): url https://qyapi.weixin.qq.com/cgi-bin/gettoken params {corpid: CORP_ID, corpsecret: SECRET} resp requests.get(url, paramsparams, timeout5).json() if resp.get(errcode) ! 0: raise RuntimeError(f获取access_token失败: {resp}) return resp[access_token] def send_wecom_text(token, user_id, content): url https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token token payload { touser: user_id, msgtype: text, agentid: int(AGENT_ID), text: {content: content} } resp requests.post(url, jsonpayload, timeout5).json() if resp.get(errcode) ! 0: raise RuntimeError(f消息发送失败: {resp}) return resp这里最容易出问题的是agentid类型。企业微信接口要求它是整型但很多配置项里存的是字符串直接传字符串会报invalid agentid。我在代码里用int()显式转换避免类型不匹配的坑。另外touser支持多个用户用竖线分隔例如user1|user2。但一条消息最多 100 人超过就需要分批发。策略推送通常是一对一内容所以这里保持单用户发送。4.3 把推送工具封装成 MCP 工具并接入 Dify企业微信的发送 API 不适合直接在 Dify 的 HTTP 节点里裸调因为 token 缓存、错误重试、频率控制这些逻辑放工作流里会很笨重。正确做法是把它封装成一个 MCP 工具push_wecom_message然后在 Dify 工作流最后一步调用。app.tool() def push_wecom_message(user_id: str, title: str, content: str, risk_tips: str) - dict: 通过企业微信应用消息推送理财策略报告。 token get_access_token() # 内部会做缓存避免每次请求都获取token message f【{title}】\n{content}\n\n{risk_tips} result send_wecom_text(token, user_id, message) return { send_status: result.get(errcode) 0, errcode: result.get(errcode), errmsg: result.get(errmsg) }封装的好处是Dify 工作流里只需要一个节点就能完成推送状态返回统一是结构化 JSON。send_status会被后续的“条件分支”节点读取如果发送失败可以走重试分支或记录日志。Dify 配置该节点时输入参数title、content、risk_tips都引用上游 LLM 节点的输出字段。推送时间一般耗时在几秒到十几秒工作流节点超时时间建议设置成 30 秒。4.4 消息模板字段与频率限制规避企业微信对应用消息的频率限制是“每应用每分钟最多 200 条”。策略推送一般每天只有一次所以限频本身不会触发但有一种情况会同一个用户信息在测试环境重复推送多次而且每次推送都触发了微信 API 的 token 刷新容易把调用量放大。我的习惯是生产环境的access_token必须做缓存有效期 7200 秒。线程安全的写法是token_cache {token: None, expire_at: 0} def get_access_token(): import time now time.time() if token_cache[token] and now token_cache[expire_at] - 60: return token_cache[token] # 重新获取并写入缓存 token_cache[token] _fetch_token() token_cache[expire_at] now 7200 return token_cache[token]做了一层 60 秒的提前过期保护避免 token 刚好在边界时失效。经验是尽量别在推送高峰期的最后几秒续期否则错过一次推送用户就会在微信上问“今天的策略去哪了”。5. 避坑指南智能理财助手上线前最容易踩的六个坑5.1 策略信号分钟级变化推送却是日级频率结果对不上现象行情信号工作流跑了多次用户收到的推送内容里策略信号是中午的推送时间却是下午五点用户疑问“这建议还成立吗”。原因定时触发只负责启动工作流但行情数据获取节点拿到的是运行时刻的最新价而不是策略决策时点的收盘价。不同节点取数时间不一导致信号和消息对不上。解决在get_market_data工具中显式传入as_of_date参数强制拉取特定交易日的日线收盘数据而工作流的定时时间设定在收盘后确保数据源已经更新。5.2 MCP 工具超时导致整个工作流卡死现象某次行情源响应很慢MCP 工具自己没设超时Dify 工作流一直停在“调用工具”状态直到整个流程超时失败。原因MCP 服务端函数没有设置网络请求超时同时 Dify 节点的超时时间设置过长掩盖了上游慢接口的问题。解决MCP 工具内部所有网络调用都显式加timeout同时在 Dify 节点配置中把超时时间压到 15 秒。一旦超时就走“失败-降级”分支推送一条“策略暂时不可用”的提示而不是让用户干等。5.3 文案里的“建议”踩到金融合规边界现象LLM 生成文案里出现“该组合预计年化收益 12%”“这只基金值得重仓”这类表述被安全审查打回。原因Prompt 约束不到位模型把“策略建议”理解成了“收益保证”。金融场景里任何收益预测和个股推荐都需要极强的前提限定营销话术不适合直接进推送。解决在 Prompt 中追加硬性负向限制例如“禁止出现收益预测”“禁止出现具体产品推荐”“必须包含风险提示”同时在 LLM 节点输出后加一个代码节点做关键词过滤命中“收益”“增持”“卖出”等高风险词时拦截消息并替换为模板话术。5.4 时区与交易日历错乱导致推送日期不准现象部署在云服务器上默认时区是 UTC在 0 点多运行时算出来的“今日策略”推送日期却是前一天的。原因Python 服务里datetime.now()返回的是服务器时区时间而不是北京时间同时没有判断当天是否交易日节假日也照常推送。解决所有时间相关逻辑统一用zoneinfo.ZoneInfo(Asia/Shanghai)获取当前时间再接入交易日历数据源非交易日直接跳过推送工作流返回{skipped: true}不产生微信消息。5.5 微信 API 返回错误码但工作流仍显示成功现象消息发送失败但 Dify 工作流运行记录里显示成功日志也没报错。原因MCP 工具内部没有检查企业微信返回的errcode直接把 HTTP 200 当成了成功返回。企业微信业务失败时 HTTP 状态码仍然是 200只有errcode是非零。解决push_wecom_message内部严格判断errcode 0否则抛异常。Dify 节点捕获异常后进入“失败分支”把错误信息写入审计日志并触发补偿流程。5.6 隔了好久改 Prompt历史策略推送记录对不上现象运营想对比“改版前后两种策略文案的效果”但旧文案已经无法复现因为 Dify 里只保留了最新配置。原因没有在工作流运行前给当前 Prompt 和参数打版本快照。Dify 的运行记录能看中间结果但 LLM 节点的 Prompt 是动态读取的改动后历史运行的输入配置变了。解决在 MCP 工具中加一个save_strategy_version工具每次工作流启动时把当前 Prompt、模型参数、策略类型存入数据库。审计时用version_id关联历史推送记录保证可追溯。6. 进阶加一层灰度验证让推送系统先证明自己再全量发系统能跑通“行情 → 信号 → 文案 → 推送”之后最值得做的一件事就是灰度验证。我见过太多项目链路通了就全量上线结果第一次推送就出问题影响一堆用户。做法分三步。第一步是“个人内测”工作流输入user_id固定填自己的企业微信 ID推送内容和线上完全一致重点是验证消息到达率、显示格式和文案观感。第二步是“影子推送”策略照常运行但同时把推送结果写一份到本地日志不实际发给用户等积累一周后拿这些影子数据跟真实市场走势做对比看策略建议的胜率和回撤是否符合预期。第三步才是“小流量灰度”按用户 ID 哈希取模先放 5% 的用户观察两三天。灰度比例控制我一般放在 Dify 工作流入口的“条件分支”节点里实现。输入新增一个gray_ratio参数判断逻辑是hash(user_id) % 100 gray_ratio才走完整推送链路否则只记录日志不推送。这样灰度比例从 5% 调到 10%、50% 只需要改一个参数不用改代码。灰度期间重点盯两个指标一个是推送送达率也就是send_status为 true 的比例另一个是用户点击或取消关注的反馈行为。如果送达率低于 95%先查 MCP 工具的错误码分布和 token 缓存是否失效。别急着查策略逻辑——大部分推送问题出在触达链路而不是策略模型。我自己的教训是上线后最容易翻车的是“策略挺好消息没发出去”。后来养成一个习惯每天推送完成后在工作流末尾加一个“健康检查”节点把当天的push_success_count、failed_user_list汇总推到自己的企业微信上几秒钟就能看出来系统是不是正常。你如果也在做类似的智能理财助手推送系统建议把这个习惯保留下来能省下不少半夜排查的时间。希望帮到你。本文还有配套的精品资源点击获取