MCPModel Context Protocol这个协议火了将近一年之后大家终于发现一个更现实的问题单个工具接入 MCP 很好搞但一旦涉及“多步骤、定时调度、依赖关系、失败重试”这类生产级工作流事情就没那么优雅了。Airflow 这类工作流引擎还在老老实实按 DAG 调度任务MCP 这边却希望 Agent 能成为流程里的执行者——两套体系怎么融合是我最近一直在折腾的事。这篇文章我会从 MCP 的底层模型讲起再把 MCP 塞进 Airflow 这类工作流引擎里的几种落地方案、关键代码思路、以及我踩过的坑都整理出来。适合谁看两类人一类是已经在用 Airflow、想给工作流加上 AI Agent 能力的技术负责人另一类是刚接触 MCP想知道这个协议除了接个聊天工具之外还能干嘛的开发者。全文不废话直接上干货。1. MCP 为什么值得工作流引擎“重做一遍”1.1 MCP 到底解决了什么问题MCP 全称 Model Context Protocol是 Anthropic 在 2024 年底开源的一个开放协议。它解决的痛点用一个生活类比就讲得清MCP 之于 AI 应用就像 USB-C 接口之于电子设备。以前每个外设都要自己的一根线、一个驱动现在接口统一插上就能用。MCP 就是给 LLM 应用提供的统一接口标准任何数据源、任何工具只要实现一个 MCP Server所有兼容 MCP 的 HostClaude Desktop、Cursor、自研 Agent 等都能直接调用。协议本身的机制并不复杂底层是 JSON-RPC 2.0核心方法就那么几个initialize 做握手协商tools/list 获取工具清单tools/call 调用具体工具resources/read 读取外部数据。MCP Host 是 AI 应用本体MCP Client 是 Host 内部负责和 Server 通信的组件MCP Server 是暴露工具和数据的那一方。很多人第一次听到“mcp host 和 mcp server”会混淆记一个对应关系就行Host 是需要 AI 能力的一方Server 是提供工具能力的一方。Airflow 如果接入 MCP它既可以作为 Host 去调用外部工具也可以作为 Server 把自身的 DAG 操作暴露给 Agent后面我会展开讲。1.2 传统工作流引擎的断点在哪Airflow 的核心能力是调度编排DAG 定义任务依赖Scheduler 按时间触发Executor 执行失败自动重试还有完善的监控告警。这套体系对“稳定、确定”的任务非常可靠但一碰到需要动态决策的场景就露怯。举个例子一个常规的数据报表工单流程可能是“检查上游任务是否完成 → 拉取数据 → 清洗 → 写入数仓 → 发通知”所有步骤都是预设好的Airflow 干得很漂亮。但如果中间加一步“让模型判断这份数据有没有异常异常则自动去查询关联系统的错误日志”原来的代码结构就僵住了。传统做法是给外部系统一个个写 API 对接、写鉴权、写参数校验工具一多就要疯。这就是 MCP 的切入点把各种能力封装成标准化的工具后AI Agent 可以在工作流节点里自己选工具、自己传参数完成那些原本需要硬编码的动态决策环节。工作流引擎不需要知道工具内部实现只需要按 MCP 协议去调用。1.3 集成之后能拿到什么实际收益我梳理下来至少有三点收益是实打实的。第一是接入成本降低。以前接一个外部工具要给这个工具单独写 SDK 适配、鉴权逻辑、错误处理现在只要对方提供一个 MCP Server你的 Agent 或工作流就能直接调。团队里新增工具的边际成本趋近于零。第二是编排变得更灵活。DAG 的任务节点不再只能执行固定函数而是可以让 Agent 根据上下文动态挑选 tool。比如自动告警节点遇到不同告警类型会选择不同处理工具不再需要维护一堆条件分支。第三是可观测性提升。MCP 的请求响应是结构化 JSON-RPC每个工具调用的入参、出参、耗时、错误都能统一记录对审计和排查非常友好。这在传统 API 对接里往往是最容易被忽略的环节。2. 工作流引擎选型与集成架构设计2.1 为什么选 Airflow 而不是更年轻的编排框架社区里其实还有 Prefect、Dagster、Temporal 这类工具各有各的长处。Dagster 对数据资产的感知更强Temporal 更擅长长流程和分布式事务。但我依然选择拿 Airflow 作为讨论基准原因有三个Airflow 的调度和依赖模型足够成熟Cron 触发、外部传感器、任务依赖、重试策略都是行业标准生态大企业里存量系统很多大多数工程师对它不陌生云厂商都有托管版本部署运维成本可控。MCP 是一个非常新的协议底层调度基础设施就别再跟着一起冒险了。维度AirflowDagsterTemporal调度能力Cron 依赖 传感器成熟稳定偏数据资产驱动调度也不弱偏长流程定时能力较基础生态与人才存量最大资料最多快速增长但用户基数小适合分布式状态流上手门槛高MCP 集成案例社区讨论最多案例少需自研需自研重点在长任务适用场景通用批处理、数据管道、AI 流程编排数据平台、血缘追踪订单、状态机、长时间运行流程当然如果你是从零建设可以考虑 Dagster 的 asset-driven 模型对数据场景更友好。但从“把 MCP 融合进工作流”这个命题看Airflow 的讨论价值最大踩坑案例也最多。2.2 两种集成架构内嵌 Client 与独立 Gateway把 MCP 接进 Airflow我见过两种主流做法。第一种是 Airflow 内置 MCP Client在 DAG 里通过自定义 Hook 或 Operator 直接调用各个 MCP Server。这种方案结构简单适合 MCP Server 数量少、权限控制集中在 Airflow 层面的场景。缺点也很明显一旦 Server 多起来每个 DAG 里都要维护 URL、凭证、重试逻辑管不过来。第二种是独立部署一个 MCP Gateway所有 MCP Server 由 Gateway 统一管理Airflow 和外部 Agent 都只跟 Gateway 通信。Gateway 负责鉴权、路由、限流、日志。我个人更倾向第二种这跟在微服务架构里加 API 网关是同一个道理控制面和数据面分离后期治理会轻松很多。这里面有一个容易踩的坑很多人会把“接入 MCP”误解为“让 Airflow 变成万能 Agent”一股脑把 Agent 逻辑写进 DAG。实际上 DAG 节点里的 Agent 只应该做“动态决策”不应该承载对话交互长上下文沟通、多轮反思这类逻辑放在 Host 侧Airflow 只负责编排和调度。2.3 组件选型与传输方式MCP Server 的开发官方 Python SDK 里 FastMCP 足够好用装饰器定义工具几行代码搞定。传输方式需要按部署位置选本地进程用 stdio跨服务调用用 HTTP。HTTP 模式又分两种老一点的是 SSE新协议标准是 Streamable HTTP。我的建议是如果是新项目直接上 Streamable HTTP双向流式支持更好长任务不会因为 SSE 单向通道而断连。这两个词在热搜里出现频率不低后面我在问题排查部分会详细对比。凭证管理方面不管 Airflow 还是 Gateway都不要在 DAG 代码里硬编码 MCP Server 的 token。Airflow 自带 Connections 机制把 token 存在加密的 Connection 里运行时通过 Hook 读取Gateway 方案则用独立的 Secret Manager。很多人接了 Figma MCP 后问“token 在哪获取”其实都离不开这个原则token 永远只存在于服务端配置里绝不进代码仓库。2.4 反向集成把 Airflow 自己暴露成 MCP Server前面聊的都是 Airflow 作为调用方去调 MCP Server还有一种思路值得重视把 Airflow 自身的 DAG 能力封装成 MCP Server让外部 Agent 能通过 MCP 协议触发 DAG、查询任务状态。这样一来Agent 可以直接对工作流引擎发号施令而不是绕到 UI 或 API 层去操作。实现上也不复杂基于 FastMCP 封装一层薄薄的适配器把 trigger_dag_run、get_dag_status 这类操作映射成 MCP Tool。需要注意这个 Server 暴露的是操作能力必须严格鉴权否则任何人都能通过 Agent 触发生产任务风险极大。我见过一个团队这么做之后直接把 MCP Server 的 token 服务独立部署只允许内部网络访问算是比较稳妥的做法。3. 实操把 MCP Server 接进 Airflow DAG3.1 先定义好 Tool 的边界动手写代码之前最重要的事情是设计 MCP Tool 的边界。Tool 的 name 和 description 会被 LLM 当作“菜单”来挑选写得含糊Agent 就会乱选。我一般遵循三条规则每个 Tool 只做一件事宁可多拆几个 Tool 也不要做成“万能工具”description 写清楚这个工具能干什么、关键参数是什么、典型使用场景是什么入参尽量少出参尽量精简不要把整个数据库表都返回给模型。举个反例一个叫 execute_sql 的通用工具description 只写“执行 SQL”参数是 sql 字符串。LLM 在不确定表结构的时候会生成各种奇怪 SQL危险且低效。更好的做法是拆成 get_order_stats、get_user_behavior、update_inventory 这类语义明确、参数固定的小工具。3.2 写一个最小可用的 MCP Server下面这个示例模拟一个航班查询服务用 FastMCP 实现。代码示范了核心思路SDK 版本更新后 API 可能有微调以官方文档为准。from mcp.server.fastmcp import FastMCP mcp FastMCP(flight-service) mcp.tool() def search_flight(departure: str, arrival: str, date: str) - dict: 查询指定日期从departure到arrival的直飞航班列表返回航班号、起降时间、余票数。 # 实际项目里这里会调用业务系统或数据库 flights [ {flight_no: CA1234, dep_time: 08:00, arr_time: 10:30, seats_left: 15}, {flight_no: MU5678, dep_time: 12:00, arr_time: 14:20, seats_left: 3}, ] return {flights: flights} if __name__ __main__: # transport 可选 stdio、sse、streamable-http mcp.run(transportstreamable-http, host0.0.0.0, port8000)启动之后这个服务就暴露了一个标准 MCP 端点任何兼容客户端都能通过 tools/list 发现 search_flight通过 tools/call 执行它。项目里可以先用官方 inspector 之类工具做验证确认服务正常再接入工作流。3.3 在 Airflow 中通过 PythonOperator 调用 MCP ToolAirflow 接入 MCP 的代码思路其实不复杂本质就是把 MCP Client 的调用逻辑封装进 Python 可调用对象。下面的示例是异步调用方式的示意实际生产建议封装成自定义 Operator。import asyncio from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client async def call_mcp_tool_async(mcp_url: str, tool_name: str, arguments: dict): async with streamablehttp_client(mcp_url) as (read, write): async with ClientSession(read, write) as session: await session.initialize() result await session.call_tool(tool_name, argumentsarguments) return result.content def call_mcp_tool(mcp_url: str, tool_name: str, arguments: dict): return asyncio.run(call_mcp_tool_async(mcp_url, tool_name, arguments))然后在 DAG 里这样用from airflow import DAG from airflow.operators.python import PythonOperator with DAG( dag_idflight_status_pipeline, schedule_interval0 9 * * *, default_args{retries: 2, retry_delay: 300}, catchupFalse, ) as dag: fetch_flights PythonOperator( task_idfetch_flights, python_callablecall_mcp_tool, op_kwargs{ mcp_url: http://flights-mcp:8000/mcp, tool_name: search_flight, arguments: {departure: PEK, arrival: SHA, date: 2025-06-01}, }, )我特别强调两点。第一asyncio.run 在 Airflow worker 的同步执行环境里是可行的但要注意不要在已经运行的 event loop 里二次调用生产环境我一般直接用同步 HTTP 客户端封装 MCP JSON-RPC 请求反而更可控。第二arguments 里的日期等动态参数建议从上一步 task 的返回值里通过 XCom 传递而不是写死在 DAG 定义里。3.4 失败重试、幂等与上下文传递工作流接入外部工具后最常见的生产事故是“重试导致的重复操作”。Airflow 的 retries 机制很好但 MCP Server 侧不一定实现了幂等。我的建议是在设计 Tool 时强制要求幂等键例如提交工单类工具必须带 request_idServer 侧用这个字段去重。上下文传递方面MCP 本身是无状态的每次 tools/call 都是独立请求。所以 DAG 里如果有一个任务的输出要喂给下一个工具需要通过 XCom 把结果取出来再拼进下一个 task 的 op_kwargs。我在实际项目里会把“上一个节点输出 全局配置”统一加工成一个 context 字典传给工具调用函数减少参数散落。再有就是超时设置。LLM 工具调用往往比普通 API 调用慢如果工具里面有 Agent 循环单个调用可能耗时几十秒不能套用普通接口的超时策略。我这边会单独把调用 MCP Tool 的超时调大到 60 秒同时由 DAG 的 sla 机制兜底监控。4. 从真实场景看 MCP 工作流的落地方式4.1 Figma MCP设计稿变成流水线节点最近“figma mcp token在哪获取”“figma mcp 在 codex 中无法使用”这类搜索热度很高。Figma MCP 本质是官方提供的 Server它把 Figma 文件里的图层、组件、样式暴露成结构化数据。Token 是 Figma 的 Personal Access Token在 Figma 账号设置里生成配置到 MCP Client 的配置文件中。放到工作流引擎里它的价值就大了设计师交付设计稿 → Airflow 定时触发 → MCP 拉取最新设计稿的组件信息 → 前端代码生成节点自动产出页面骨架 → 推送到测试环境。整个链路里Figma 不再是“人看一眼然后手工写页面”的静态交付物而是流水线的一个数据源。4.2 代码、安全与 CI 工具协同热搜词里还出现了 codex 配置 mcp、burp mcp、jenkins mcp、cheat engine mcp 这类组合。这些本质上是同一件事原来只能通过 Agent 聊天界面手动操作的安全扫描、代码分析、CI 触发现在都变成标准 MCP Tool可以放进自动化流程。在内部我见过一个比较典型的用法Airflow 每天凌晨触发代码扫描流水线MCP Server 封装了静态扫描工具的调用扫描报告生成后再由 Agent 通过 MCP 调用缺陷管理系统的接口自动创建工单、分配负责人。整个过程没人参与但每个动作都有 JSON-RPC 日志可查。这对安全合规团队来说价值非常大。4.3 垂直行业数据从本地文件到票务行情“通达信股票软件本地数据 mcp”“12306 mcp”“mcp本地文件”这些搜索词说明很多人真正想解决的是“把私有数据接入 AI 工作流”。这类需求通常分成两种数据源型 MCP Server 和任务型 MCP Server。数据源型 Server 暴露的是查询能力比如本地行情数据库的查询接口Agent 只能读任务型 Server 暴露的是操作能力比如提交订单、写文件、调用接口。设计工作流时一定要严格区分这两种能力查询类工具可以给 Agent 较大自由度操作类工具必须做权限控制。4.4 RAG 和 MCP 别搞混“rag和mcp区别”也是高频搜索。RAG 解决的是知识检索核心是文档切分、向量化、相似度检索目标是从非结构化文档里找到相关内容MCP 解决的是工具调用目标是以标准协议读写数据、触发外部动作。两者的应用场景完全不同但可以配合使用Agent 先用 RAG 检索出操作手册里某段规则再通过 MCP 调用实际业务系统执行该规则。很多团队一开始把所有外部能力都堆进 RAG文档越接越多效果越来越差就是因为没搞清楚这一点。RAG 给模型“知识”MCP 给模型“手脚”两者各司其职。5. 接入过程中的常见问题与排查实录5.1 连接超时或握手失败SSE vs Streamable HTTP我遇到最多的报错是 Client 连不上 Server或者 initialize 之后连接就断。这里十有八九是传输方式不匹配。老版本 SDK 走 SSESSE 是单向推送Client 发请求可以Server 向 Client 推送消息要靠额外机制长任务时连接很容易断。Streamable HTTP 是双向流式更适合同步调用场景。对比项SSEStreamable HTTP通信方向单向推送客户端请求需另走通道双向流式请求响应同通道长任务适用性容易断连需额外重连逻辑更适合支持持续推送新项目推荐不推荐推荐常见报错连接中断、握手超时代理缓冲导致的延迟排查思路很简单先确认 Server 暴露的端点是 /mcp 还是 /sse用 curl 测一下握手请求再看 Client SDK 配置的 transport 是否和 Server 一致。另外如果 Server 部署在反向代理后面要确保代理支持 HTTP 流式响应Nginx 的 proxy_buffering 默认是开的不关掉会有诡异超时。5.2 MCP Tool 返回太慢把任务拖垮LLM 工具调用慢是常态我见过最夸张的 Tool 内部嵌套了多轮模型推理一次调用跑了 3 分钟。工作流引擎对任务时长很敏感处理方式有两个方向一是让 MCP Server 对耗时操作先返回“已受理”及任务 ID再通过 Streamable HTTP 的消息推送结果二是在 DAG 里把超时阈值调大并针对长任务单独设置 SLA 提醒。实际操作中我建议先用监控数据说话把每个 MCP Tool 的耗时曲线跑出来慢的单独优化而不是一刀切调大超时。调大超时只是延长了发现问题的时间窗口。5.3 上下文膨胀与结果截断Agent 场景下tools/call 的返回内容会进入模型上下文。如果一个查询工具一次返回几百行数据上下文很快用完后续 Agent 就没法正常推理了。解决思路是工具设计阶段就得克制返回字段精简只给核心指标支持 limit、filter 参数大型结果分段返回。在 Airflow 集成里我额外加了一个“结果摘要”节点MCP 返回原始数据后先经过摘要逻辑再把精简结果传给下游 Agent。数据全量存储保留在数仓模型上下文里只放结论。5.4 权限过大与安全问题最后说个严肃的。MCP 把工具调用变得太容易反而容易让人忽略权限。Agent 能调用工具不等于 Agent 能调用所有工具。我在项目里限制得很死工作流内使用的 MCP Server 全部走独立服务账号只授予流程必需的最小权限敏感操作工具比如删除、上线、转账必须有双人审批或额外确认参数。日志审计也不能省。MCP 的 JSON-RPC 日志天然适合做审计我在 Gateway 层把每次 tools/call 的方法名、入参、出参、调用方、耗时都落库出事时能快速回溯。5.5 工具验证与调试技巧最后分享一个我常用的调试路径。新接入一个 MCP Server 时先不要直接写进 DAG而是用 MCP 官方的 Inspector 或简单的 Python 脚本验证三件事tools/list 返回的工具清单是否符合预期tools/call 返回的数据结构是否稳定异常输入时返回的错误信息是否有提示性。这三步过了再接入工作流能省下大量排查时间。遇到“MCP Server 连接正常但 DAG 里调用失败”的情况优先检查 Airflow worker 所在网络能否访问 MCP Server 地址。很多公司开发机和生产 worker 不在同一个网络环境本地 curl 通不代表生产环境通。我在实际折腾中最大的感受是MCP 给了工作流引擎一双能干活的“手”但也把设计工具边界、权限控制、超时管理这些脏活累活重新摆到了桌面上。别指望接上 MCP 就万事大吉把 Tool 当接口来治理把 Agent 当受限的执行者来管理这套组合才能真正跑稳。如果后面社区对 MCP 在长任务、事务、编排语义上的支持能进一步完善工作流引擎这一层一定会长出不少新东西。