Streamable HTTP

Streamable HTTP 가 — URL 뒤에 살고 싶은 — MCP server 의 transport: 공유 service, cloud host, multi-tenant server. 이전 'HTTP+SSE' transport (지금 deprecated) 를 — 양 방향 다 표준 HTTP 로 처리하는 — single endpoint 로 교체.

Wire 모델: client 가 server endpoint 에 JSON-RPC request POST, server 가 둘 중 하나로 응답. Simple request 는 일반 HTTP response body 에 JSON-RPC result. 오래 걸리거나 streaming 한 request 는 Server-Sent Events (SSE) stream 반환 — JSON-RPC 메시지를 ready 될 때마다 data: 줄로 적음. 어느 쪽이든 JSON-RPC framing 동일; HTTP envelope 가 server 한테 'one shot' vs 'stream' 을 runtime 에 고르게 해줌.

Server → client notification (e.g. notifications/tools/listChanged) 도 같은 SSE stream 으로 흘러. Client 가 endpoint 에 long-lived GET 열어서 받음; server 가 발생할 때마다 push. 한 TCP connection, 두 logical channel.

옛 HTTP+SSE 보다 운영적 승리: endpoint 하나, auth flow 하나, CORS rule 하나, 여러 URL fan-out 없음. Stdio 보다 승리는 명백: 네트워크. MCP server-as-a-service 의 모양.

Code

Streamable HTTP — request/response·text

POST /mcp HTTP/1.1
Content-Type: application/json
Authorization: Bearer ...

{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{...}}

HTTP/1.1 200 OK
Content-Type: application/json

{"jsonrpc":"2.0","id":1,"result":{...}}

Streamable HTTP — streaming 결과 SSE·text

POST /mcp HTTP/1.1
Content-Type: application/json
Accept: text/event-stream
Authorization: Bearer ...

{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"build_codebase",...}}

HTTP/1.1 200 OK
Content-Type: text/event-stream

data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"requestId":2,"progress":0.4}}

data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"requestId":2,"progress":0.8}}

data: {"jsonrpc":"2.0","id":2,"result":{"content":[{"type":"text","text":"build complete"}]}}

Streamable HTTP server 연결 (Python client)·python

from mcp.client.streamable_http import streamablehttp_client

async with streamablehttp_client("https://api.example.com/mcp", headers={"Authorization": "Bearer ..."}) as (read, write, _):
    async with ClientSession(read, write) as s:
        await s.initialize()
        tools = await s.list_tools()

Code

External links

Exercise

Progress

댓글 0