企业官网建设流程全解析

这次我们直接拆一个组合型项目：MCP + LangGraph，从零构建一个基于 Streamable HTTP 和 SSE 传输系统的工具调用链路。

标题里的关键词很明确：MCP（Model Context Protocol）做工具标准化，LangGraph 做 Agent 工作流编排，Streamable HTTP 和 SSE 做服务端流式传输。这三件事单独看都不复杂，但把它们完整串起来，并且能在本地调试、远程调用、批量验证，才是真正的项目落地能力。

这篇文章会从环境准备、服务端工具定义、Streamable HTTP / SSE 传输层实现、LangGraph 状态图集成，到 curl 验证、API 调用、批处理建议、性能观察和排错清单，按“能动手”的方式完整过一遍。适合三类读者：想把现有函数/API 接入 MCP 的开发者、正在为 LangGraph Agent 做工具层落地的工程师、以及对 SSE 流式服务端传输出于学习目的的初学者。

1. 核心能力速览

在开始写代码之前，先用一张表把项目的关键信息固定下来。MCP + LangGraph 的组合不像单模型部署那样有“显存占用”“显卡驱动”这类硬性参数，它的核心指标是协议支持、传输模式、服务化能力和可扩展性。

需要说明的是：这个项目不涉及 GPU 推理，也不依赖本地大模型。如果要在 LangGraph 里调用大模型做意图判断，通常接云端模型 API；如果使用本地模型，则额外增加模型服务部署，但这不影响传输层的实现方式。

2. 适用场景与使用边界

MCP + LangGraph 的典型价值，是把“大模型调用工具的流程”标准化。有了 MCP Server，工具就是协议化的接口；有了 LangGraph，Agent 的思考、调用、观察、再决策就能用状态图表达；有了 Streamable HTTP 和 SSE，这些交互就能跨进程、跨网络。

这个组合适合以下场景：

• 工具数量多、类型杂，需要一个统一协议层，避免每个 Agent 框架单独适配一套工具接口。
• Agent 流程有分支和循环，比如“先查库，再判断是否需要查文档，最后生成结论”。
• 远程服务需要返回增量信息，希望客户端在长耗时任务中能一直看到进度。
• 开发者希望把 MCP Server 作为一个独立服务部署，供多个 Agent 共用。

它不适合什么场景？

• 纯 CRUD 的简单 HTTP 接口，没必要引入 MCP + LangGraph，反而增加复杂度。
• 需要高频双向实时通信的聊天室类业务，SSE 是服务端单向推送，双向通信更适合 WebSocket。
• 对响应延迟要求极低的计算型服务，MCP 的 JSON-RPC 包装和传输层会有一定开销。

使用边界方面需要特别提醒：MCP Server 一旦服务化，暴露在网络上，工具就相当于可被远程调用的 API。不要默认开放所有本机能力。生产环境必须做鉴权、限流、白名单和操作审计。演示环境应使用脱敏数据，不要拿真实账号、真实个人信息、未授权版权素材做测试案例。

3. 环境准备与前置条件

这个项目的核心依赖是 Python 生态，环境准备比较轻。建议用独立虚拟环境，避免和系统 Python 或其它项目冲突。

3.1 基础要求

• 操作系统：Windows / macOS / Linux 均可，命令行和 Python 环境正常即可。
• Python：建议 Python 3.10 或 3.11，实际以 mcp、langgraph、fastapi 等包当前兼容范围为准。
• 包管理：建议使用uv或venv + pip。
• 网络：如果 LangGraph 中需要调用云端大模型 API，需要能访问对应服务；如果只做工具调用本地验证，可以不依赖外部模型。
• 端口：规划一个未被占用的端口，本文示例统一使用 8000。

3.2 创建虚拟环境并安装依赖

python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate pip install --upgrade pip pip install mcp langgraph fastapi uvicorn sse-starlette requests

依赖说明：

• mcp：提供 MCP Server / Client 基础能力，包括工具定义和传输。
• langgraph：提供状态图、节点、条件边等 Agent 编排能力。
• fastapi：用于承载 HTTP 服务和 SSE 传输端点。
• uvicorn：本地启动 ASGI 服务。
• sse-starlette：在 FastAPI 中快速实现 SSE 响应。
• requests：用于最后的 API 调用验证。

版本问题处理：如果安装时出现依赖冲突，建议优先参考mcp和langgraph官方仓库 README 中标注的兼容版本组合，不要一次性强行升级全部包。

4. 项目结构与实现流程

“从零构建”不是把所有代码写在一个文件里。按下面的目录结构组织，服务端、传输层、Agent 编排可以分别维护，后续替换实现方式时不用推翻重来。

mcp-langgraph-demo/ server/ __init__.py tools.py # MCP 工具定义 transport.py # Streamable HTTP / SSE 传输层 agent/ __init__.py state.py # LangGraph 状态定义 graph.py # LangGraph 工作流 app.py # FastAPI 入口，挂载传输层和调试路由 requirements.txt client_test.py # API 调用和批量验证脚本

实现流程分为四步：

1. 在server/tools.py中定义 MCP 工具，也就是 Agent 可以被模型调用的外部能力。
2. 在server/transport.py中实现传输层，选择 Streamable HTTP 并兼容 SSE 流式输出。
3. 在agent/graph.py中构建 LangGraph 工作流，把 Agent 的多次工具调用组织成状态图。
4. 启动app.py，用 curl 和 Python 脚本验证链路。

这个顺序也是实际项目中推荐的做法：先让“工具能被调用”，再做“ Agent 能编排调用”。

5. 服务端开发：MCP Server 与工具定义

5.1 最小化 MCP 工具

下面的代码基于mcp官方的 FastMCP 风格编写，这是当前最直接、最常用的 MCP 工具定义方式。如果后续mcp版本有 API 调整，以实际安装版本的文档为准。

# server/tools.py from mcp.server.fastmcp import FastMCP from typing import Optional # 实例化 MCP Server，name 会出现在工具名中 mcp = FastMCP("book-tools") @mcp.tool() def lookup_book(keyword: str, limit: Optional[int] = 5) -> str: """根据关键词查询书籍信息。""" # 这一层通常是真实业务 API 的适配器 mock_db = [ {"title": "MCP in Action", "author": "demo", "tags": ["MCP", "AI"]}, {"title": "LangGraph Guide", "author": "demo", "tags": ["LangGraph", "Agent"]}, ] results = [book for book in mock_db if keyword.lower() in book["title"].lower()] return str(results[:limit]) @mcp.tool() def summarise_text(text: str, max_len: int = 200) -> str: """将一段文本裁剪为摘要。此处保留接口形态，后续可替换为真实 LLM 调用。""" if len(text) <= max_len: return text return text[:max_len] + "..."

这里的工具并不只是演示代码。在实际项目中，lookup_book的函数体内会替换成数据库查询、RAG 检索、HTTP API 调用等真实逻辑。MCP 的职责是把这些业务能力包装成模型可发现的标准化工具。

5.2 工具定义要点

一个规范的 MCP 工具定义需要包含：

• 清晰的函数名，作为tools/call中的name。
• 参数类型标注，MCP 会根据类型自动生成工具入参 schema。
• 文档字符串，描述该工具的用途，模型会依据这段描述判断何时调用。
• 返回结构化文本，方便模型二次处理。

6. 传输层实现：Streamable HTTP 与 SSE

6.1 为什么需要 Streamable HTTP

MCP 早期最常用的传输方式是stdio，适合本地进程内调用。一旦 Agent 和工具不在同一个进程，就必须启用 HTTP 传输。最初支持的是“HTTP + SSE”模式，服务端提供一个 SSE 端点用于推流，再配合一个独立的收件端点接收消息。

Streamable HTTP 则是更进一层的设计：把服务端点收窄为一个统一的 HTTP 端点，客户端在同一连接上发送初始请求，服务端按需切换为 SSE 流式响应。长耗时任务可以通过多条 SSE 消息持续返回结果，避免“必须一次响应全部内容”的阻塞。

6.2 启用 Streamable HTTP 传输

如果使用 FastMCP，直接指定传输方式即可：

# server/transport.py from server.tools import mcp def run_streamable_server(host: str = "127.0.0.1", port: int = 8000): # 示例：FastMCP 在部分版本中支持 streamable-http 传输； # 若当前版本不支持，请使用 mcp.run(transport="http") 并检查服务端路由。 mcp.run(transport="streamable-http")

上面代码是简化写法。实际以安装后的mcp版本为准，需要查询当前 FastMCP 支持的传输参数。更通用、更可控的做法是：自己用 FastAPI 构建传输层，把 MCP 协议消息显式映射到 HTTP 和 SSE 端点上。下面给出一个可以直接理解机制的演示结构。

6.3 用 FastAPI + SSE 实现传输端点

如果希望完整掌控路由、鉴权、日志，可以基于 FastAPI 和 sse-starlette 自己实现简化的 SSE 端点。注意：这是教学用的简化结构，真实场景中你需要处理 MCP 协议的初始化握手和 JSON-RPC 消息格式。

# server/transport.py import asyncio from fastapi import FastAPI, Request from sse_starlette.sse import EventSourceResponse app = FastAPI() @app.get("/health") async def health(): return {"status": "ok"} @app.post("/rpc") async def rpc(request: Request): payload = await request.json() # 这里应解析 JSON-RPC 2.0 请求，调用对应 MCP 工具 # 演示：直接返回一个 JSON-RPC 响应 return { "jsonrpc": "2.0", "id": payload.get("id"), "result": "tool result placeholder" } @app.get("/sse") async def sse_endpoint(request: Request): async def event_generator(): for i in range(5): if await request.is_disconnected(): break yield { "event": "message", "data": f'{{"jsonrpc":"2.0","id":1,"result":"第 {i} 条流式数据"}}' } await asyncio.sleep(1) return EventSourceResponse(event_generator())

启动方式：

uvicorn server.transport:app --host 127.0.0.1 --port 8000 --reload

启动后先访问http://127.0.0.1:8000/health，确认服务存活。然后分别测试/rpc和/sse，验证两条路径是否正常。这也是从零构建时最容易理解和排查的部分：先把 HTTP 端点跑通，再接入 MCP 协议解析和 LangGraph。

6.4 Streamable HTTP 与 SSE 的关系

两者不是互斥，而是配合关系。Streamable HTTP 定义了“客户端如何发起请求、服务端如何回包”的整体传输规则。SSE 是底层流式数据推送机制，用于实现“服务端到客户端”的长连接增量传输。一个完整的 Streamable HTTP + SSE 系统，通常包含：

• 一个 HTTP 入口地址，接收 MCP 客户端发来的 JSON-RPC 消息。
• 服务端处理消息后，如果需要持续输出，把响应以 SSE 事件流方式返回。
• 客户端按事件流逐步读取结果，而不是等待单个 HTTP 响应体全部结束。

相比一次性 HTTP 响应，这种模式特别适合工具调用耗时较长的场景，比如“生成报告”“批量检索多个数据源”“逐步执行多步工具链”。

7. LangGraph 集成：把 MCP 工具接入 Agent 工作流

传输层解决的是“MCP Server 怎么被远程调用”。LangGraph 解决的是“Agent 怎么在多次思考、多次工具调用之间维持状态和流程”。

7.1 LangGraph 核心概念

LangGraph 的构建通常围绕三类对象：

• 状态（State）：承载多轮对话消息、中间结果、工具返回。
• 节点（Node）：一个可执行函数，接收状态，处理后返回新状态。
• 边（Edge）：节点之间的连接，可以是普通边，也可以是条件边。

下面定义一个最简状态和状态图，用于说明 Agent 节点如何调用 MCP 工具。

# agent/state.py from typing import TypedDict, Annotated, List from langgraph.graph import add_messages class AgentState(TypedDict): messages: Annotated[List[dict], add_messages] tool_results: List[dict]

# agent/graph.py from langgraph.graph import StateGraph, END from agent.state import AgentState def call_tool_node(state: AgentState): """模拟通过 MCP 协议调用远程工具。""" result = { "lookup_book": "MCP in Action", "status": "success" } return {"tool_results": [result]} def agent_finish_node(state: AgentState): """根据工具结果生成最终回答。""" messages = state.get("messages", []) last_user_msg = messages[-1] if messages else {"content": "无输入"} combined = f"用户问题：{last_user_msg.get('content', '')}；工具结果：{state.get('tool_results', [])}" return {"messages": [{"role": "assistant", "content": combined}]} graph = StateGraph(AgentState) graph.add_node("call_tool", call_tool_node) graph.add_node("agent_finish", agent_finish_node) graph.add_edge("call_tool", "agent_finish") graph.add_edge("agent_finish", END) agent = graph.compile()

这个示例很短，但结构就是 LangGraph 应用的最小闭环。实际项目里，call_tool_node内部要完成三件事：

1. 解析当前状态，找出需要调用的 MCP 工具名称和参数。
2. 通过 MCP 客户端或 HTTP 请求，向 MCP Server 发送tools/call。
3. 把返回结果写入状态，交给下一个节点继续判断。

7.2 在 LangGraph 中调用 MCP 工具

MCP 协议基于 JSON-RPC 2.0。在 LangGraph 节点里，如果不用官方客户端 SDK，可以直接用requests按协议格式调用：

# agent/mcp_client.py import requests MCP_SERVER_URL = "http://127.0.0.1:8000/rpc" def call_mcp_tool(name: str, arguments: dict, request_id: int = 1): payload = { "jsonrpc": "2.0", "method": "tools/call", "params": { "name": name, "arguments": arguments }, "id": request_id } response = requests.post(MCP_SERVER_URL, json=payload, timeout=120) response.raise_for_status() return response.json()

调用时注意：真实 MCP Server 的请求地址、初始化流程、请求头都可能和上面这个简化端点不一样。这个示例的目的是说明“ LangGraph 节点内调用远程 MCP 工具”这个模式。生产环境建议优先使用官方 mcp 客户端连接器，避免手动维护协议细节。

7.3 Agent 循环与条件跳转

真实 Agent 通常不是“调用一次工具就结束”，而是：模型判断需要工具 -> 调用工具 -> 观察结果 -> 再判断是否还需要调用 -> 循环直到可以回答。LangGraph 里用条件边实现循环：

from langgraph.graph import END, StateGraph def should_continue(state: AgentState): # 判断依据：当工具结果满足条件时结束，否则继续调用 if state.get("tool_results"): return "agent_finish" return "call_tool" graph.add_conditional_edges( "call_tool", should_continue, {"continue": "call_tool", "agent_finish": "agent_finish"} )

从零构建时不要一开始就做复杂循环，先让“调用一次工具 -> 结束”的链路跑通，再逐步叠加判断逻辑。

8. 接口 API 调用与运行验证

8.1 启动服务

先启动 MCP / 传输服务：

uvicorn app:app --host 127.0.0.1 --port 8000

再单独启动 LangGraph Agent（如果是脚本式验证，直接执行 Python 脚本即可）。实际项目中，LangGraph 可以作为一个函数库被调度服务调用，也可以封装成自己的 API 服务。

验证链路时，建议按下面的顺序操作：

1. 访问/health，确认传输层在线。
2. 调用 MCP 初始化接口或工具列表接口，确认工具可被发现。
3. 调用tools/call，确认单个工具执行成功。
4. 接入 LangGraph 节点，执行一个完整 Agent 流程。

8.2 curl 验证

工具列表请求示例：

curl -X POST http://127.0.0.1:8000/rpc \ -H "Content-Type: application/json" \ -d '{"jsonrpc":"2.0","method":"tools/list","params":{},"id":1}'

预期响应中包含已注册的lookup_book、summarise_text工具描述。如果返回空列表，说明工具注册没有成功，优先检查FastMCP实例是否被正确实例化。

调用工具示例：

curl -X POST http://127.0.0.1:8000/rpc \ -H "Content-Type: application/json" \ -d '{"jsonrpc":"2.0","method":"tools/call","params":{"name":"lookup_book","arguments":{"keyword":"MCP"}},"id":2}'

SSE 流式验证：

curl -N http://127.0.0.1:8000/sse

8.3 Python 批量调用

批量验证时，维护一个任务列表，依次调用 MCP 工具并收集结果。建议在函数中增加超时和重试，避免单个任务拖垮整个批处理。

# client_test.py import time import requests MCP_SERVER_URL = "http://127.0.0.1:8000/rpc" RETRY_TIMES = 3 def call_tool_once(name, arguments, request_id): payload = { "jsonrpc": "2.0", "method": "tools/call", "params": {"name": name, "arguments": arguments}, "id": request_id, } response = requests.post(MCP_SERVER_URL, json=payload, timeout=60) response.raise_for_status() return response.json() def batch_call(tasks): results = [] for idx, task in enumerate(tasks, start=10): for attempt in range(RETRY_TIMES): try: result = call_tool_once(task["name"], task["arguments"], idx) results.append({"task": task, "result": result}) break except requests.RequestException as exc: print(f"task {idx} 第 {attempt + 1} 次失败：{exc}") time.sleep(2) else: results.append({"task": task, "error": "超过重试次数"}) return results if __name__ == "__main__": tasks = [ {"name": "lookup_book", "arguments": {"keyword": "MCP"}}, {"name": "summarise_text", "arguments": {"text": "A" * 300, "max_len": 50}}, ] for r in batch_call(tasks): print(r)

批量任务是否适合放进 LangGraph 中，需要看业务类型。如果一批任务相互独立，可以用普通循环或asyncio.gather并发处理；如果任务之间有先后依赖，或者需要根据前一步结果调整下一步，才需要放进 LangGraph 状态图里。

9. 资源占用与性能观察

这个项目没有固定的 GPU 显存指标，性能观察重点应该在四个方面：CPU、内存、连接数、外部 API 耗时。

• 服务进程 CPU：启动后先观察空闲状态 CPU 是否接近 0。如果持续过高，优先检查是否有轮询任务或未关闭的连接。
• 内存：MCP Server 和 LangGraph 状态图都是常驻进程，内存增长异常说明可能存在消息列表无限追加问题。LangGraph 的 messages 字段建议定期裁剪或摘要。
• 连接数：SSE 长连接会持续占用文件描述符，并发用户多时需要关注连接清理。客户端断开后，SSE 生成器需要及时退出。
• 外部 API 耗时：Agent 调用云端模型或第三方业务 API 的耗时往往是整个链路的主要瓶颈。建议在关键调用点埋日志，记录 request_id、开始时间、结束时间。

如果后续接入本地大模型做 Agent 思考，再单独观察模型服务的显存占用。本章节明确不涉及：本文项目本身不加载模型权重，运行资源消耗以框架和业务函数为准。

优化思路：

• 使用流式响应，不把大结果一次性拼在内存里。
• 合理设置超时，避免一个慢工具阻塞整条 SSE 流。
• 批量任务使用有界队列，控制并发上限。
• 日志里记录每个工具的输入输出大小，方便定位异常。

10. 常见问题与排查方法

下面是这个项目中最高频的问题。排查时先看日志，再看请求消息格式，最后检查连接状态。

11. 最佳实践与使用建议

从零构建这个项目时，先不要想着把模型接入、鉴权、多 Agent、数据库全做完。建议按“最小闭环”顺序推进：

第一步，把 MCP Server 跑起来，注册两个工具，用 curl 完成tools/list和tools/call。

第二步，把 Streamable HTTP / SSE 传输层跑通，保证长耗时任务能流式返回。

第三步，构建最简 LangGraph 状态图，在节点中调用 MCP 工具。

第四步，加批量任务入口，加日志，加超时重试。

第五步，再考虑鉴权、限流、模型决策、多工具并行调用。

工程化建议：

• 工具函数保持单一职责，一个工具对应一个业务能力，方便模型理解和路由。
• 所有 MCP 工具调用都加上 request_id，日志里用 request_id 串起整条链路。
• 不要在生产环境直接暴露 MCP 服务，必须在网关层加认证、限流和操作审计。
• 工具返回内容长度要控制。模型上下文有限，一次性返回过多内容会导致 Agent 决策质量下降。
• 批量任务建议用任务队列管理，而不是直接开启无限线程；先限制并发数。
• 涉及人脸、声音、隐私数据、版权素材的工具和测试数据，一律确认授权后再使用，测试环境优先使用模拟数据。

12. 总结与下一步

这个项目最值得尝试的点，是把 MCP 的工具标准化、LangGraph 的状态编排、Streamable HTTP / SSE 的流式传输组合成一条完整链路。单独看每一个技术点都不复杂，但组合起来就非常贴近真实 Agent 工程的调用形态。

你最先应该验证的，不是复杂的 Agent 多轮循环，而是：MCP Server 能不能被远程请求调通工具、SSE 端点能不能持续返回增量结果。这两点通了，后面往上加 LangGraph 状态图、加批量调度、加鉴权限流，都是增量工程。

最容易踩的坑有三个：一是传输模式不匹配，客户端用普通 HTTP 请求去接 SSE 流式端点；二是 JSON-RPC 消息格式不标准，id字段缺失或params结构不对；三是 LangGraph 条件边返回值和实际注册的边名不一致，导致循环进入死循环。

下一步扩展方向建议从三个点入手：接入真实业务 API，把工具从 mock 数据换成实际查询；在 LangGraph 中加入模型判断节点，让 Agent 能自主决定是否继续调用工具；把 MCP Server 包装成带鉴权限流的可独立部署服务，供多个 Agent 共用。当你完成这些扩展之后，MCP + LangGraph + Streamable HTTP / SSE 就不再是教学项目，而是能支撑真实业务落地的 Agent 基础设施。