之前在做 MCPModel Context Protocol服务时我遇到过两个非常典型的问题网上能找到的 Demo 基本都是基于stdio传输方式离真实生产环境太远而涉及Streamable HTTP的文章又大多只讲解协议概念没有完整的 LangGraph 接入案例。断断续续折腾了大半个月后我整理出了一套从零搭建 Streamable HTTP/SSE 传输系统的完整流程并且把 LangGraph 工作流成功挂载到了 MCP Server 中。本文将围绕这条链路展开先说清楚 MCP、LangGraph、Streamable HTTP/SSE 到底是什么。再拆解 Streamable HTTP 的核心请求/响应模型和 SSE 长连接机制。然后从空目录开始手把手实现一个可运行的 MCP Server并在其中集成 LangGraph 工作流。最后给出高频报错排查表和一套可以直接落到生产环境的工程建议。无论你是刚接触 MCP 的新手还是已经在做 Agent 应用落地的后端开发者这篇文章都值得收藏。1. 背景与核心概念1.1 MCP模型上下文协议到底是什么MCPModel Context Protocol模型上下文协议是由 Anthropic 在 2024 年底提出的一种开放协议目的是让 AI 模型能够以标准化的方式连接外部工具、数据源和业务系统。你可以把它理解成“AI 世界的 USB-C 接口”模型不需要关心每个数据源私有 API 的实现细节只要通过统一的协议就能发现工具、调用工具、读取资源。从协议设计上看MCP 采用客户端-服务端架构MCP Client负责与大模型交互并把模型需要的工具列表、资源列表暴露给模型。MCP Server实现具体业务能力暴露 tools、resources、prompts 三类核心能力。在真实项目里MCP Server 往往是夹在“模型”和“企业系统”中间的适配层。比如一个客服机器人通过 MCP Server 调用订单查询接口、工单创建接口、库存查询接口模型只需要学会调用 MCP 工具而不需要理解每个系统的对接协议。1.2 LangGraph从 Chain 到 GraphLangGraph 是 LangChain 团队推出的低层级编排框架专门用于构建有状态、多步骤的 Agent 工作流。与传统的 Chain 线性串联不同LangGraph 允许你像画流程图一样定义节点和边支持分支、循环、人工介入、状态持久化等复杂能力。一个最小的 LangGraph 应用包含四个核心概念State工作流的全局状态通常是一个 TypedDict。Node一个执行单元输入 State输出 State 片段。Edge定义节点之间的流转关系可以是固定边也可以是条件边。Graph把节点和边组装成一张可执行的状态图最后编译成可运行对象。LangGraph 特别适合用来构建“意图识别 - 决策 - 工具调用 - 结果生成”这类多节点 Agent 工作流。MCP 负责工具接入LangGraph 负责流程编排两者结合后复杂业务可以被拆成一张状态图由 MCP Server 暴露给上层模型。1.3 Streamable HTTP/SSE 解决了什么问题在 MCP 协议的发展过程中传输层经历过几次调整。最早的传输方式是stdio适合本地进程通信但无法用于远程部署。后来官方提供了 HTTPJSON 传输但仍然存在几个痛点一次请求只能对应一次响应服务器没有办法向客户端推送渐进式的中间结果。长任务要么让客户端一直等待要么只能通过轮询方式获取状态。旧规范限制了 MCP 在远程生产环境中的可用性。于是 MCP 规范引入了Streamable HTTP传输模式它基于 HTTP SSEServer-Sent Events服务器发送事件实现。核心特征有两个第一客户端通过 HTTP POST 向服务器发送 JSON-RPC 2.0 请求。第二服务器的响应既可以是普通的 JSON也可以是一个text/event-stream的 SSE 流。在这个 SSE 流上服务器可以源源不断地把任务状态、中间结果、最终结果推送给客户端。这样既能保持 HTTP 协议普遍兼容的优势又能解决长任务流式输出的问题。本文的实战项目就以 Streamable HTTP/SSE 为重点。1.4 本文要搭建的完整链路为了避免文章停留在概念层面我准备了一个端到端可运行的样例一个客户支持场景的 LangGraph 工作流。用户提交一条任务文本后工作流会先进行意图分析再根据分析结果决定走简单回复分支还是深度处理分支最后生成答案。这个工作流整体暴露为一个 MCP 工具通过 Streamable HTTP 传输方式对外提供服务。链路如下MCP ClientHTTP 请求 - Streamable HTTP 传输层 - MCP ServerFastMCP - tool: run_customer_workflow - LangGraph 状态图执行2. 环境准备与项目结构2.1 环境与版本说明本文示例运行环境如下具体版本需要根据你的项目实际情况调整操作系统macOS / Linux / WindowsWSL2 均可。Python 版本3.10 或以上。MCP 相关库fastmcp0.4底层依赖mcp1.x。LangGraph 版本langgraph0.2。运行方式命令行启动 MCP Server使用 HTTP 客户端验证。需要说明的是MCP 和 LangGraph 的 API 都还在快速迭代中不同小版本的参数名称可能存在差异。如果你的版本和示例不一致优先阅读对应版本的官方文档本文重点演示的是思路和完整流程。建议使用虚拟环境隔离项目依赖cd mcp-langgraph-demo python -m venv .venv source .venv/bin/activate # Windows 下执行 .venv\Scripts\activate2.2 初始化项目结构与依赖创建一个干净的项目目录结构如下mcp-langgraph-demo/ ├── requirements.txt ├── workflow.py # LangGraph 工作流定义 ├── server.py # MCP Server 入口 └── client_test.py # 本地测试脚本requirements.txt内容如下fastmcp0.4.0 langgraph0.2.0 langchain-core0.3.0 uvicorn0.30.0 httpx0.27.0然后执行安装命令pip install -r requirements.txt如果安装速度较慢可以临时使用国内镜像源pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple有一点需要特别注意fastmcp的版本直接影响传输方式参数写法。早期版本使用transporthttp0.4 之后的版本才明确支持transportstreamable-http。安装完成后建议先确认版本pip show fastmcp3. Streamable HTTP / SSE 核心原理拆解3.1 JSON-RPCMCP 的消息骨架MCP 的所有请求和响应都使用 JSON-RPC 2.0 格式。一个典型的请求长这样{ jsonrpc: 2.0, id: 1, method: tools/call, params: { name: run_customer_workflow, arguments: { task: 用户咨询订单什么时候发货 } } }响应则有两种形态一种是普通 JSON{ jsonrpc: 2.0, id: 1, result: { content: [ { type: text, text: 根据当前订单状态预计明天下午发货。 } ] } }另一种是 SSE 流中携带多个 JSON-RPC 消息。我们需要理解 JSON-RPC 中的几个核心字段jsonrpc固定为2.0。id请求标识用于匹配响应。如果是通知消息则没有id。method需要调用的方法名比如initialize、tools/list、tools/call。params调用参数。result成功响应。error失败响应包含错误码和错误信息。3.2 Streamable HTTP 的请求-响应模型Streamable HTTP 传输模式下客户端需要向服务端的一个固定端点发送 POST 请求。这个端点通常就是 MCP Server 的入口地址例如 FastMCP 默认的/mcp/v1。一次完整的交互流程如下客户端发送initialize请求完成协议版本协商。服务端返回支持的能力列表和协议版本。客户端发送notifications/initialized通知告诉服务端初始化完成。客户端发送普通请求例如tools/list或tools/call。服务端返回响应可能是 JSON也可能是 SSE 流。在这个模型中有几点很关键第一initialize是必做步骤否则服务端会拒绝后续请求。第二如果服务端支持会话管理会在响应头中返回Mcp-Session-Id客户端后续请求需要携带该 header。第三请求头中的Accept字段通常是application/json, text/event-stream表示客户端既能接受普通 JSON也能接受 SSE 流。3.3 SSE 消息格式与长连接机制SSE 是 HTML5 标准中的一种服务端推送技术它允许服务器通过一个 HTTP 长连接持续向客户端发送数据。与 WebSocket 不同SSE 是单向的适合服务器向客户端推送状态更新。一个 SSE 流的基本格式如下event: message data: {jsonrpc: 2.0, id: 1, result: {content: [{type: text, text: step1 done}]}} event: message data: {jsonrpc: 2.0, id: 1, result: {content: [{type: text, text: step2 done}]}}每个事件之间用空行分隔。event: message表示事件类型data:后面是实际传输的数据。MCP 规范中服务端推送的 JSON-RPC 消息都是以message事件发送的。在 Streamable HTTP 中SSE 的意义在于一次 HTTP 请求可以携带多个 JSON-RPC 响应客户端不需要反复发起轮询请求。3.4 一次完整握手的报文演示下面用 curl 模拟一次完整的初始化握手。假设服务端已经启动监听 8000 端口。第一步发送initialize请求curl -X POST http://127.0.0.1:8000/mcp/v1 \ -H Content-Type: application/json \ -H Accept: application/json, text/event-stream \ -d { jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 2025-03-26, capabilities: {}, clientInfo: { name: curl-test, version: 1.0.0 } } }服务端会返回协议版本、服务器能力和会话 ID。第二步发送初始化完成通知curl -X POST http://127.0.0.1:8000/mcp/v1 \ -H Content-Type: application/json \ -H Accept: application/json, text/event-stream \ -H Mcp-Session-Id: 上一步返回的会话ID \ -d { jsonrpc: 2.0, method: notifications/initialized }第三步调用工具列表接口。之后就可以正常发送tools/call请求了。4. 从零搭建 MCP LangGraph 实战4.1 定义 LangGraph 工作流我们先编写workflow.py。这个文件负责定义一张完整的 LangGraph 状态图。业务逻辑分为两个主要节点analyze节点对用户输入进行意图分析。handle_simple节点如果任务比较简单直接生成回复。handle_complex节点如果任务比较复杂走深度处理分支。为了避免示例依赖大模型 API这里先用规则模拟分析和处理逻辑。实际项目中你可以在节点内部嵌入任何 LLM 调用。# 文件路径workflow.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): 工作流全局状态 task: str # 用户输入 analysis: str # 意图分析结果 level: str # simple / complex answer: str # 最终回复 def analyze(state: State) - dict: 意图分析节点 task state[task] if 订单 in task or 退款 in task: analysis 用户咨询订单相关问题 level simple elif 投诉 in task or 复杂 in task: analysis 用户投诉需要人工介入 level complex else: analysis 普通咨询可以自动回复 level simple return { analysis: analysis, level: level, } def handle_simple(state: State) - dict: 简单问题处理节点 return {answer: f已识别为简单问题{state[analysis]}。回复感谢咨询我们会在 1 小时内处理。} def handle_complex(state: State) - dict: 复杂问题处理节点 return {answer: f已识别为复杂问题{state[analysis]}。回复已升级给人工客服预计 2 小时内电话联系您。} def build_workflow(): 构建并编译 LangGraph 状态图 graph StateGraph(State) # 添加节点 graph.add_node(analyze, analyze) graph.add_node(handle_simple, handle_simple) graph.add_node(handle_complex, handle_complex) # 固定边 graph.add_edge(START, analyze) # 条件边根据 level 字段决定走向 graph.add_conditional_edges( analyze, lambda state: state[level], { simple: handle_simple, complex: handle_complex, }, ) # 收束到结束 graph.add_edge(handle_simple, END) graph.add_edge(handle_complex, END) return graph.compile() if __name__ __main__: app build_workflow() result app.invoke({ task: 我的订单还没有发货请帮我查一下, analysis: , level: , answer: , }) print(result[answer])在命令行先运行一次确认工作流本身是可用的python workflow.py预期输出类似已识别为简单问题用户咨询订单相关问题。回复感谢咨询我们会在 1 小时内处理。这里解释一下 LangGraph 的关键点。StateGraph(State)使用 TypedDict 定义状态结构add_node注册节点函数add_conditional_edges根据状态内容动态决定下一步走向。这种条件路由能力比传统的 Chain 更适合做复杂 Agent 编排。4.2 将工作流封装成 MCP 工具接下来编写server.py把build_workflow()编译出来的图封装成 MCP 工具。# 文件路径server.py from fastmcp import FastMCP from workflow import build_workflow # 创建 FastMCP 服务实例 # transport 指定为 streamable-http这是 MCP 远程部署的关键配置 mcp FastMCP( namecustomer-support-agent, transportstreamable-http, host0.0.0.0, port8000, ) # 启动时构建一次工作流避免每次调用都重新编译 workflow build_workflow() mcp.tool() def run_customer_workflow(task: str) - str: 执行客户支持工作流。 Args: task: 用户提交的问题描述例如“我的订单还没有发货”。 Returns: 工作流最终生成的回复内容。 result workflow.invoke({ task: task, analysis: , level: , answer: , }) return result[answer] if __name__ __main__: mcp.run()这段代码中包含几个重要设计第一FastMCP的transportstreamable-http表示服务以 HTTP 方式对外提供 MCP 能力并且支持 SSE 流式响应。第二工具函数使用mcp.tool()注册。工具的名称和描述会直接暴露给大模型所以描述要写清楚参数含义和适用场景。第三workflow在服务启动阶段构建一次避免每次工具调用都重新编译状态图提升响应速度。需要注意的是某些版本的 FastMCP 对装饰器的参数名支持不同有的版本允许在mcp.tool()中指定自定义名称例如mcp.tool(namecustomer_support) def run_customer_workflow(task: str) - str: ...如果你的版本报“unexpected keyword argument”错误去掉name参数即可。4.3 配置 Streamable HTTP 传输并启动服务启动命令很简单python server.pyFastMCP 会启动一个本地的 HTTP 服务。正常情况下控制台会输出类似下面的日志INFO - Starting MCP server on http://0.0.0.0:8000/mcp/v1注意端点路径如果你的版本默认不是/mcp/v1按实际日志调整。如果你需要自定义端点路径例如加一层前缀方便网关转发可以传入endpoint参数mcp FastMCP( namecustomer-support-agent, transportstreamable-http, host0.0.0.0, port8000, endpoint/mcp/v1, )实际生产环境中建议通过反向代理绑定域名并使用 HTTPS 对外提供服务。4.4 用 HTTP 请求验证服务服务启动后我们先用 curl 验证最基本的健康检查。直接请求根路径通常会返回 404这是正常的因为 MCP 端点固定为/mcp/v1。正确的做法是直接向该端点发送 POST 请求。先测试initializecurl -X POST http://127.0.0.1:8000/mcp/v1 \ -H Content-Type: application/json \ -H Accept: application/json, text/event-stream \ -d { jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 2025-03-26, capabilities: {}, clientInfo: {name: curl, version: 1.0.0} } }返回结果中会包含{ jsonrpc: 2.0, id: 1, result: { protocolVersion: 2025-03-26, capabilities: { tools: {listChanged: false} }, serverInfo: {name: customer-support-agent, version: 0.1.0} } }接着发送notifications/initialized如果你使用的服务端开启了会话保持需要带上Mcp-Session-Idcurl -X POST http://127.0.0.1:8000/mcp/v1 \ -H Content-Type: application/json \ -H Accept: application/json, text/event-stream \ -H Mcp-Session-Id: SESSION_ID \ -d {jsonrpc: 2.0, method: notifications/initialized}然后查询工具列表curl -X POST http://127.0.0.1:8000/mcp/v1 \ -H Content-Type: application/json \ -H Accept: application/json, text/event-stream \ -H Mcp-Session-Id: SESSION_ID \ -d {jsonrpc: 2.0, id: 2, method: tools/list}最后调用工作流工具curl -X POST http://127.0.0.1:8000/mcp/v1 \ -H Content-Type: application/json \ -H Accept: application/json, text/event-stream \ -H Mcp-Session-Id: SESSION_ID \ -d { jsonrpc: 2.0, id: 3, method: tools/call, params: { name: run_customer_workflow, arguments: { task: 我的订单还没有发货 } } }如果一切正常响应中会包含result.content[0].text字段内容就是工作流的最终答案。4.5 结果说明从上面的验证过程可以看到MCP 协议层面的链路已经完整跑通通过initialize完成协议握手。通过tools/list发现工具。通过tools/call调用工具。工具内部执行的是 LangGraph 状态图。如果你希望看到 SSE 流式输出效果有两种方式。第一种是直接使用 FastMCP 内置的异步工具函数让工作流在多个节点之间逐步返回结果第二种是在 MCP Server 之上再包装一层业务 API把 MCP 的 SSE 流透传给前端页面。对于本示例中的同步工具调用服务端返回的是普通 JSON因为整个工作流很快执行完毕。只有当中途存在耗时的 LLM 调用或外部 API 请求时SSE 流才会发挥真正的价值。5. 常见问题与排查清单5.1 高频问题速查表问题现象常见原因解决思路启动时报错Unsupported transportfastmcp 版本过低不支持 streamable-http升级 fastmcp 到 0.4 以上版本请求/mcp/v1返回 404端点路径不对查看启动日志中的实际路径返回 405 Method Not Allowed使用了 GET 请求MCP HTTP 端点只接受 POST返回 400Session not found缺少 initialize 或未携带会话 ID先完成 initialize再带上Mcp-Session-Id调用tools/call返回Tool not found工具名称拼写错误或工具未注册成功先调用tools/list确认名称LangGraph 调用报RecursionLimit工作流存在循环回路设置recursion_limit检查条件边逻辑工具函数内部日志不输出日志处理器被覆盖重新配置 logging或使用 print 临时排查5.2 高频问题详解第一个常见问题是版本兼容。FastMCP 版本迭代较快我最初使用 0.3.x 版本时transport参数只支持http并不支持streamable-http导致启动时报ValueError。解决方法是升级到 0.4 以上或者根据源码提示修改参数名。第二个常见问题是会话管理。在无状态 HTTP 模式下如果服务端返回了Mcp-Session-Id客户端必须保存并在后续请求中携带否则服务端会认为会话无效。如果你不希望客户端管理会话状态可以在服务端配置中关闭会话保持或者统一走无会话模式。第三个常见问题是 SSE 连接中断。SSE 依赖 HTTP 长连接如果服务前面挂了 Nginx 或云负载均衡需要调整代理的超时时间否则长任务执行到一半连接会被切断。建议将代理的proxy_read_timeout设置为大于任务最大执行时间。第四个常见问题是 LangGraph 初始状态字段缺失。State中定义的字段在执行时必须全部传入否则会抛出MissingRequiredField类型错误。一个保险的做法是在invoke()时显式传入所有字段的默认值。6. 最佳实践与工程建议6.1 工具命名与描述优化MCP 工具对大模型的可见性完全依赖工具名称和描述因此这两项必须写得足够清晰。举个例子下面的写法就不够明确mcp.tool(description处理任务) def run_workflow(task: str) - str: ...更好的写法是mcp.tool() def run_customer_workflow(task: str) - str: 执行客户支持工作流根据用户问题分析意图并生成回复。 Args: task: 用户提交的完整问题文本例如“我的订单还没有发货”。 Returns: 最终生成的客服回复文本。 ...描述中包含了业务场景、参数含义和返回值格式模型才能准确判断“什么情况下应该调用这个工具”。6.2 长任务与异步优化在真实业务中LangGraph 节点里通常会有大模型调用、数据库查询、外部 API 请求耗时可能达到几十秒。如果所有逻辑都在一个同步工具函数中完成客户端会长时间阻塞。推荐的做法是把 MCP 工具设计成异步任务模式客户端调用tools/call提交任务。服务端立刻返回一个任务 ID并告知“任务已接收”。服务端后台异步执行 LangGraph 工作流。客户端通过轮询接口或 SSE 流获取任务进度。在 FastMCP 中你可以把工具函数定义为async def并在内部把耗时操作放到后台执行。对于需要流式推送结果的场景则可以结合 SSE 逐步发送中间节点结果。6.3 安全与生产环境加固MCP Server 对外暴露后需要考虑以下几个安全点第一认证。在生产环境中不要让 MCP 端点裸奔。可以在 FastMCP 前加一层网关使用 API Key、OAuth2、JWT 等机制校验调用方身份。第二CORS 策略。如果前端页面会直接跨域调用 MCP Server需要配置allow_origins只允许指定域名访问。第三HTTPS。通过 Nginx 或云负载均衡强制 HTTPS避免请求内容在传输过程中被篡改。第四协议版本锁定。MCP 规范仍在演进protocolVersion要显式声明避免客户端和服务端使用不一致的子版本导致兼容问题。6.4 LangGraph 工程化要点当 LangGraph 工作流从 Demo 走向生产时有几个点值得注意状态持久化默认情况下LangGraph 的状态只存在内存中。如果服务重启工作流状态会丢失。生产环境建议使用langgraph-checkpoint将状态持久化到 SQLite、PostgreSQL 等存储中。超时控制给节点函数增加超时逻辑防止某个外部依赖长时间无响应拖垮整个工作流。日志追踪在每个节点的输入输出中写入结构化日志方便定位问题。条件边兜底add_conditional_edges的映射表最好包含一个默认分支避免出现意外状态值导致执行异常。7. 总结与学习路线这一篇文章围绕 MCP 和 LangGraph 做了完整的实战拆解。你不仅理解了 Streamable HTTP 和 SSE 的底层交互模型还从零实现了一个可运行的 MCP Server并且把 LangGraph 状态图封装成了标准 MCP 工具。接下来你可以继续深入这几个方向学习 MCP 官方规范中的resources和prompts能力完善 Server 的接入范围。尝试在 LangGraph 节点中接入真实大模型让工作流具备更强的推理能力。研究langgraph-checkpoint的状态持久化机制为生产部署做准备。探索 MCP Client 与前端页面的 SSE 对接方式实现 Web 端流式输出。如果在实践过程中遇到版本差异引起的 API 变更不要急着怀疑自己的代码先确认库版本和官方文档是否一致。技术迭代期掌握原理远比记忆 API 更重要。希望这篇文章能帮你把 MCP 和 LangGraph 真正落地到自己的项目里。