Streamable HTTP

Streamable HTTP 는 URL 뒤에 살고 싶은 MCP server 를 위한 transport 야. 공유 서비스, cloud 에 올린 server, 여러 고객을 한꺼번에 받는 server. 예전의 'HTTP+SSE' transport 를 (지금은 deprecated 됐어) endpoint 하나로 갈아치웠고, 그 하나가 양쪽 방향을 다 평범한 HTTP 로 처리해.

선 위의 모양은 이래. Client 가 server endpoint 에 JSON-RPC request 를 POST 하면, server 는 둘 중 하나로 답해. 간단한 요청이면 평범한 HTTP response 본문에 JSON-RPC 결과를 담아 보내. 오래 걸리거나 흘려보내야 하는 요청이면 Server-Sent Events (SSE) stream 을 돌려주고, JSON-RPC 메시지를 준비되는 대로 data: 줄에 실어 보내. 어느 쪽이든 JSON-RPC 의 틀은 똑같아. HTTP 봉투 덕에 server 가 '한 번에 끝낼지' '흘려보낼지' 를 그때그때 고를 수 있는 것뿐이야.

Server 에서 client 로 가는 알림도 (notifications/tools/listChanged 같은 거) 같은 SSE stream 을 타고 흘러. Client 가 endpoint 에 오래 열어두는 GET 을 하나 걸어두면, server 가 일이 생길 때마다 밀어줘. TCP 연결 하나에 논리적으로는 채널 둘인 셈이지.

옛 HTTP+SSE 보다 운영이 편해진 게 커. Endpoint 하나, auth 흐름 하나, CORS 규칙 하나면 되고 여러 URL 로 흩어질 일이 없어. Stdio 와 비교하면 이점은 말할 것도 없고 — 네트워크가 되니까. MCP server 를 서비스로 내놓는다면 이 모양이야.

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