barrel_mcp_client_http (barrel_mcp v3.0.0)

View Source

Streamable HTTP transport for barrel_mcp_client.

Implements MCP's Streamable HTTP transport (2025-03-26 onward) on the client side:

  • POST every request with Accept: application/json, text/event-stream. The server may answer with a single JSON envelope or with an SSE stream that interleaves server-initiated requests/notifications until the matching response arrives.
  • GET opens a long-lived SSE channel for unsolicited server-to-client traffic. Optional: a server may return 405, in which case server messages only arrive on POST streams.
  • DELETE on close, with the captured Mcp-Session-Id.
  • MCP-Protocol-Version header echoed on every request after the initialize response has been processed by the client.
  • 401, or 403 with insufficient_scope, hands the challenge to the configured auth handle in a worker and reissues the request when it comes back; three rounds at most.

Each parsed SSE event's data: payload is forwarded to the owning client as {mcp_in, self(), Json}. The owner sees the same shape as it does from the stdio transport.

Sections, in file order

  • Transport API and public helpers.
  • gen_server: init, the info handlers for hackney, the auth worker and resume timers.
  • POST request lifecycle: start_post/3, response parsing, finalize_request, the challenge path, SEP-1699 resumption.
  • SSE GET stream, SSE parsing, header helpers, DELETE on close.

State

#state{}: requests maps each in-flight POST's hackney ref to a #req{} (attempts, last event id, resumes); challenged holds the requests waiting on the single auth flow in auth_flow; sse_mode and legacy_sse record how the server delivers unsolicited traffic (a GET, a subscriptions/listen POST, or the 2024-11-05 endpoint); resumes keeps resumption timers so a cancel can drop one; tool_headers caches the x-mcp-header bindings from tools/list. Each field carries its reason inline.

Processes

One gen_server per connection, owned by the client. hackney delivers asynchronously; the auth flow runs in a spawn_monitor worker so a person's consent step never blocks the transport.

Summary

Functions

cancel_request(Pid, RequestId)

Cancel one in-flight request by closing its response stream, which "is itself the cancellation signal" (2026-07-28/basic/patterns/cancellation.mdx:38).

close(Pid)
close_subscription(Pid)

Close the long-lived stream and stop reopening it.

code_change(OldVsn, State, Extra)
connect(Owner, Opts)
handle_call(Msg, From, State)
handle_cast(Msg, State)
handle_info(Msg, State)
init(_)
open_event_stream(Pid)

Open the long-lived GET SSE for unsolicited server messages. Idempotent: a second call while the stream is open is a no-op.

open_subscription(Pid, Body)

Open (or replace) the long-lived stream as a subscriptions/listen POST carrying Body.

same_origin_endpoint(StreamUrl, Ref)

Resolve the endpoint event's URI reference against the stream's URL and require the same origin. The POST carries the session's Authorization header, so an endpoint on another origin would hand the credential to whoever the stream named; the reference client refuses it the same way (mcp/client/sse.py, "Endpoint origin does not match connection origin").

send(Pid, Body)
set_protocol_version(Pid, Version)

Set the negotiated protocol version. Once set, every outgoing request includes the MCP-Protocol-Version header.

set_session_id(Pid, SessionId)

Capture the Mcp-Session-Id returned on the initialize POST so subsequent requests can echo it.

set_tool_headers(Pid, Bindings)

Record which tool arguments each tool wants mirrored into headers, as learned from tools/list.

terminate(Reason, State)

Functions

cancel_request(Pid, RequestId)

Cancel one in-flight request by closing its response stream, which "is itself the cancellation signal" (2026-07-28/basic/patterns/cancellation.mdx:38).

close(Pid)

close_subscription(Pid)

-spec close_subscription(pid()) -> ok.

Close the long-lived stream and stop reopening it.

code_change(OldVsn, State, Extra)

connect(Owner, Opts)

handle_call(Msg, From, State)

handle_cast(Msg, State)

handle_info(Msg, State)

init(_)

open_event_stream(Pid)

Open the long-lived GET SSE for unsolicited server messages. Idempotent: a second call while the stream is open is a no-op.

open_subscription(Pid, Body)

-spec open_subscription(pid(), binary()) -> ok.

Open (or replace) the long-lived stream as a subscriptions/listen POST carrying Body.

Replacing rather than adding: one stream whose filter covers everything subscribed is simpler to reason about than several, and the spec's multiple-subscription support is not needed to express it.

same_origin_endpoint(StreamUrl, Ref)

-spec same_origin_endpoint(binary(), binary()) ->
                              {ok, binary()} |
                              {error, {cross_origin, binary()} | {bad_endpoint, binary()}}.

Resolve the endpoint event's URI reference against the stream's URL and require the same origin. The POST carries the session's Authorization header, so an endpoint on another origin would hand the credential to whoever the stream named; the reference client refuses it the same way (mcp/client/sse.py, "Endpoint origin does not match connection origin").

send(Pid, Body)

set_protocol_version(Pid, Version)

Set the negotiated protocol version. Once set, every outgoing request includes the MCP-Protocol-Version header.

set_session_id(Pid, SessionId)

Capture the Mcp-Session-Id returned on the initialize POST so subsequent requests can echo it.

set_tool_headers(Pid, Bindings)

-spec set_tool_headers(pid(), map()) -> ok.

Record which tool arguments each tool wants mirrored into headers, as learned from tools/list.

terminate(Reason, State)