使用模型上下文协议 (MCP) 的 HTTPS 流式传输

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import time

app = FastAPI()

async def event_stream():
    for i in range(1, 6):
        yield f"data: Message {i}\n\n"
        time.sleep(1)

@app.get("/stream")
def stream():
    return StreamingResponse(event_stream(), media_type="text/event-stream")

客户端（使用 requests）：

import requests

with requests.get("http://localhost:8000/stream", stream=True) as r:
    for line in r.iter_lines():
        if line:
            print(line.decode())

此示例展示了服务器如何在消息准备好时逐条发送给客户端，而不是等待所有消息准备好后再发送。

工作原理：

服务器在每条消息准备好时生成并发送。
客户端在每个数据块到达时接收并打印。

要求：

服务器必须使用流式响应（例如 FastAPI 中的 StreamingResponse）。
客户端必须将响应作为流处理（stream=True）。
Content-Type 通常为 text/event-stream 或 application/octet-stream。

Java

服务器端（使用 Spring Boot 和服务器发送事件）：

@RestController
public class CalculatorController {

    @GetMapping(value = "/calculate", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<String>> calculate(@RequestParam double a,
                                                   @RequestParam double b,
                                                   @RequestParam String op) {
        
        double result;
        switch (op) {
            case "add": result = a + b; break;
            case "sub": result = a - b; break;
            case "mul": result = a * b; break;
            case "div": result = b != 0 ? a / b : Double.NaN; break;
            default: result = Double.NaN;
        }

        return Flux.<ServerSentEvent<String>>just(
                    ServerSentEvent.<String>builder()
                        .event("info")
                        .data("Calculating: " + a + " " + op + " " + b)
                        .build(),
                    ServerSentEvent.<String>builder()
                        .event("result")
                        .data(String.valueOf(result))
                        .build()
                )
                .delayElements(Duration.ofSeconds(1));
    }
}

客户端（使用 Spring WebFlux WebClient）：

@SpringBootApplication
public class CalculatorClientApplication implements CommandLineRunner {

    private final WebClient client = WebClient.builder()
            .baseUrl("http://localhost:8080")
            .build();

    @Override
    public void run(String... args) {
        client.get()
                .uri(uriBuilder -> uriBuilder
                        .path("/calculate")
                        .queryParam("a", 7)
                        .queryParam("b", 5)
                        .queryParam("op", "mul")
                        .build())
                .accept(MediaType.TEXT_EVENT_STREAM)
                .retrieve()
                .bodyToFlux(String.class)
                .doOnNext(System.out::println)
                .blockLast();
    }
}

Java 实现注意事项：

使用 Spring Boot 的响应式栈，通过 Flux 实现流式传输。
ServerSentEvent 提供了带有事件类型的结构化事件流。
WebClient 的 bodyToFlux() 支持响应式流式消费。
delayElements() 模拟事件之间的处理时间。
事件可以包含类型（如 info、result），以便客户端更好地处理。

对比：经典流式传输与 MCP 流式传输

经典 HTTP 流式传输与 MCP 流式传输的区别如下：

特性	经典 HTTP 流式传输	MCP 流式传输（通知）
主响应	分块传输	最终单一响应
进度更新	作为数据块发送	作为通知发送
客户端要求	必须处理流式响应	必须实现消息处理器
使用场景	大型文件、AI 令牌流	进度、日志、实时反馈

观察到的主要差异

此外，还有以下关键差异：

通信模式：
- 经典 HTTP 流式传输：使用简单的分块传输编码发送数据块。
- MCP 流式传输：使用 JSON-RPC 协议的结构化通知系统。
消息格式：
- 经典 HTTP：纯文本块，使用换行符分隔。
- MCP：带有元数据的结构化 LoggingMessageNotification 对象。
客户端实现：
- 经典 HTTP：简单的客户端处理流式响应。
- MCP：更复杂的客户端，需要消息处理器来处理不同类型的消息。
进度更新：
- 经典 HTTP：进度是主响应流的一部分。
- MCP：进度通过单独的通知消息发送，而主响应在最后返回。

MCP 中的流式传输

好了，您已经看到了经典流式传输与 MCP 流式传输的对比和建议。接下来，我们将详细介绍如何在 MCP 中利用流式传输。

在 MCP 框架中，流式传输的重点不是将主响应分块发送，而是在工具处理请求时向客户端发送通知。这些通知可以包括进度更新、日志或其他事件。

工作原理

主结果仍然作为单一响应发送。然而，在处理过程中，通知可以作为单独的消息发送，从而实时更新客户端。客户端必须能够处理并显示这些通知。

什么是通知？

我们提到“通知”，那么在 MCP 中它具体指什么？

通知是服务器在长时间运行的操作期间发送给客户端的消息，用于告知进度、状态或其他事件。通知提高了透明度和用户体验。

例如，在客户端与服务器完成初始握手后，服务器应发送一条通知。

通知的 JSON 消息格式如下：

{
  jsonrpc: "2.0";
  method: string;
  params?: {
    [key: string]: unknown;
  };
}

通知属于 MCP 中称为 "Logging" 的主题。

要启用日志记录，服务器需要像这样将其作为功能/能力启用：

{
  "capabilities": {
    "logging": {}
  }
}

!NOTE 根据所使用的 SDK，日志记录可能默认启用，也可能需要在服务器配置中显式启用。

通知有不同的类型：

级别	描述	示例使用场景
debug	详细的调试信息	函数入口/退出点
info	一般信息性消息	操作进度更新
notice	正常但重要的事件	配置更改
warning	警告条件	使用已弃用的功能
error	错误条件	操作失败
critical	严重条件	系统组件故障
alert	必须立即采取行动	检测到数据损坏
emergency	系统不可用	完全系统故障

在 MCP 中实现通知

要在 MCP 中实现通知，您需要在服务器端和客户端设置处理实时更新的功能。这使您的应用程序能够在长时间运行的操作期间向用户提供即时反馈。

服务器端：发送通知

从服务器端开始。在 MCP 中，您可以定义工具，在处理请求时发送通知。服务器使用上下文对象（通常是 ctx）向客户端发送消息。

Python

@mcp.tool(description="A tool that sends progress notifications")
async def process_files(message: str, ctx: Context) -> TextContent:
    await ctx.info("Processing file 1/3...")
    await ctx.info("Processing file 2/3...")
    await ctx.info("Processing file 3/3...")
    return TextContent(type="text", text=f"Done: {message}")

在上述示例中，process_files 工具在处理每个文件时向客户端发送三条通知。ctx.info() 方法用于发送信息性消息。

此外，为了启用通知，请确保您的服务器使用流式传输（如 streamable-http），并且客户端实现了消息处理器来处理通知。以下是设置服务器使用 streamable-http 传输的方法：

mcp.run(transport="streamable-http")

.NET

[Tool("A tool that sends progress notifications")]
public async Task<TextContent> ProcessFiles(string message, ToolContext ctx)
{
    await ctx.Info("Processing file 1/3...");
    await ctx.Info("Processing file 2/3...");
    await ctx.Info("Processing file 3/3...");
    return new TextContent
    {
        Type = "text",
        Text = $"Done: {message}"
    };
}

在此 .NET 示例中，ProcessFiles 工具通过 Tool 属性定义，并在处理每个文件时向客户端发送三条通知。ctx.Info() 方法用于发送信息性消息。

要在 .NET MCP 服务器中启用通知，请确保您使用流式传输：

var builder = McpBuilder.Create();
await builder
    .UseStreamableHttp() // Enable streamable HTTP transport
    .Build()
    .RunAsync();

客户端：接收通知

客户端必须实现消息处理器，以便在通知到达时处理并显示。

Python

async def message_handler(message):
    if isinstance(message, types.ServerNotification):
        print("NOTIFICATION:", message)
    else:
        print("SERVER MESSAGE:", message)

async with ClientSession(
   read_stream, 
   write_stream,
   logging_callback=logging_collector,
   message_handler=message_handler,
) as session:

在上述代码中，message_handler 函数检查传入消息是否为通知。如果是，则打印通知；否则将其作为常规服务器消息处理。此外，注意如何使用 message_handler 初始化 ClientSession 来处理传入通知。

.NET

// Define a message handler
void MessageHandler(IJsonRpcMessage message)
{
    if (message is ServerNotification notification)
    {
        Console.WriteLine($"NOTIFICATION: {notification}");
    }
    else
    {
        Console.WriteLine($"SERVER MESSAGE: {message}");
    }
}

// Create and use a client session with the message handler
var clientOptions = new ClientSessionOptions
{
    MessageHandler = MessageHandler,
    LoggingCallback = (level, message) => Console.WriteLine($"[{level}] {message}")
};

using var client = new ClientSession(readStream, writeStream, clientOptions);
await client.InitializeAsync();

// Now the client will process notifications through the MessageHandler

在此 .NET 示例中，MessageHandler 函数检查传入消息是否为通知。如果是，则打印通知；否则将其作为常规服务器消息处理。通过 ClientSessionOptions 将消息处理器与 ClientSession 一起初始化。

为了启用通知，请确保您的服务器使用流式传输（如 streamable-http），并且客户端实现了消息处理器来处理通知。

进度通知与场景

本节解释了 MCP 中进度通知的概念、重要性，以及如何使用可流式 HTTP 实现它们。您还将找到一个实践任务来巩固您的理解。

进度通知是服务器在长时间运行的操作期间发送给客户端的实时消息。服务器在整个过程完成之前向客户端更新当前状态。这提高了透明度、用户体验，并使调试更容易。

示例：


"Processing document 1/10"
"Processing document 2/10"
...
"Processing complete!"

为什么使用进度通知？

进度通知的重要性体现在以下几个方面：

更好的用户体验： 用户可以在操作进行时看到更新，而不是等到结束。
实时反馈： 客户端可以显示进度条或日志，使应用程序更具响应性。
更容易调试和监控： 开发者和用户可以看到流程可能变慢或卡住的地方。

如何实现进度通知

以下是如何在 MCP 中实现进度通知：

在服务器端： 使用 ctx.info() 或 ctx.log() 在每个项目处理时发送通知。这会在主结果准备好之前向客户端发送消息。
在客户端： 实现一个消息处理器，监听并显示到达的通知。该处理器需要区分通知和最终结果。

服务器示例：

Python

@mcp.tool(description="A tool that sends progress notifications")
async def process_files(message: str, ctx: Context) -> TextContent:
    for i in range(1, 11):
        await ctx.info(f"Processing document {i}/10")
    await ctx.info("Processing complete!")
    return TextContent(type="text", text=f"Done: {message}")

客户端示例：

Python

async def message_handler(message):
    if isinstance(message, types.ServerNotification):
        print("NOTIFICATION:", message)
    else:
        print("SERVER MESSAGE:", message)