MCP与LangGraph构建Streamable HTTP服务并接入Agent

这次我们来看一条完整的技术落地路线：MCP + LangGraph。标题里的 Streamable HTTP、SSE、从零构建、手把手，这些词放在一起很容易劝退人，但拆开之后你会发现，它就是“写一个带 HTTP 接口的 MCP Server，再把工具交给 LangGraph Agent 去调用”这么一件事。

本文不是概念复读，我会按照实际做项目的顺序走一遍：先解释 MCP 和 Streamable HTTP/SSE 到底解决了什么问题，然后搭建环境、创建 MCP Server、用 curl 验证 JSON-RPC 协议，再把 MCP 工具接入 LangGraph Agent，最后跑通工具调用、流式输出和批量任务。文章包含可直接复制的代码、启动命令、验证脚本和问题排查表，适合想系统掌握 MCP 服务端开发与 LangGraph 编排的 Python 开发者。

先说结论：这个项目不需要本地 GPU，模型调用走 API，LangGraph 本身只是编排层。所以硬件门槛很低，门槛主要在“能不能把协议链路理解清楚、能不能把调试流程走通”。

1. 核心能力速览

这套方案最大的优势是“一次开发，多处接入”。MCP 把工具调用、资源读取、提示词模板都规范成了标准协议，LangGraph 负责把“调用哪个工具、什么时候调、结果怎么回流”的决策过程编排成状态图。两者结合后，大模型不再是只能聊天的对话框，而是一个能操作真实系统的执行器。

不要被“传输系统”四个字吓到。Streamable HTTP 解决的是“客户端和服务端之间怎么传数据”，SSE 解决的是“服务端怎么把长任务的结果实时推给客户端”。理解这两个东西，是本章节实战的核心。

2. 技术栈拆解：MCP、LangGraph、Streamable HTTP

2.1 MCP 是什么

MCP 是一个开放协议，用于统一“AI 应用”和“数据源/工具”之间的连接方式。它的架构包含三部分：

• Host：大模型应用，比如 LangGraph Agent。
• Client：Host 内部的通信客户端，负责和 Server 建立会话。
• Server：独立进程或服务，对外提供 tools、resources、prompts。

以前接入一个天气接口，需要在业务代码里写一遍 HTTP 调用；接入一个数据库，又要换一套驱动。MCP 的思路是让工具本身变成“可被发现、可被描述、可被调用”的标准资源，模型通过协议去查询工具清单、获取工具参数说明、发起工具调用。这套协议最关键的价值，是把工具从“写死代码”变成“动态接入”。

2.2 Streamable HTTP / SSE 传输

MCP 有多个传输方式。早期远程场景常用stdio，只适合本地子进程通信，不适合跨网络调用。远程场景经历过SSE over HTTP、HTTP+SSE，再到现在的Streamable HTTP。

Streamable HTTP 的核心特征是：

• 服务端提供一个 HTTP 端点，客户端通过 POST 发送 JSON-RPC 请求。
• 服务端根据请求情况返回普通 JSON，或者text/event-stream的 SSE 流。
• 长耗时任务（比如生成、分析、批量操作）通过 SSE 实时推送进度和结果。
• 支持会话管理，会话 ID 通过Mcp-Session-Id请求头传递。

SSE 是单向传输协议，只允许服务端向客户端推送。它和 WebSocket 的区别是：SSE 更轻、基于普通 HTTP、不需要额外的握手协议，适合“服务端有数据就向下推”的场景。MCP 的 Streamable HTTP 就是用 POST + SSE 组合来覆盖“请求-响应”和“服务端主动推送”两类需求。

2.3 LangGraph 干什么

LangGraph 是专门做 Agent 编排的库。它的核心是状态图：把一次 Agent 执行拆成多个节点，节点之间通过边连接，执行过程维护一个共享状态。

常用的节点编排是“LLM 节点 + ToolNode”：

• LLM 节点：大模型接收用户消息，判断是否要调用工具。
• ToolNode：执行工具，把结果写回状态。
• 条件边：如果 LLM 决定调用工具，就跳转到 ToolNode；否则直接结束。

MCP 负责“工具从哪来、怎么调用”，LangGraph 负责“工具在什么时候被调用、结果怎么处理”。两者分工明确，拼接点就是langchain-mcp-adapters，它能把 MCP Server 暴露的工具转换成 LangChain 的 Tool 对象。

3. 环境准备与前置条件

3.1 检查基础环境

本项目以 Python 为主，两个前置条件必须先确认：

• Python 版本 3.10 及以上。建议使用 3.11/3.12，兼容性最好。
• 一个可用的 OpenAI 兼容模型 API，用于驱动 LangGraph Agent。可以使用 OpenAI、DeepSeek、通义千问或其他兼容接口，地址和 Key 在环境变量里配置。

先创建项目目录和虚拟环境：

mkdir mcp-langgraph-demo cd mcp-langgraph-demo python -m venv .venv

激活虚拟环境：

# macOS / Linux source .venv/bin/activate # Windows PowerShell .venv\Scripts\Activate.ps1

3.2 安装依赖

安装以下依赖：

pip install "mcp[cli]" langgraph langchain-mcp-adapters langchain-openai

• mcp[cli]：官方 MCP Python SDK，自带 FastMCP 和命令行工具。
• langgraph：Agent 编排框架。
• langchain-mcp-adapters：MCP 工具转 LangChain Tool 的适配层。
• langchain-openai：OpenAI 兼容接口的 LangChain 封装。

安装完成后确认版本：

python -c "import mcp; print(mcp.__version__)" python -c "import langgraph; print(langgraph.__version__)"

学习资料检索关键词：MCP Python SDK、FastMCP 文档、LangGraph 快速开始、langchain-mcp-adapters 示例仓库。这几个文档会把上文中提到的协议细节和版本差异补齐，强烈建议作为配套资料收藏。

4. 从零构建 MCP Server：Streamable HTTP/SSE

4.1 项目目录结构

mcp-langgraph-demo/ ├── server.py # MCP Server ├── agent.py # LangGraph Agent ├── .venv/ └── requirements.txt

4.2 编写 MCP Server

官方 MCP Python SDK 内置了 FastMCP，可以大幅简化 Server 开发。下面是完整代码：

# server.py from mcp.server.fastmcp import FastMCP mcp = FastMCP("demo-mcp-server") @mcp.tool() def get_weather(city: str) -> str: """查询指定城市的实时天气""" # 这里在实际项目中替换成真实天气 API return f"{city}：多云，24℃，东南风 2 级" @mcp.tool() def add(a: int, b: int) -> int: """两个整数相加，用于验证工具参数与返回值""" return a + b if __name__ == "__main__": # 关键点：指定 streamable-http 传输 mcp.run(transport="streamable-http")

这里有两个细节要说明。第一，工具函数必须带类型注解和 docstring，因为 MCP 协议要把这些信息传给模型做工具选择。第二，transport="streamable-http"会启动一个 HTTP 服务，默认绑定127.0.0.1:8000，默认路由是/mcp。不同版本 SDK 对 host 和 port 的配置方式略有差异，更稳妥的做法是看启动日志里的实际地址。

4.3 启动 MCP Server

python server.py

启动成功后，日志中会显示类似“Uvicorn running on http://127.0.0.1:8000”的信息。这里使用哪个端口就记住哪个端口，后续 Agent 和 curl 测试都要用。

4.4 Streamable HTTP 的启动方式说明

MCP Server 的传输方式不同，启动命令也不一样：

• stdio：比较适合被 LangGraph 作为子进程拉起。
• sse：老版本远程传输，需要独立的 SSE 端点。
• streamable-http：当前推荐，一个端点同时处理请求和 SSE 流。

本教程选择 streamable-http，是因为它更贴近生产环境：Server 可以独立部署，Agent 可以通过 HTTP 远程访问，后续还能在中间加反向代理、鉴权和服务发现。

5. 验证 MCP Server：Inspector 与 curl

5.1 用 MCP Inspector 快速调试

mcp[cli]安装后会自带mcp命令。在开发阶段可以用它打开可视化调试面板：

mcp dev server.py

命令执行后会在本地开一个 Web UI（默认端口 6274），页面里能看到 Server 暴露的所有工具，可以逐个填参数试调用。这个方法适合做第一步功能验证，先确认工具本身没问题，再进入协议验证。

5.2 用 curl 验证 Streamable HTTP 协议

可视化面板好用，但要真正理解 Streamable HTTP，建议用 curl 裸调一遍协议。先测 initialize：

curl -X POST http://127.0.0.1:8000/mcp \ -H "Content-Type: application/json" \ -H "Accept: application/json, text/event-stream" \ -d '{ "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "curl-client", "version": "1.0"} } }'

正常响应会返回协议版本、服务端能力和服务信息。注意响应可能是普通 JSON，也可能是 SSE 分帧格式，这取决于服务端实现和Accept头，两种都是正常情况。

接下来发送notifications/initialized通知，完成握手：

curl -X POST http://127.0.0.1:8000/mcp \ -H "Content-Type: application/json" \ -H "Accept: application/json, text/event-stream" \ -d '{ "jsonrpc": "2.0", "method": "notifications/initialized" }'

然后列出工具清单：

curl -X POST http://127.0.0.1:8000/mcp \ -H "Content-Type: application/json" \ -H "Accept: application/json, text/event-stream" \ -d '{ "jsonrpc": "2.0", "id": 2, "method": "tools/list" }'

最后调用工具：

curl -X POST http://127.0.0.1:8000/mcp \ -H "Content-Type: application/json" \ -H "Accept: application/json, text/event-stream" \ -d '{ "jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": { "name": "add", "arguments": {"a": 3, "b": 5} } }'

这一步是整个项目最关键的单点验证。只要这条命令返回8，说明 MCP Server 的协议链路、JSON-RPC 编码、工具注册和返回值解析全部正常，后面 LangGraph 接入基本不会出大问题。

5.3 失败时的排查顺序

• 请求返回 404：路由不对，确认实际启动日志中的路径。
• 请求返回 405：检查请求方法，Streamable HTTP 走 POST。
• 请求超时：确认 Server 还在运行，端口没被其他进程占用。
• 返回 JSON-RPC error：大概率是参数名或类型与工具定义不一致。

6. LangGraph Agent 集成 MCP 工具

6.1 理解适配层

LangGraph Agent 不能直接消费 MCP 工具，需要经过langchain-mcp-adapters转换。这个适配层的作用是：

• 建立 MCP Client 会话。
• 拉取远程工具列表。
• 把 MCP 工具包装成 LangChain Tool。
• 调用时自动转换参数格式和返回结构。

关键点在于：转换出来的 Tool 依赖活跃的 MCP 会话，所以 Agent 的启动过程要包在会话上下文里。下面是完整代码：

# agent.py import asyncio from langchain_openai import ChatOpenAI from langchain_mcp_adapters.tools import load_mcp_tools from langgraph.graph import StateGraph, START, END, MessagesState from langgraph.prebuilt import ToolNode, tools_condition from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client MCP_SERVER_URL = "http://127.0.0.1:8000/mcp" MODEL_API_KEY = "your-api-key" MODEL_BASE_URL = "https://api.openai.com/v1" # 按实际模型服务商调整 MODEL_NAME = "gpt-4o-mini" async def run(): llm = ChatOpenAI( model=MODEL_NAME, api_key=MODEL_API_KEY, base_url=MODEL_BASE_URL, temperature=0, ) async with streamablehttp_client(MCP_SERVER_URL) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools = await load_mcp_tools(session) llm_with_tools = llm.bind_tools(tools) builder = StateGraph(MessagesState) builder.add_node("llm", llm_with_tools) builder.add_node("tools", ToolNode(tools)) builder.add_edge(START, "llm") builder.add_conditional_edges("llm", tools_condition) builder.add_edge("tools", "llm") graph = builder.compile() result = await graph.ainvoke({ "messages": [{"role": "user", "content": "计算 3 + 5 等于多少"}] }) print(result["messages"][-1].content) if __name__ == "__main__": asyncio.run(run())

6.2 核心节点说明

• MessagesState：LangGraph 内置的对话状态结构，保存消息列表。
• llm节点：把工具 bound 到模型上，模型输出可能是文本，也可能是工具调用指令。
• tools_condition：内置条件边，判断上一步是否产生了工具调用，有则跳转到 tools，无则走到 END。
• tools节点：执行工具，把工具返回结果作为后续消息继续交给 LLM 节点。
• builder.add_edge("tools", "llm")：工具结果回流给模型做最终总结。

这里要注意，实际项目中不要把所有逻辑都塞进async with块里。更工程化的做法是编写一个 MCP Session 管理器，在 Agent 执行期间动态维护会话生命周期。本教程为了演示链路，先保持这种最简写法。

7. 端到端测试与效果验证

7.1 测试一：基础工具调用链路

执行：

python agent.py

输入问题“计算 3 + 5 等于多少”，预期执行路径是：

1. LLM 节点收到问题，输出 tool_call，工具名是add。
2. tools 节点调用 MCP Server 的add工具。
3. 结果8回流到 LLM 节点。
4. LLM 组织最终回复“3 + 5 = 8”。

判断成功标准：最终输出包含计算结果，且 MCP Server 终端有新的请求日志。

7.2 测试二：多工具决策

把问题换成“查询北京的天气，然后计算 10 加 20”。这条测试的意图是验证模型能连续调用多个 MCP 工具。注意，如果模型服务商不支持多工具并行，LangGraph 会走“先调一个，回流后再调下一个”的顺序路径，这没问题，关键看中间是否有tool_call事件。

7.3 测试三：LangGraph 流式输出

Agent 的流式输出通过astream_events拿到。改造agent.py：

async for event in graph.astream_events( {"messages": [{"role": "user", "content": "计算 3 + 5 等于多少"}]}, version="v2", ): kind = event["event"] if kind == "on_chat_model_stream": chunk = event["data"]["chunk"] delta = getattr(chunk, "content", "") if delta: print(delta, end="", flush=True) if kind == "on_tool_start": print("\n[工具调用]", event["name"], event["input"], flush=True) if kind == "on_tool_end": print("\n[工具返回]", event["output"], flush=True)

输出会同时看到模型增量文本和工具调用日志。这里的流式输出指的是 LangGraph/模型层面的 token 流，它是实现类 ChatGPT 打字机效果的基础。而 MCP Server 侧还有另一层 SSE 流，用来推送工具执行进度，两者不要混为一谈。

7.4 测试四：批量任务

批量调用可以写一个问题列表，循环执行：

questions = [ "计算 3 + 5", "计算 10 * 20", "查询北京的天气", ] for q in questions: result = await graph.ainvoke({ "messages": [{"role": "user", "content": q}] }) print(q, "=>", result["messages"][-1].content)

批量时注意两点：一是 MCP 会话的并发能力有限，不能无限制同时发起大量 tools/call；二是生产环境一定要加信号量限流和失败重试。稳妥做法是先跑串行，确认稳定性后再考虑并发。

8. 资源占用与性能观察

8.1 这个项目消耗什么资源

LangGraph 本身是编排层，几乎不消耗显存。真正的资源消耗集中在两处：

• 大模型 API：网络请求耗时和 token 费用是主要成本。
• MCP Server：进程常驻内存，单个简单工具服务占用内存通常在几十 MB 级别，具体以本机实测为准。

如果你把模型换成本地 Ollama 或 vLLM 部署的开源模型，才需要关注显卡显存。从这套架构来看，项目对硬件要求很低，普通开发机即可完成全部调试。

8.2 性能观察方法

• 用htop或任务管理器观察 MCP Server 进程的 CPU 和内存。
• 用lsof -i :8000（macOS/Linux）或netstat -ano | findstr 8000（Windows）确认端口占用。
• 看 MCP Server 日志，统计单次 tools/call 的耗时。
• 看网络连接数，SSE 长连接会保持 TCP 连接，如果有大量并发任务，连接数会上升。

8.3 影响性能的主要因素

• 模型推理速度：决定“LLM 节点判断要不要调工具”的延迟。
• 工具内部耗时：决定 ToolNode 阶段延迟。
• SSE 推送频率：工具执行过程中如果频繁推进度事件，会增加网络开销。
• 会话管理：频繁创建和销毁 MCP 会话会带来握手开销，长连接复用更优。

如果工具调用延迟明显，优先检查 MCP Server 内部是否有同步阻塞调用。FastMCP 默认基于 asyncio，工具函数里不要写同步的长时间轮询逻辑，可以用异步工具或把耗时操作交给任务队列。

9. 常见问题与排查方法

9.1 最容易踩的三个坑

第一个坑是“没 initialize 就加载工具”。MCP 会话必须先握手，否则tools/list拿不到数据，LangGraph 侧工具列表为空，模型会直接回答“我无法计算”。

第二个坑是“Agent 脚本运行时 MCP Server 没启动”。MCP 会话建立失败会直接抛连接错误。建议先单独跑通第 5 节的 curl 验证，再启动 Agent。

第三个坑是“流式输出事件解析出错”。astream_events的事件结构会随 LangGraph 版本变化，代码里要加getattr兜底，不要直接索引字段。

10. 最佳实践与使用边界

10.1 工程化建议

• 第一次跑通后再加功能，先保留“curl tools/call 正常返回”的最小可运行基线。
• MCP Server、Agent、测试脚本分目录管理，模型 Key 用环境变量或配置文件注入，不要硬编码提交到仓库。
• MCP Server 的工具输出加结构化日志，方便排查。
• 批量任务必须设计日志、超时和重试，不要一把梭并发。
• 接口服务在开发阶段监听127.0.0.1，生产环境放在反向代理后面，统一做 TLS 和鉴权。

10.2 安全与合规边界

• Streamable HTTP 服务如果不做鉴权，任何能访问该端口的人都能调用工具。生产环境必须加 Bearer Token、OAuth 或网关鉴权。
• MCP 工具要遵循最小权限原则，只暴露当前业务确实需要的操作。
• 涉及高风险的写操作、数据删除、支付类工具，必须在 LangGraph 流程中加入人工审批节点。
• 调用云端模型 API 时，注意不要向不信任的第三方服务发送敏感数据。处理用户数据和业务数据前，明确授权范围。
• 不要用这套工具链做未授权的数据抓取、系统渗透或绕过平台限制的操作。
• AI 生成的内容在使用和发布前，要按业务要求做人工复核，避免错误信息和版权风险。

这些边界不是可选项。MCP + LangGraph 最大的风险点在于“工具调用自动执行”，一个权限控制不严的工具链，可能比普通代码接口造成更严重的后果。

11. 配套学习资料与下一步方向

11.1 配套资料检索清单

本项目的配套学习资料建议按下面几个方向整理：

• MCP 官方协议规范：重点看Streamable HTTP章节和2025-06-18协议版本。
• MCP Python SDK 示例：官方仓库有streamable-http的 server/client 示例。
• FastMCP 文档：工具定义、鉴权配置、SSE 行为。
• langchain-mcp-adapters 示例：看load_mcp_tools的标准用法。
• LangGraph 文档：StateGraph、ToolNode、astream_events。

检索关键词：MCP Streamable HTTP example、fastmcp streamable-http、langchain mcp adapters load_mcp_tools、langgraph tool calling streaming。

11.2 后续扩展方向

• 增加持久化记忆：给 LangGraph 加Checkpointer，让 Agent 能跨会话记住上下文。
• 增加鉴权：MCP Server 请求头校验 Bearer Token。
• 服务化部署：把 MCP Server 放到 Docker，Agent 独立部署，通过 HTTP 远程访问工具。
• 多 Agent 协作：用 LangGraph 编排多个专用 Agent，分别对接不同 MCP Server。
• 接入真实业务工具：把内部 API、数据库查询封装成 MCP 工具，规范描述输入输出，让模型安全调用。

最后给一个实操建议：拿到这套代码后，最优先验证的不是 LangGraph，而是第 5 节那条curl tools/call链路。MCP Server 的协议链路通了，Agent 侧只要确认bind_tools生效、工具列表非空，剩下的就是 LangGraph 标准编排流程。先把这条最小闭环跑起来，再逐步加鉴权、批量、并发和持久化，项目落地会顺畅很多。