拓冰网站定制 
教程文档人工智能【免费下载链接】mcp-for-beginnersThis open-source curriculum introduces the fundamentals of Model Context Protocol (MCP) through real-world, cross-language examples in .NET, Java, TypeScript, JavaScript, Rust and Python. Designed for developers, it focuses on practical techniques for building modular, scalable, and secure AI workflows from session setup to service orchestration.项目地址https://gitcode.com/GitHub_Trending/mc/mcp-for-beginners点击查看免费下载导读本篇指南基于 mcp-for-beginners 课程中 06-http-streaming 章节 的 Python 解决方案完整讲解如何在一套代码中同时实现两种流式方案基于 FastAPI 的经典 HTTP 分块流式传输以及基于 MCP streamable-http 传输的实时进度通知。读完本文你将掌握 FastMCP 工具定义、ctx.info()通知发送、streamablehttp_client客户端会话、消息处理器message handler等核心技能能够独立构建带实时进度反馈的 MCP 流式应用。环境准备与项目结构前提条件Python 3.9 或更高版本mcpPython 包pip install mcpfastapi与requests用于经典 HTTP 流式演示。安装与初始化git clone https://gitcode.com/GitHub_Trending/mc/mcp-for-beginners创建并激活虚拟环境推荐python -m venv venv .\venv\Scripts\Activate.ps1 # Windows # 或 source venv/bin/activate # Linux/macOS安装依赖pip install mcp[cli] fastapi requestsmcp[cli]额外安装命令行工具fastapi提供经典流式 HTTP 服务端requests用于经典流式客户端。涉及的核心文件文件作用server.py一个文件双模式默认启动经典 HTTP 流式服务加mcp参数启动 MCP streamable-http 服务client.py默认作为经典 HTTP 流式客户端加mcp参数作为 MCP 客户端welcome.html根路径/的演示说明页展示两种流式方案的入口章节主文档流式概念、传输机制对比与完整理论讲解经典 HTTP 流式传输最简单直接的流式方案经典 HTTP 流式传输采用分块传输编码chunked transfer encoding服务端逐块产出数据客户端逐行消费适合大文件下载、AI Token 流式输出等简单场景。启动服务端cd 03-GettingStarted/06-http-streaming/solution python server.py预期输出Starting FastAPI server for classic HTTP streaming... INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRLC to quit)对应源码逻辑在 server.py 的__main__分支不带mcp参数时通过uvicorn.run(server:app, host127.0.0.1, port8000, reloadTrue)启动 FastAPI。流式端点实现async def event_stream(message: str): for i in range(1, 4): yield fProcessing file {i}/3...\n await asyncio.sleep(1) yield fHeres the file content: {message}\n app.get(/stream) async def stream(message: str hello): return StreamingResponse(event_stream(message), media_typetext/plain)关键点服务端用async生成器逐条yield数据配合asyncio.sleep(1)模拟每项 1 秒的处理耗时StreamingResponse负责将生成器转化为分块流式响应media_typetext/plain声明纯文本内容message是可选查询参数默认值为hello。启动客户端另开一个终端保持同一虚拟环境与目录cd 03-GettingStarted/06-http-streaming/solution python client.py预期输出Running classic HTTP streaming client... Connecting to http://localhost:8000/stream with message: hello --- Streaming Progress --- Processing file 1/3... Processing file 2/3... Processing file 3/3... Heres the file content: hello --- Stream Ended ---客户端核心在 client.py 的stream_progress()def stream_progress(messagehello, urlhttp://localhost:8000/stream): params {message: message} with requests.get(url, paramsparams, streamTrue, timeout10) as r: r.raise_for_status() for line in r.iter_lines(): if line: decoded_line line.decode().strip() print(decoded_line)两个不可省略的要素客户端streamTrue开启流式请求否则requests会等待完整响应服务端必须使用流式响应对象如StreamingResponse。MCP 流式传输基于通知的实时进度反馈MCP 的流式并不把主响应分块而是在工具处理过程中通过通知Notification向客户端推送进度、日志等事件主结果仍在结束时一次性返回。通知是不带id、不需要响应的 JSON-RPC 消息格式如下{ jsonrpc: 2.0; method: string; params?: { [key: string]: unknown; }; }启动 MCP 服务端cd 03-GettingStarted/06-http-streaming/solution python server.py mcp预期输出Starting MCP server with streamable-http transport... INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRLC to quit)服务端源码拆解服务端用 FastMCP 定义工具server.pymcp.tool(descriptionA tool that simulates file processing and sends progress notifications) async def process_files(message: str, ctx: Context) - TextContent: files [ffile_{i}.txt for i in range(1, 4)] for idx, file in enumerate(files, 1): await ctx.info(fProcessing {file} ({idx}/{len(files)})...) await asyncio.sleep(1) await ctx.info(All files processed!) return TextContent(typetext, textfProcessed files: {, .join(files)} | Message: {message})实现要点mcp.tool装饰器声明这是一个 MCP 工具description描述工具用途供客户端发现ctx: Context参数是 FastMCP 注入的上下文对象ctx.info()发送 info 级别日志通知ctx.log()可发送其他级别每次通知前用asyncio.sleep(1)模拟处理耗时让客户端能实时看到进度返回TextContent主结果最后一次性返回与实时通知互相独立。启动 MCP 模式时使用 streamable-http 传输server.pyif mcp in sys.argv: print(Starting MCP server with streamable-http transport...) mcp.run(transportstreamable-http)mcp.run(transportstreamable-http)会由 SDK 自建一个带/mcp端点的 FastAPI 应用你无需手动声明路由。启动 MCP 客户端cd 03-GettingStarted/06-http-streaming/solution python client.py mcp预期输出Session ID 每次运行不同Running MCP client... Starting client... Session ID before init: None Session ID after init: a30ab7fca9c84f5fa8f5c54fe56c9612 Session initialized, ready to call tools. Received message: rootLoggingMessageNotification(...) NOTIFICATION: rootLoggingMessageNotification(...) ... Tool result: metaNone content[TextContent(typetext, textProcessed files: file_1.txt, file_2.txt, file_3.txt | Message: hello from client)]客户端源码拆解客户端入口在 client.pyasync def main(): async with streamablehttp_client(fhttp://localhost:{port}/mcp) as ( read_stream, write_stream, session_callback, ): async with ClientSession( read_stream, write_stream, logging_callbacklogging_collector, message_handlermessage_handler, ) as session: id_before session_callback() logger.info(Session ID before init: %s, id_before) await session.initialize() id_after session_callback() logger.info(Session ID after init: %s, id_after) tool_result await session.call_tool(process_files, {message: hello from client}) logger.info(Tool result: %s, tool_result)要点streamablehttp_client建立到http://localhost:8000/mcp的连接解包出读写流和session_callback用于获取会话 IDClientSession接收读写流并挂载logging_callback收集日志通知与message_handler处理任意入站消息session.initialize()完成初始化握手后session_callback()会返回真实的Mcp-Session-Id初始化前为Nonesession.call_tool(process_files, ...)调用服务端工具主结果最后返回期间通知由回调实时接收。消息处理器client.pyasync def message_handler( message: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception, ) - None: logger.info(Received message: %s, message) if isinstance(message, Exception): logger.error(Exception received!) raise message elif isinstance(message, types.ServerNotification): logger.info(NOTIFICATION: %s, message) elif isinstance(message, RequestResponder): logger.info(REQUEST_RESPONDER: %s, message) else: logger.info(SERVER_MESSAGE: %s, message)日志收集器client.py把每个通知参数追加到列表中方便统一回放class LoggingCollector: def __init__(self): self.log_messages: list[types.LoggingMessageNotificationParams] [] async def __call__(self, params: types.LoggingMessageNotificationParams) - None: self.log_messages.append(params) logger.info(MCP Log: %s - %s, params.level, params.data)经典流式 vs MCP 流式如何选型特性经典 HTTP 流式MCP 流式通知主响应分块传输结束时一次性返回进度更新作为数据块发送作为独立通知发送客户端要求处理流式响应实现消息处理器消息格式带换行的纯文本带元数据的结构化 LoggingMessageNotification适用场景大文件、AI Token 流进度、日志、实时反馈选型建议简单流式需求经典 HTTP 流式实现成本低足够胜任复杂交互式应用MCP 流式提供结构化通知与主结果分离元数据更丰富AI 长任务MCP 通知机制特别适合让用户持续感知长任务的执行进度。关键实现步骤速查用FastMCP创建 MCP 服务器定义处理列表并调用ctx.info()/ctx.log()发送通知的工具以transportstreamable-http运行服务器实现带消息处理器的客户端实时展示通知与最终结果。版本兼容性警告章节主文档 README 明确提示本课程的实现示例针对MCP 规范2025-11-25演示的是旧的initialize握手、Mcp-Session-Id、GET 事件流与可恢复模型。MCP2026-07-28规范已移除这些特性新版 Streamable HTTP 请求是自包含的 POST 请求携带MCP-Protocol-Version、Mcp-Method必要时含Mcp-Name头。在新实现中复用这些示例前请先阅读 MCP 2026-07-28 规范变更说明。此外Logging 功能在2026-07-28规范中虽保留兼容但计划在后续修订中移除新实现应优先使用 stdio 的stderr或 OpenTelemetry 做结构化可观测性。故障排查与最佳实践异步优先服务端与客户端均使用async/await保证非阻塞操作避免流式连接被阻塞异常处理客户端message_handler中对Exception分支显式raise服务端也应对工具内异常做兜底保证健壮性多客户端验证同时启动多个python client.py mcp观察每个客户端都收到实时通知验证广播能力依赖检查若启动报错优先确认 Python 版本不低于 3.9、虚拟环境已激活、mcp[cli] fastapi requests三个依赖均已安装会话 ID 观察客户端日志中Session ID before init: None→Session ID after init: id是理解 streamable-http 会话建立过程最直观的验证点安全基线生产环境建议校验Origin头防 DNS rebinding、本地开发绑定 localhost、启用认证API Key/OAuth、配置 CORS 与 HTTPS详见章节主文档的安全小节。延伸阅读章节主文档HTTPS Streaming with MCP传输机制对比、通知级别表、SSE 迁移指南与自建流式 MCP 应用作业Python 解决方案 README.NET 流式服务实现Java SSE 计算器示例MCP 2026-07-28 规范变更赞分享教程文档人工智能【免费下载链接】mcp-for-beginnersThis open-source curriculum introduces the fundamentals of Model Context Protocol (MCP) through real-world, cross-language examples in .NET, Java, TypeScript, JavaScript, Rust and Python. Designed for developers, it focuses on practical techniques for building modular, scalable, and secure AI workflows from session setup to service orchestration.项目地址https://gitcode.com/GitHub_Trending/mc/mcp-for-beginners点击查看免费下载相关推荐MCP for Beginners 实战用 Python 实现 HTTP Streaming 与 Streamable HTTP 进度通知MCP for Beginners 实战用 Python 实现 HTTP Streaming 与 Streamable HTTP 进度通知 本指南以 mcp教程文档人工智能Agent Memory Systems 实战指南为 AI Agent 构建可检索、可遗忘、可演进的多层记忆架构AAS 项目详解Agent Memory Systems 实战指南为 AI Agent 构建可检索、可遗忘、可演进的多层记忆架构AAS 项目详解 导读 本文是 agent教程文档人工智能n8n-mcp HTTP流传输实现AI与工作流实时通信的高级配置n8n mcp HTTP流传输实现AI与工作流实时通信的高级配置 为什么选择HTTP流传输 在现代工作流自动化中AI与工作流引擎的实时通信成为提升效率的关MCP 服务AI 应用后端开发工具上一篇react-native-elements Tooltip 组件完全指南从基础用法、全部 Props 到四象限定位算法源码解析下一篇Dear PyGui 快速上手从安装、首个窗口到内置 Demo 的完整入门指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Python 实战：用 FastMCP 实现 HTTP 流式传输与 MCP 实时进度通知
教程文档人工智能【免费下载链接】mcp-for-beginnersThis open-source curriculum introduces the fundamentals of Model Context Protocol (MCP) through real-world, cross-language examples in .NET, Java, TypeScript, JavaScript, Rust and Python. Designed for developers, it focuses on practical techniques for building modular, scalable, and secure AI workflows from session setup to service orchestration.项目地址https://gitcode.com/GitHub_Trending/mc/mcp-for-beginners点击查看免费下载导读本篇指南基于 mcp-for-beginners 课程中 06-http-streaming 章节 的 Python 解决方案完整讲解如何在一套代码中同时实现两种流式方案基于 FastAPI 的经典 HTTP 分块流式传输以及基于 MCP streamable-http 传输的实时进度通知。读完本文你将掌握 FastMCP 工具定义、ctx.info()通知发送、streamablehttp_client客户端会话、消息处理器message handler等核心技能能够独立构建带实时进度反馈的 MCP 流式应用。环境准备与项目结构前提条件Python 3.9 或更高版本mcpPython 包pip install mcpfastapi与requests用于经典 HTTP 流式演示。安装与初始化git clone https://gitcode.com/GitHub_Trending/mc/mcp-for-beginners创建并激活虚拟环境推荐python -m venv venv .\venv\Scripts\Activate.ps1 # Windows # 或 source venv/bin/activate # Linux/macOS安装依赖pip install mcp[cli] fastapi requestsmcp[cli]额外安装命令行工具fastapi提供经典流式 HTTP 服务端requests用于经典流式客户端。涉及的核心文件文件作用server.py一个文件双模式默认启动经典 HTTP 流式服务加mcp参数启动 MCP streamable-http 服务client.py默认作为经典 HTTP 流式客户端加mcp参数作为 MCP 客户端welcome.html根路径/的演示说明页展示两种流式方案的入口章节主文档流式概念、传输机制对比与完整理论讲解经典 HTTP 流式传输最简单直接的流式方案经典 HTTP 流式传输采用分块传输编码chunked transfer encoding服务端逐块产出数据客户端逐行消费适合大文件下载、AI Token 流式输出等简单场景。启动服务端cd 03-GettingStarted/06-http-streaming/solution python server.py预期输出Starting FastAPI server for classic HTTP streaming... INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRLC to quit)对应源码逻辑在 server.py 的__main__分支不带mcp参数时通过uvicorn.run(server:app, host127.0.0.1, port8000, reloadTrue)启动 FastAPI。流式端点实现async def event_stream(message: str): for i in range(1, 4): yield fProcessing file {i}/3...\n await asyncio.sleep(1) yield fHeres the file content: {message}\n app.get(/stream) async def stream(message: str hello): return StreamingResponse(event_stream(message), media_typetext/plain)关键点服务端用async生成器逐条yield数据配合asyncio.sleep(1)模拟每项 1 秒的处理耗时StreamingResponse负责将生成器转化为分块流式响应media_typetext/plain声明纯文本内容message是可选查询参数默认值为hello。启动客户端另开一个终端保持同一虚拟环境与目录cd 03-GettingStarted/06-http-streaming/solution python client.py预期输出Running classic HTTP streaming client... Connecting to http://localhost:8000/stream with message: hello --- Streaming Progress --- Processing file 1/3... Processing file 2/3... Processing file 3/3... Heres the file content: hello --- Stream Ended ---客户端核心在 client.py 的stream_progress()def stream_progress(messagehello, urlhttp://localhost:8000/stream): params {message: message} with requests.get(url, paramsparams, streamTrue, timeout10) as r: r.raise_for_status() for line in r.iter_lines(): if line: decoded_line line.decode().strip() print(decoded_line)两个不可省略的要素客户端streamTrue开启流式请求否则requests会等待完整响应服务端必须使用流式响应对象如StreamingResponse。MCP 流式传输基于通知的实时进度反馈MCP 的流式并不把主响应分块而是在工具处理过程中通过通知Notification向客户端推送进度、日志等事件主结果仍在结束时一次性返回。通知是不带id、不需要响应的 JSON-RPC 消息格式如下{ jsonrpc: 2.0; method: string; params?: { [key: string]: unknown; }; }启动 MCP 服务端cd 03-GettingStarted/06-http-streaming/solution python server.py mcp预期输出Starting MCP server with streamable-http transport... INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRLC to quit)服务端源码拆解服务端用 FastMCP 定义工具server.pymcp.tool(descriptionA tool that simulates file processing and sends progress notifications) async def process_files(message: str, ctx: Context) - TextContent: files [ffile_{i}.txt for i in range(1, 4)] for idx, file in enumerate(files, 1): await ctx.info(fProcessing {file} ({idx}/{len(files)})...) await asyncio.sleep(1) await ctx.info(All files processed!) return TextContent(typetext, textfProcessed files: {, .join(files)} | Message: {message})实现要点mcp.tool装饰器声明这是一个 MCP 工具description描述工具用途供客户端发现ctx: Context参数是 FastMCP 注入的上下文对象ctx.info()发送 info 级别日志通知ctx.log()可发送其他级别每次通知前用asyncio.sleep(1)模拟处理耗时让客户端能实时看到进度返回TextContent主结果最后一次性返回与实时通知互相独立。启动 MCP 模式时使用 streamable-http 传输server.pyif mcp in sys.argv: print(Starting MCP server with streamable-http transport...) mcp.run(transportstreamable-http)mcp.run(transportstreamable-http)会由 SDK 自建一个带/mcp端点的 FastAPI 应用你无需手动声明路由。启动 MCP 客户端cd 03-GettingStarted/06-http-streaming/solution python client.py mcp预期输出Session ID 每次运行不同Running MCP client... Starting client... Session ID before init: None Session ID after init: a30ab7fca9c84f5fa8f5c54fe56c9612 Session initialized, ready to call tools. Received message: rootLoggingMessageNotification(...) NOTIFICATION: rootLoggingMessageNotification(...) ... Tool result: metaNone content[TextContent(typetext, textProcessed files: file_1.txt, file_2.txt, file_3.txt | Message: hello from client)]客户端源码拆解客户端入口在 client.pyasync def main(): async with streamablehttp_client(fhttp://localhost:{port}/mcp) as ( read_stream, write_stream, session_callback, ): async with ClientSession( read_stream, write_stream, logging_callbacklogging_collector, message_handlermessage_handler, ) as session: id_before session_callback() logger.info(Session ID before init: %s, id_before) await session.initialize() id_after session_callback() logger.info(Session ID after init: %s, id_after) tool_result await session.call_tool(process_files, {message: hello from client}) logger.info(Tool result: %s, tool_result)要点streamablehttp_client建立到http://localhost:8000/mcp的连接解包出读写流和session_callback用于获取会话 IDClientSession接收读写流并挂载logging_callback收集日志通知与message_handler处理任意入站消息session.initialize()完成初始化握手后session_callback()会返回真实的Mcp-Session-Id初始化前为Nonesession.call_tool(process_files, ...)调用服务端工具主结果最后返回期间通知由回调实时接收。消息处理器client.pyasync def message_handler( message: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception, ) - None: logger.info(Received message: %s, message) if isinstance(message, Exception): logger.error(Exception received!) raise message elif isinstance(message, types.ServerNotification): logger.info(NOTIFICATION: %s, message) elif isinstance(message, RequestResponder): logger.info(REQUEST_RESPONDER: %s, message) else: logger.info(SERVER_MESSAGE: %s, message)日志收集器client.py把每个通知参数追加到列表中方便统一回放class LoggingCollector: def __init__(self): self.log_messages: list[types.LoggingMessageNotificationParams] [] async def __call__(self, params: types.LoggingMessageNotificationParams) - None: self.log_messages.append(params) logger.info(MCP Log: %s - %s, params.level, params.data)经典流式 vs MCP 流式如何选型特性经典 HTTP 流式MCP 流式通知主响应分块传输结束时一次性返回进度更新作为数据块发送作为独立通知发送客户端要求处理流式响应实现消息处理器消息格式带换行的纯文本带元数据的结构化 LoggingMessageNotification适用场景大文件、AI Token 流进度、日志、实时反馈选型建议简单流式需求经典 HTTP 流式实现成本低足够胜任复杂交互式应用MCP 流式提供结构化通知与主结果分离元数据更丰富AI 长任务MCP 通知机制特别适合让用户持续感知长任务的执行进度。关键实现步骤速查用FastMCP创建 MCP 服务器定义处理列表并调用ctx.info()/ctx.log()发送通知的工具以transportstreamable-http运行服务器实现带消息处理器的客户端实时展示通知与最终结果。版本兼容性警告章节主文档 README 明确提示本课程的实现示例针对MCP 规范2025-11-25演示的是旧的initialize握手、Mcp-Session-Id、GET 事件流与可恢复模型。MCP2026-07-28规范已移除这些特性新版 Streamable HTTP 请求是自包含的 POST 请求携带MCP-Protocol-Version、Mcp-Method必要时含Mcp-Name头。在新实现中复用这些示例前请先阅读 MCP 2026-07-28 规范变更说明。此外Logging 功能在2026-07-28规范中虽保留兼容但计划在后续修订中移除新实现应优先使用 stdio 的stderr或 OpenTelemetry 做结构化可观测性。故障排查与最佳实践异步优先服务端与客户端均使用async/await保证非阻塞操作避免流式连接被阻塞异常处理客户端message_handler中对Exception分支显式raise服务端也应对工具内异常做兜底保证健壮性多客户端验证同时启动多个python client.py mcp观察每个客户端都收到实时通知验证广播能力依赖检查若启动报错优先确认 Python 版本不低于 3.9、虚拟环境已激活、mcp[cli] fastapi requests三个依赖均已安装会话 ID 观察客户端日志中Session ID before init: None→Session ID after init: id是理解 streamable-http 会话建立过程最直观的验证点安全基线生产环境建议校验Origin头防 DNS rebinding、本地开发绑定 localhost、启用认证API Key/OAuth、配置 CORS 与 HTTPS详见章节主文档的安全小节。延伸阅读章节主文档HTTPS Streaming with MCP传输机制对比、通知级别表、SSE 迁移指南与自建流式 MCP 应用作业Python 解决方案 README.NET 流式服务实现Java SSE 计算器示例MCP 2026-07-28 规范变更赞分享教程文档人工智能【免费下载链接】mcp-for-beginnersThis open-source curriculum introduces the fundamentals of Model Context Protocol (MCP) through real-world, cross-language examples in .NET, Java, TypeScript, JavaScript, Rust and Python. Designed for developers, it focuses on practical techniques for building modular, scalable, and secure AI workflows from session setup to service orchestration.项目地址https://gitcode.com/GitHub_Trending/mc/mcp-for-beginners点击查看免费下载相关推荐MCP for Beginners 实战用 Python 实现 HTTP Streaming 与 Streamable HTTP 进度通知MCP for Beginners 实战用 Python 实现 HTTP Streaming 与 Streamable HTTP 进度通知 本指南以 mcp教程文档人工智能Agent Memory Systems 实战指南为 AI Agent 构建可检索、可遗忘、可演进的多层记忆架构AAS 项目详解Agent Memory Systems 实战指南为 AI Agent 构建可检索、可遗忘、可演进的多层记忆架构AAS 项目详解 导读 本文是 agent教程文档人工智能n8n-mcp HTTP流传输实现AI与工作流实时通信的高级配置n8n mcp HTTP流传输实现AI与工作流实时通信的高级配置 为什么选择HTTP流传输 在现代工作流自动化中AI与工作流引擎的实时通信成为提升效率的关MCP 服务AI 应用后端开发工具上一篇react-native-elements Tooltip 组件完全指南从基础用法、全部 Props 到四象限定位算法源码解析下一篇Dear PyGui 快速上手从安装、首个窗口到内置 Demo 的完整入门指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考