使用 mcp-agent 连接 Streamable HTTP 服务端从无状态服务器部署到工具调用实战【免费下载链接】mcp-agentBuild effective agents using Model Context Protocol and simple workflow patterns项目地址: https://gitcode.com/GitHub_Trending/mc/mcp-agent导读本文以开源仓库 mcp-agent 中的examples/mcp/mcp_streamable_http示例为主线完整讲解如何让 mcp-agent 应用通过 Streamable HTTP 传输层连接远程 MCP 服务器完成连接握手、工具列表获取与会话 ID 观察等操作。阅读完本文你将掌握 Streamable HTTP 与 SSE、stdio 等传输方式的差异能够独立编写一个无状态 Streamable HTTP MCP 服务端并在mcp_agent.config.yaml中正确配置streamable_http传输类型最终在 mcp-agent 中调用其工具。一、示例概览Streamable HTTP 传输在 mcp-agent 中的定位examples/mcp/mcp_streamable_http/目录演示了 mcp-agent 与一个Streamable HTTP 服务端的完整交互流程。示例中的服务端是一个无状态stateless服务器整个应用预期完成三件事连接connect、初始化initialize并列出其工具list tools。该目录包含以下关键文件文件作用stateless_server.py基于mcp-pythonSDK 编写的一个无状态 Streamable HTTP 演示服务端main.pymcp-agent 客户端应用连接服务器并调用list_toolsmcp_agent.config.yaml声明streamable_http传输类型的 MCP 服务器配置mcp_agent.secrets.yaml.exampleAPI Key 等敏感信息模板requirements.txt示例专属依赖click、openai及本地 mcp-agent 项目本体1.1 Streamable HTTP 是什么Streamable HTTP 是 MCP 规范中较新的传输层方案它把请求/响应与服务器主动推送通知、日志统一承载在 HTTP 之上。与 SSE 相比Streamable HTTP 支持客户端与服务器之间双向、流式的消息交互是 MCP 生态中面向生产环境的推荐传输方式之一。在 mcp-agent 的配置体系中它被显式列为一种独立传输类型可从 配置模型定义 中看到完整的取值集合transport: Literal[stdio, sse, streamable_http, websocket] stdio也就是说mcp-agent 的每个 MCP 服务器连接都可以在这四种传输方式中选择streamable_http与基于进程的stdio、基于单向推送的sse并列。二、服务端实现一个真正的无状态 Streamable HTTP 服务器2.1 无状态的含义所谓无状态stateless指的是服务器不在自身持久化会话状态每次请求独立处理。这在 stateless_server.py 中通过显式参数体现session_manager StreamableHTTPSessionManager( appapp, event_storeNone, json_responsejson_response, statelessTrue, )event_storeNone不为事件提供持久化存储statelessTrue启用真正的无状态模式json_response由命令行参数控制用于决定响应格式是 JSON 还是 SSE 流。2.2 工具定义与路由挂载服务器注册了一个名为start-notification-stream的工具接受interval通知间隔秒数、count通知条数、caller调用方标识三个必填参数。当被调用时它会在会话内按指定间隔发送多条日志通知最后返回一条文本结果见 stateless_server.py。服务器通过 Starlette 挂载 ASGI 路由将/mcp端点交给 Streamable HTTP 会话管理器处理并默认监听0.0.0.0:3156端口可通过--port修改starlette_app Starlette( debugTrue, routes[ Mount(/mcp, apphandle_streamable_http), ], lifespanlifespan, ) uvicorn.run(starlette_app, host0.0.0.0, portport)因此本示例服务器的完整端点地址是http://0.0.0.0:3156/mcp。命令行的完整选项如下选项默认值说明--port3156HTTP 监听端口--log-levelINFO日志级别DEBUG/INFO/WARNING/ERROR/CRITICAL--json-responseFalseflag开启后以 JSON 响应替代 SSE 流三、客户端配置在 mcp_agent.config.yaml 中声明 Streamable HTTP 服务器mcp-agent 客户端通过 mcp_agent.config.yaml 描述要连接的 MCP 服务器mcp: servers: stateless_http: description: A streamable HTTP server that is stateless. transport: streamable_http url: http://0.0.0.0:3156/mcp headers: my-header: some_value三个关键字段transport: streamable_http声明传输方式服务端代码中的streamable_http、streamable-http、http三种写法均可被识别见 mcp_server_registry.pyurl服务器端点必须与上一节/mcp挂载路径严格一致否则连接会失败headers可选的自定义 HTTP 请求头可用于携带自定义认证信息或其他元数据。3.1 与 Streamable HTTP 相关的完整配置项除上述字段外mcp-agent 还为远程 HTTP 类传输提供了更细粒度的连接控制参数均定义在 配置模型 中配置项默认值说明url无SSE、Streamable HTTP 或 WebSocket 服务器地址headers无附加到 SSE / Streamable HTTP 请求的 HTTP 头http_timeout_seconds无HTTP 请求超时秒注意与下一条区分read_timeout_seconds无客户端在断连前等待新事件的超时秒terminate_on_closeTrue连接关闭时是否终止 Streamable HTTP 会话其中http_timeout_seconds与read_timeout_seconds的语义差异值得注意前者是单个 HTTP 请求的生命周期上限后者是流式连接上等待下一条事件的最大时间。生产环境中若服务器采用长连接推送模式通常需要适当调大read_timeout_seconds避免客户端因短暂空闲被误判为断连。3.2 客户端底层是如何建立 Streamable HTTP 连接的从源码看当config.transport为streamable_http时连接管理器会走到 mcp_server_registry.py 的分支逻辑其核心流程为校验url未配置 URL 时直接抛出ValueError提示 Streamable HTTP 传输必须提供 URL注入会话 ID若携带历史session_id则把MCP_SESSION_ID头合并进请求头用于无状态模式下向服务器重建/延续会话上下文构造连接参数将url、headers、terminate_on_close、超时配置等打包调用mcp.client.streamable_http.streamablehttp_client建立底层连接注册会话 ID 回调通过session.set_session_id_callback(session_id_callback)让客户端会话能够持续跟踪服务器返回的会话 ID封装会话以MCPAgentClientSession形式 yield 给上层 Agent 使用。此外该分支还预留了 OAuth 认证挂载点当服务器配置了auth.oauth.enabled时会自动接入OAuthHttpxAuth处理器需要上下文提供token_manager。这意味着在 mcp-agent 中Streamable HTTP 服务器可以走 OAuth 授权流程而不仅仅是裸 HTTP。四、客户端应用连接、握手与工具列举4.1 应用入口main.py 展示了 mcp-agent 的最小化应用骨架from mcp_agent.app import MCPApp from mcp_agent.agents.agent import Agent app MCPApp(namemcp_streamable_http) async def example_usage(): async with app.run() as agent_app: logger agent_app.logger context agent_app.context logger.info(Current config:, datacontext.config.model_dump()) agent Agent( namestreamable-http-agent, instructionYou are an agent whose job is to interact with various MCP servers over streamable HTTP transport., server_names[stateless_http], ) async with agent: result await agent.list_tools() logger.info(Tools available:, dataresult.model_dump()) session_id (await agent.get_server_session(stateless_http)).session_id logger.info(Session ID:, datasession_id)关键步骤拆解MCPApp(name...)创建应用实例设置既可编程指定也可从mcp_agent.config.yaml/mcp_agent.secrets.yaml加载Agent(..., server_names[stateless_http])把 Agent 与配置中声明的stateless_http服务器绑定agent.list_tools()在async with agent:上下文内触发连接与初始化握手随后列出服务器全部工具结果被完整打印出来agent.get_server_session(stateless_http)查询该服务器的会话 ID注释明确预期——无状态服务器返回None。4.2 为什么无状态服务器会返回None的会话 ID从 get_server_session 的实现 可以看到该方法最终从MCPAgentClientSession.get_session_id()取值。对于无状态模式statelessTrue、event_storeNone的服务器握手响应中不产生有意义的持久化会话标识因此session_id为None是符合预期的行为——这恰好成为验证服务器确实运行在无状态模式的一个直观观测点。4.3 预期运行效果应用运行后日志应依次出现当前配置的完整 dump来自context.config.model_dump()streamable-http-agent: Connected to servers, calling list_tools...连接并初始化成功Tools available:及工具列表其中应包含start-notification-stream及其输入 SchemaSession ID:为None。最后程序会打印Total run time: X.XXs统计整体耗时。五、从零运行本示例5.1 环境准备克隆仓库并进入示例目录git clone https://github.com/lastmile-ai/mcp-agent.git cd mcp-agent/examples/mcp/mcp_streamable_http/安装uv若尚未安装pip install uv同步 mcp-agent 项目依赖uv sync安装本示例专属依赖示例的 requirements.txt 以file://../../../形式链接到仓库根目录的 mcp-agent 项目本体并额外引入click、openaiuv pip install -r requirements.txt5.2 配置密钥与环境变量cp mcp_agent.secrets.yaml.example mcp_agent.secrets.yaml打开mcp_agent.secrets.yaml填入你所用 LLM 提供商的 API Key以及各 MCP 服务器需要的密钥/令牌。该文件通常应加入.gitignore避免敏感信息进入版本库。5.3 启动服务端与客户端先在一个终端启动无状态服务器uv run stateless_server.py此时服务器会以 INFO 级别日志输出Application started with StreamableHTTP session manager!并监听0.0.0.0:3156。另开一个终端运行 mcp-agent 应用uv run main.py即可观察到上述连接、工具列举与会话 ID 的输出。六、扩展从无状态到有状态的 Streamable HTTP 会话本示例刻意选择了无状态服务器目的是让读者先看清 Streamable HTTP 传输的最小闭环。如果需要构造有状态会话只需调整服务端的StreamableHTTPSessionManager参数提供event_store并关闭stateless此时客户端在后续请求中会自动携带MCP_SESSION_ID头见 会话 ID 注入逻辑get_server_session()返回的session_id将不再为空。这一机制为跨请求保持对话上下文、实现长任务与断点续传提供了基础。七、小结通过examples/mcp/mcp_streamable_http示例你可以一次性掌握 Streamable HTTP 传输在 mcp-agent 中的完整落地路径服务端用StreamableHTTPSessionManager(statelessTrue)快速搭建无状态 MCP 服务客户端用transport: streamable_httpurl在 mcp_agent.config.yaml 中声明连接运行时通过list_tools验证握手、通过get_server_session观察会话状态。若需在生产环境使用可进一步组合headers、http_timeout_seconds、read_timeout_seconds、terminate_on_close等连接参数或为服务器配置 OAuth 认证将 Streamable HTTP 打造成 agent 应用与远程 MCP 服务之间的可靠通信通道。【免费下载链接】mcp-agentBuild effective agents using Model Context Protocol and simple workflow patterns项目地址: https://gitcode.com/GitHub_Trending/mc/mcp-agent创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
