HTTP streaming với MCP (Streamable HTTP)

Bài này là hướng dẫn toàn diện về streaming an toàn, mở rộng được và theo thời gian thực với Model Context Protocol (MCP) qua HTTPS: vì sao cần streaming, các cơ chế truyền tải (transport) có sẵn, cách cài đặt Streamable HTTP trong MCP, best practice bảo mật, chuyển đổi từ SSE, và hướng dẫn thực hành để xây ứng dụng MCP có streaming của riêng bạn.

Cơ chế truyền tải và streaming trong MCP

Phần này xem xét các cơ chế truyền tải có trong MCP và vai trò của chúng trong việc hỗ trợ streaming để giao tiếp thời gian thực giữa client và server.

Cơ chế truyền tải là gì?

Cơ chế truyền tải quy định cách dữ liệu được trao đổi giữa client và server. MCP hỗ trợ nhiều loại transport cho các môi trường và yêu cầu khác nhau:

• stdio: standard input/output, phù hợp với tool local và dòng lệnh. Đơn giản nhưng không phù hợp cho web hay cloud.
• HTTP+SSE: transport từ xa cũ, bị ngừng khuyến nghị (deprecated) từ MCP 2025-03-26 và được thay bằng Streamable HTTP. Không dùng cho phần cài đặt mới.
• Streamable HTTP: transport streaming hiện đại dựa trên HTTP, hỗ trợ thông báo (notification) và mở rộng tốt hơn. Được khuyến nghị cho phần lớn tình huống production và cloud.

Bảng so sánh

Các transport chuẩn là stdio và Streamable HTTP. HTTP+SSE chỉ còn xuất hiện trong các ví dụ cũ.

Streaming: khái niệm và động lực

Hiểu khái niệm và lý do của streaming là thiết yếu để xây hệ thống giao tiếp thời gian thực hiệu quả.

Streaming là kỹ thuật lập trình mạng cho phép gửi và nhận dữ liệu thành các mẩu nhỏ, dễ quản lý, hoặc như một chuỗi sự kiện, thay vì chờ toàn bộ phản hồi sẵn sàng. Đặc biệt hữu ích cho:

• File hoặc tập dữ liệu lớn.
• Cập nhật thời gian thực (chat, thanh tiến độ).
• Tính toán chạy lâu mà bạn muốn người dùng luôn được cập nhật.

Ở mức tổng quan, cần biết:

• Dữ liệu được gửi dần dần, không phải một lần.
• Client xử lý được dữ liệu ngay khi nhận.
• Giảm độ trễ cảm nhận và cải thiện trải nghiệm người dùng.

Vì sao dùng streaming?

• Người dùng nhận phản hồi ngay, không phải chờ tới cuối.
• Cho phép ứng dụng thời gian thực và giao diện phản hồi nhanh.
• Dùng tài nguyên mạng và tính toán hiệu quả hơn.

Ví dụ đơn giản: server và client HTTP streaming

Đây là ví dụ đơn giản về cách cài đặt streaming:

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")

import requests

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

@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));
    }
}

@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();
    }
}

So sánh: streaming cổ điển và streaming trong MCP

Các khác biệt chính

• Kiểu giao tiếp: HTTP streaming cổ điển dùng chunked transfer encoding đơn giản để gửi dữ liệu thành từng mẩu; streaming MCP dùng hệ thống notification có cấu trúc với giao thức JSON-RPC.
• Định dạng thông điệp: HTTP cổ điển gửi các mẩu văn bản thuần kèm ký tự xuống dòng; MCP gửi các đối tượng LoggingMessageNotification có cấu trúc kèm metadata.
• Cài đặt client: client HTTP cổ điển đơn giản, chỉ xử lý phản hồi dạng stream; client MCP phức tạp hơn, có message handler để xử lý nhiều loại thông điệp.
• Cập nhật tiến độ: với HTTP cổ điển, tiến độ là một phần của luồng phản hồi chính; với MCP, tiến độ được gửi qua các notification riêng, còn phản hồi chính tới ở cuối.

Khuyến nghị

Khi chọn giữa streaming cổ điển (như endpoint /stream ở trên) và streaming qua MCP:

• Nhu cầu streaming đơn giản: HTTP streaming cổ điển dễ cài đặt hơn và đủ dùng.
• Ứng dụng phức tạp, tương tác: streaming MCP có cấu trúc hơn, metadata phong phú hơn và tách notification khỏi kết quả cuối.
• Ứng dụng AI: hệ thống notification của MCP đặc biệt hữu ích cho các tác vụ AI chạy lâu, khi bạn muốn người dùng biết tiến độ.

Streaming trong MCP

Hiểu streaming hoạt động thế nào trong MCP là thiết yếu để xây ứng dụng phản hồi nhanh, cập nhật thời gian thực cho người dùng trong các thao tác chạy lâu.

Trong MCP, streaming không phải là gửi phản hồi chính thành từng mẩu, mà là gửi notification tới client trong khi một tool đang xử lý yêu cầu (request). Các notification này có thể gồm cập nhật tiến độ, log hoặc sự kiện khác.

Cách hoạt động

Kết quả chính vẫn được gửi như một phản hồi (response) duy nhất. Tuy nhiên, notification có thể được gửi thành các thông điệp riêng trong lúc xử lý, nhờ đó cập nhật client theo thời gian thực. Client phải xử lý và hiển thị được các notification này.

Bài thực hành tuỳ chọn: kết nối tới một MCP server được host sẵn

Bạn cũng có thể dùng Streamable HTTP mà không cần chạy server local. Ví dụ này kết nối tới Parallel Search MCP, khám phá các tool của nó và tìm tài liệu MCP công khai, dùng cùng Python SDK với client local.

Endpoint ẩn danh của Parallel không cần tài khoản hay API key. Truy cập miễn phí bị giới hạn tần suất. Chạy script này sẽ gửi các truy vấn tìm kiếm, mục tiêu và một mã phiên ngẫu nhiên tới Parallel. Dịch vụ còn có web_fetch, gửi các URL được yêu cầu và mọi ngữ cảnh đi kèm tới Parallel. Hãy dùng thông tin công khai cho bài này; xem điều khoản và chính sách quyền riêng tư của dịch vụ.

Với Python 3.10 trở lên và môi trường ảo đã kích hoạt, cài SDK:

python -m pip install "mcp>=1.10,<2"

Lưu đoạn sau thành hosted_search.py và chạy python hosted_search.py:

import asyncio
from uuid import uuid4

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client


async def main() -> None:
    session_id = str(uuid4())
    async with streamablehttp_client("https://search.parallel.ai/mcp") as (
        read_stream,
        write_stream,
        _,
    ):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            tools = await session.list_tools()
            print("Available tools:", [tool.name for tool in tools.tools])

            result = await session.call_tool(
                "web_search",
                {
                    "objective": "Find the official MCP Streamable HTTP documentation",
                    "search_queries": ["MCP Streamable HTTP documentation"],
                    "session_id": session_id,
                },
            )
            if result.isError:
                raise RuntimeError(f"Search tool failed: {result.content}")
            for block in result.content:
                if block.type == "text":
                    print(block.text)


async def run() -> None:
    await asyncio.wait_for(main(), timeout=60)


if __name__ == "__main__":
    asyncio.run(run())

Kết quả khám phá sẽ có web_search và web_fetch, tiếp theo là kết quả tìm kiếm gồm URL nguồn và trích đoạn. Kết quả có thể khác nhau hoặc rỗng. Script kiểm tra isError vì một tool có thể thất bại ngay cả khi request HTTP thành công. Nếu bị giới hạn tần suất, hãy chờ rồi thử lại. Dùng lại cùng session_id nếu bạn mở rộng script với các lời gọi tìm kiếm hoặc fetch liên quan.

Streamable HTTP cho phép cả phản hồi JSON lẫn SSE; server này có thể trả về kết quả JSON hoàn chỉnh mà không có notification tiến độ. SDK lo phần transport. Hãy tiếp tục với ví dụ local bên dưới để học về notification. Script tuỳ chọn này thực hiện một lần tìm kiếm tường minh và đóng kết nối khi xong. Nếu sau này bạn đưa các tool này cho một agent, agent có thể gọi chúng trong quá trình làm việc; hãy coi văn bản web lấy về là dữ liệu không đáng tin.

Notification là gì?

Notification (thông báo) là một thông điệp JSON-RPC không có id và không nhận phản hồi. MCP dùng notification cho tiến độ, huỷ bỏ và các sự kiện một chiều khác.

Trong MCP 2025-11-25, client gửi notifications/initialized sau khi bắt tay khởi tạo. MCP 2026-07-28 không còn bước bắt tay khởi tạo, nên notification này là hành vi cũ.

Một notification có dạng thông điệp JSON như sau:

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

Logging là một tính năng dùng notification; bản thân notification là một kiểu thông điệp JSON-RPC tổng quát.

Với phần cài đặt cũ theo 2025-11-25, server bật capability Logging như sau:

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

Tuỳ SDK, logging có thể được bật sẵn, hoặc bạn cần bật tường minh trong cấu hình server.

Các mức notification:

Cài đặt notification trong MCP

Để cài đặt notification trong MCP, bạn cần thiết lập cả phía server lẫn client để xử lý cập nhật thời gian thực, nhờ đó ứng dụng phản hồi ngay cho người dùng trong các thao tác chạy lâu.

Phía server: gửi notification

Trong MCP, bạn định nghĩa tool có thể gửi notification khi đang xử lý request. Server dùng đối tượng context (thường là ctx) để gửi thông điệp tới client.

@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}")

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

[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}"
    };
}

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

Phía client: nhận notification

Client phải cài đặt một message handler để xử lý và hiển thị notification khi chúng tới.

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:

// 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

Để bật notification, hãy đảm bảo server dùng transport streaming (như streamable-http) và client cài đặt message handler để xử lý notification.

Notification tiến độ và các tình huống

Notification tiến độ là thông điệp thời gian thực server gửi tới client trong các thao tác chạy lâu. Thay vì chờ cả quá trình xong, server liên tục cập nhật trạng thái cho client. Điều này tăng tính minh bạch, cải thiện trải nghiệm và giúp gỡ lỗi dễ hơn.

Ví dụ:

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

Vì sao dùng notification tiến độ?

• Trải nghiệm tốt hơn: người dùng thấy cập nhật trong lúc xử lý, không chỉ ở cuối.
• Phản hồi thời gian thực: client hiển thị được thanh tiến độ hoặc log, ứng dụng có cảm giác phản hồi nhanh.
• Gỡ lỗi và giám sát dễ hơn: lập trình viên và người dùng thấy được chỗ quá trình chậm hoặc bị kẹt.

Cách cài đặt notification tiến độ

• Ở server: dùng ctx.info() hoặc ctx.log() để gửi notification khi xử lý từng mục. Thông điệp được gửi tới client trước khi kết quả chính sẵn sàng.
• Ở client: cài đặt message handler lắng nghe và hiển thị notification khi chúng tới. Handler này phân biệt notification với kết quả cuối.

Ví dụ server (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}")

Ví dụ client (Python):

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

Cân nhắc bảo mật

Bảo mật phải được ưu tiên hàng đầu khi cài đặt bất kỳ server nào, nhất là khi dùng transport dựa trên HTTP như Streamable HTTP trong MCP. Streamable HTTP mở ra bề mặt tấn công mới và cần cấu hình cẩn thận.

Tổng quan

Các cân nhắc bảo mật chính:

• Kiểm tra header Origin: luôn kiểm tra header Origin để chống tấn công DNS rebinding.
• Bind vào localhost: khi phát triển local, bind server vào localhost để không lộ ra Internet công khai.
• Xác thực: cài đặt xác thực (như API key, OAuth) cho môi trường production.
• CORS: cấu hình chính sách Cross-Origin Resource Sharing (CORS) để giới hạn truy cập.
• HTTPS: dùng HTTPS trong production để mã hoá lưu lượng.

Best practice

• Không bao giờ tin request tới khi chưa kiểm tra.
• Ghi log và giám sát mọi truy cập và lỗi.
• Cập nhật dependency thường xuyên để vá lỗ hổng bảo mật.

Thách thức

• Cân bằng giữa bảo mật và sự thuận tiện khi phát triển.
• Đảm bảo tương thích với nhiều môi trường client khác nhau.

Nâng cấp từ SSE lên Streamable HTTP

Với ứng dụng đang dùng Server-Sent Events (SSE), chuyển sang Streamable HTTP mang lại nhiều khả năng hơn và bền vững hơn về lâu dài.

Vì sao nâng cấp?

• Streamable HTTP mở rộng tốt hơn, tương thích tốt hơn và hỗ trợ notification phong phú hơn SSE.
• Đây là transport được khuyến nghị cho ứng dụng MCP mới.

Các bước chuyển đổi

• Cập nhật code server để dùng transport="streamable-http" trong mcp.run().
• Cập nhật code client để dùng streamablehttp_client thay cho client SSE.
• Cài đặt message handler trong client để xử lý notification.
• Kiểm thử khả năng tương thích với các tool và quy trình hiện có.

Duy trì tương thích

Nên duy trì tương thích với các client SSE hiện có trong quá trình chuyển đổi:

• Có thể hỗ trợ cả SSE lẫn Streamable HTTP bằng cách chạy hai transport trên hai endpoint khác nhau.
• Chuyển dần các client sang transport mới.

Thách thức

• Đảm bảo mọi client đều được cập nhật.
• Xử lý khác biệt trong cách gửi notification.

Bài tập: xây ứng dụng MCP có streaming

Tình huống: xây một MCP server và client, trong đó server xử lý một danh sách mục (như file hoặc tài liệu) và gửi một notification cho mỗi mục đã xử lý. Client hiển thị từng notification khi nó tới.

Các bước:

1. Cài đặt một tool phía server xử lý danh sách và gửi notification cho từng mục.
2. Cài đặt client có message handler để hiển thị notification theo thời gian thực.
3. Kiểm thử bằng cách chạy cả server và client, quan sát các notification.

Lời giải: solution.

Đọc thêm và bước tiếp theo

Đọc thêm

• Microsoft: Introduction to HTTP Streaming
• Microsoft: Server-Sent Events (SSE)
• Microsoft: CORS in ASP.NET Core
• Python requests: Streaming Requests

Bước tiếp theo

• Thử xây các tool MCP nâng cao hơn dùng streaming cho phân tích thời gian thực, chat hoặc chỉnh sửa cộng tác.
• Tìm hiểu cách tích hợp streaming MCP với framework frontend (React, Vue…) để cập nhật giao diện trực tiếp.
• Bài tiếp theo trong khoá này: Gỡ lỗi với MCP Inspector.

Nguồn

• Dịch và biên tập từ microsoft/mcp-for-beginners · 03-GettingStarted/06-http-streaming/README.md, giấy phép MIT.