Dify 加 MCP 实战：搭建金融理财助手自动化推送系统

简介这份PDF资料面向具备一定Python基础、希望深入AI智能体开发的金融科技从业者围绕Dify与MCP协议讲解如何构建可执行金融操作的理财助手打通行情分析、策略生成到微信端自动推送的全链路。内容涵盖金融助手核心能力架构、MCP智能体协议配置、Dify工作流搭建、微信消息路由服务与RAG知识库集成并给出关键代码片段同时深入实时行情获取、资产组合分析引擎与微信模板消息生成等实现细节还分享部署压测结果与一键部署指南。资源包为单个PDF文件共1个文件约248KB便于集中阅读与对照实践。目前已有262人学习。读者可借此掌握各模块交互逻辑理解意图识别、工具链调用与消息格式化的具体做法并获得高并发稳定性与合规性方面的排错思路适合边读边动手复现。1. 金融理财助手为什么需要 Dify 加 MCP 这套组合微信群里每天都有客户问“今天有什么稳健的配置建议”靠人工一条条回既慢又容易漏。我去年帮一个做投顾的朋友搭过一套自动化推送系统核心思路就是用 Dify 做策略编排和话术生成用 MCP 把行情数据、组合计算、微信推送这几个外部能力接进来最后跑成一个能定时触发、按客户标签分发的闭环。这套方案解决的不是“AI 能不能写理财文案”而是“策略从哪来、数据怎么进、消息怎么出”这三个工程问题。适合有基础 Python 能力、手头有 Dify 环境、想给现有理财业务加一层自动化推送的开发者。读完你能自己搭出一条从行情拉取到微信端送达的最小链路也能看清哪些参数必须调、哪些坑一定会踩。2. Dify 工作流与 MCP 协议的分工边界2.1 为什么不让 Dify 直接调微信接口很多人第一反应是在 Dify 的代码节点里直接写 requests 调微信 API我一开始也这么干过结果翻车了。Dify 的代码节点运行在沙箱里出网策略、超时时间、依赖库版本都不受你控制一旦微信侧要求签名校验或者需要维护 access_token 缓存代码节点就会变得又长又脆。更关键的是Dify 工作流本身是“无状态编排”它擅长的是把 LLM 调用、条件分支、变量赋值串起来而不是做长连接和凭证管理。MCP 的定位正好补上这一块。MCP 是 Model Context Protocol你可以把它理解成一套标准化的“工具插座”Dify 作为客户端通过 MCP 协议去调用外部 MCP Server 暴露的工具。行情查询、组合计算、微信消息发送各自封装成独立的 MCP ServerDify 只负责在合适的节点调用合适的工具。这样职责清晰微信侧的 token 刷新、重试、限流都放在 MCP Server 里Dify 工作流保持干净。常见做法是Dify 工作流负责“决策”MCP Server 负责“执行”。决策包括判断今天是否交易日、客户风险等级是否匹配、推送话术用哪套模板执行包括拉取实时行情、计算组合权重、调用微信接口发送。两者通过 MCP 的 tools/call 请求通信Dify 侧只需要配置好 MCP Server 的地址和工具名。2.2 用 Docker 把 Dify 和 MCP Server 跑在同一台机器上本地验证阶段我一般用 Docker Compose 把 Dify 社区版和一个自写的 MCP Server 放在同一网络里。Dify 的安装用官方 docker-compose.yamlMCP Server 单独写一个 Dockerfile。这样 Dify 容器里配置 MCP Server 地址时直接用服务名而不是 localhost避免容器网络不通的玄学问题。# 目录结构 # dify/ 官方 docker-compose.yaml 所在目录 # mcp-finance/ 自写 MCP Server # Dockerfile # server.py # requirements.txt # 在 dify 目录下启动 Dify cd dify docker compose up -d # 在 mcp-finance 目录下构建并启动 MCP Server cd ../mcp-finance docker build -t mcp-finance:latest . docker run -d --name mcp-finance \ --network dify_default \ -p 8765:8765 \ -e WECHAT_APPIDyour_appid \ -e WECHAT_SECRETyour_secret \ mcp-finance:latest逻辑说明--network dify_default让 MCP Server 加入 Dify 创建的默认网络Dify 容器内就能用http://mcp-finance:8765访问它。端口映射到宿主机是为了方便你用 curl 单独调试 MCP Server正式环境可以去掉-p。环境变量传微信凭证避免硬编码在代码里。参数说明WECHAT_APPID和WECHAT_SECRET来自微信公众平台用于获取 access_token。MCP Server 启动后监听 8765Dify 侧配置 MCP 连接时填http://mcp-finance:8765/sse或对应的 streamable HTTP 端点具体路径取决于你用的 MCP SDK 版本。提示Dify 社区版 1.10 之后对多租户和 MCP 连接的支持有调整如果你用的是更早版本MCP 配置入口可能在“工具”而不是“外部知识库”里先确认版本再动手。3. 把行情、组合计算、微信推送封装成 MCP Server3.1 MCP Server 的最小实现三个工具函数MCP Server 的核心是暴露工具。我用 Python 的 mcp 库写一个最小版本包含三个工具get_market_data拉取行情calc_portfolio计算组合权重send_wechat_message发送微信模板消息。每个工具用装饰器注册参数用 Pydantic 模型定义这样 Dify 侧调用时能自动拿到参数 schema。# server.py from mcp.server.fastmcp import FastMCP from pydantic import BaseModel import requests import time mcp FastMCP(finance-assistant) class MarketQuery(BaseModel): symbols: list[str] period: str 1d class PortfolioInput(BaseModel): symbols: list[str] weights: list[float] capital: float class WeChatMessage(BaseModel): openid: str template_id: str data: dict mcp.tool() def get_market_data(query: MarketQuery) - dict: 拉取指定标的的行情数据 # 这里替换成你实际使用的行情源 result {} for sym in query.symbols: result[sym] {price: 100.0, change_pct: 0.5} return result mcp.tool() def calc_portfolio(inp: PortfolioInput) - dict: 按权重计算组合预期收益和波动 total sum(inp.weights) if abs(total - 1.0) 1e-6: return {error: weights must sum to 1} expected sum(w * 0.05 for w in inp.weights) return {expected_return: expected, capital: inp.capital} mcp.tool() def send_wechat_message(msg: WeChatMessage) - dict: 发送微信模板消息 token get_access_token() url fhttps://api.weixin.qq.com/cgi-bin/message/template/send?access_token{token} payload { touser: msg.openid, template_id: msg.template_id, data: msg.data } resp requests.post(url, jsonpayload, timeout5) return resp.json() def get_access_token() - str: # 实际项目里要做缓存这里简化 appid os.environ[WECHAT_APPID] secret os.environ[WECHAT_SECRET] url fhttps://api.weixin.qq.com/cgi-bin/token?grant_typeclient_credentialappid{appid}secret{secret} return requests.get(url, timeout5).json()[access_token] if __name__ __main__: mcp.run(transportsse, host0.0.0.0, port8765)逻辑说明FastMCP自动把函数签名转成 MCP 工具定义Dify 侧连接后能看到get_market_data、calc_portfolio、send_wechat_message三个工具。get_market_data里我用了占位数据实际要接你的行情源注意加超时和异常捕获否则行情源抖动会拖垮整个工作流。参数说明symbols是标的代码列表period控制周期weights必须归一化我在函数里做了校验不满足直接返回 errorDify 侧可以根据这个分支走兜底话术openid是微信用户标识template_id是模板消息 IDdata是模板占位符内容。3.2 在 Dify 工作流里挂载 MCP 工具并做变量赋值Dify 工作流里先加一个“工具调用”节点选择 MCP 类型填入 MCP Server 地址。连接成功后节点里会出现三个工具。我一般把get_market_data放在工作流前段calc_portfolio放中段send_wechat_message放末段。中间用“变量赋值”节点把上一步的输出映射到下一步的输入。# Dify 工作流关键节点配置示意非完整 YAML仅说明字段 nodes: - id: start type: start variables: - name: customer_openid type: string - name: risk_level type: string - id: fetch_market type: tool provider: mcp tool: get_market_data inputs: symbols: [000001, 000300, 511880] period: 1d outputs: - name: market_data - id: calc type: tool provider: mcp tool: calc_portfolio inputs: symbols: [000001, 000300, 511880] weights: [0.4, 0.4, 0.2] capital: 100000 outputs: - name: portfolio_result - id: llm_generate type: llm prompt: | 根据以下行情和组合计算结果生成一段适合微信推送的理财建议 行情{{market_data}} 组合{{portfolio_result}} 客户风险等级{{risk_level}} - id: send type: tool provider: mcp tool: send_wechat_message inputs: openid: {{customer_openid}} template_id: your_template_id data: first: 今日组合建议 keyword1: {{portfolio_result.expected_return}} remark: {{llm_generate.text}}逻辑说明fetch_market和calc两个工具节点的输出通过变量名传给 LLM 节点LLM 生成话术后再传给send节点。risk_level从 start 节点传入可以在 LLM prompt 里做条件话术比如低风险客户只推货币基金建议。参数说明weights我写死在配置里实际项目应该根据客户风险等级动态计算可以在前面加一个代码节点生成权重数组。template_id要换成你微信公众平台里实际创建的模板 IDdata里的字段名必须和模板占位符完全一致否则微信侧会报 47003 错误。注意Dify 工作流里变量引用用{{}}但 MCP 工具输入如果是数组或对象要确认 Dify 版本是否支持直接传 JSON部分版本需要先序列化成字符串再在 MCP Server 里反序列化。4. 微信端推送链路的避坑与排查4.1 现象Dify 调用 MCP 工具超时工作流卡在工具节点原因MCP Server 默认没有设置请求超时Dify 侧工具节点默认超时可能是 30 秒但行情源响应慢或者微信接口限流时MCP Server 内部 requests 调用没有超时导致整个链路挂住。解决在 MCP Server 的每个 requests 调用里显式加timeout5并且在 Dify 工具节点配置里把超时调到 10 秒。如果行情源本身不稳定加一层本地缓存比如用functools.lru_cache缓存 60 秒内的行情结果。4.2 现象微信模板消息发送成功但用户没收到原因微信模板消息有“防骚扰”限制同一用户短时间内收到多条会静默丢弃或者用户已经取关。另外openid如果来自不同公众号也会发送失败。解决在 MCP Server 里记录每个 openid 的发送时间做频率控制比如同一用户 24 小时内最多推 3 条。发送前先调微信的“获取用户信息”接口确认关注状态取关的直接跳过。openid必须和template_id属于同一个公众号。4.3 现象Dify 工作流里 LLM 生成的话术包含未替换的变量原因LLM 节点的 prompt 里用了{{market_data}}但上一步工具节点的输出变量名不是market_data或者输出是嵌套对象LLM 拿到的是[object Object]。解决在 Dify 的变量赋值节点里把工具输出显式映射成扁平字符串。比如market_data是 dict用代码节点转成 JSON 字符串再传给 LLM。我一般会在 LLM prompt 里加一句“如果变量为空输出默认话术”避免空值导致生成中断。4.4 现象Docker 网络里 Dify 容器访问 MCP Server 报 connection refused原因MCP Server 容器启动时没有加入 Dify 的网络或者 MCP Server 监听的是 127.0.0.1 而不是 0.0.0.0。解决docker run时加--network dify_defaultMCP Server 代码里mcp.run(host0.0.0.0)。用docker exec -it dify-api curl http://mcp-finance:8765/sse在 Dify 容器内测试连通性能通再配到工作流里。4.5 现象Dify 升级后 MCP 连接配置丢失原因Dify 社区版升级时数据库迁移可能重置部分插件配置MCP 连接信息如果存在环境变量里而不是数据库里升级后需要重新配。解决把 MCP Server 地址和工具名写进 Dify 的环境变量文件.env升级后检查docker compose config确认变量还在。更稳妥的做法是 MCP Server 用固定域名或固定容器名Dify 侧配置一次就不用改。5. 让推送策略可验证回测与灰度发布的具体做法5.1 用历史数据回测组合策略再上线直接让 LLM 生成策略就推给客户风险太大。我一般会在 MCP Server 里加一个backtest_portfolio工具输入历史区间和权重输出累计收益和最大回撤。Dify 工作流在正式推送前先调这个工具跑一遍最近 30 个交易日如果最大回撤超过客户风险等级对应的阈值就降级成货币基金建议。mcp.tool() def backtest_portfolio(inp: PortfolioInput, start: str, end: str) - dict: 用历史数据回测组合表现 # 拉取历史净值计算累计收益和最大回撤 nav_series fetch_history(inp.symbols, start, end) portfolio_nav sum(w * nav for w, nav in zip(inp.weights, nav_series)) cum_return portfolio_nav[-1] / portfolio_nav[0] - 1 peak max(portfolio_nav) max_drawdown min(nav / peak - 1 for nav in portfolio_nav) return {cum_return: cum_return, max_drawdown: max_drawdown}逻辑说明fetch_history是你接的历史行情函数返回每个标的的净值序列。组合净值按权重加权再算累计收益和最大回撤。Dify 侧拿到max_drawdown后用条件分支判断是否超过阈值。参数说明start和end用YYYY-MM-DD格式建议至少取 30 个交易日样本太少回撤指标不可信。阈值我一般设低风险客户 -3%中风险 -8%高风险 -15%超过就触发降级。5.2 灰度发布先推给自己再推给客户新策略上线前我会在 Dify 工作流里加一个“测试模式”开关。开启时send_wechat_message的openid固定成我自己的模板消息里加“[测试]”前缀。跑一周确认话术、数据、送达都正常后再按客户标签分批放开。Dify 的变量赋值节点可以读一个test_mode变量从 start 节点传入这样不用改工作流结构。灰度期间重点看三个指标MCP 工具调用成功率、微信模板消息送达率、LLM 生成话术的变量替换完整率。成功率低于 95% 就回滚别硬撑。我自己的习惯是每次改完工作流先用 curl 直接调 MCP Server 的三个工具各一次确认底层没问题再去 Dify 里跑完整链路。这样排查问题时能快速定位是 MCP 层还是 Dify 层。希望帮到你。本文还有配套的精品资源点击获取