Toolnexus.Mcp.Transport.StreamableHttp (toolnexus v0.9.4)

Copy Markdown View Source

Streamable-HTTP transport for a remote MCP server (SPEC §2 "remote"), built on Req. POSTs JSON-RPC to the configured url with the configured headers (after ${ENV_VAR} expansion — values are never logged), accepts either a JSON response or an SSE stream, carries the Mcp-Session-Id response header onto subsequent requests, and DELETEs the session on close.

Legacy SSE fallback (js/src/mcp.ts decision flow): if the streamable POST initialize fails for any reason, GET an SSE stream that supplies an endpoint event, then POST subsequent messages to that endpoint; responses arrive on the GET stream (delivered to the owner as {:transport_msg, msg}).

Summary

Types

t()

Functions

close(t)

Close: DELETE the session if one was established; kill the legacy streamer.

connect(cfg, owner, elicitation?)

Connect: POST initialize (streamable), capture the session header, send notifications/initialized. On any initialize failure, try the legacy SSE fallback. owner is the connection pid that will receive {:transport_msg, _} messages in legacy mode (retargeted from the calling process after connect).

expand_env_headers(headers)

Expand ${ENV_VAR} in header values from the environment so tokens live in the environment, not the committed config. Values are never logged.

post_rpc(t, id, msg)

Streamable-mode request/response: POST the message, parse the JSON or SSE reply, return {:ok, result, extra_msgs, t} where extra_msgs are any other JSON-RPC messages (server requests/notifications) seen on the stream.

send_msg(t, msg)

Fire-and-forget send (notifications, responses to server requests; legacy-mode requests).

Types

t()

@type t() :: %Toolnexus.Mcp.Transport.StreamableHttp{
  endpoint: term(),
  headers: term(),
  mode: term(),
  protocol_version: term(),
  session: term(),
  streamer: term(),
  timeout: term(),
  url: term()
}

Functions

close(t)

Close: DELETE the session if one was established; kill the legacy streamer.

connect(cfg, owner, elicitation?)

Connect: POST initialize (streamable), capture the session header, send notifications/initialized. On any initialize failure, try the legacy SSE fallback. owner is the connection pid that will receive {:transport_msg, _} messages in legacy mode (retargeted from the calling process after connect).

expand_env_headers(headers)

Expand ${ENV_VAR} in header values from the environment so tokens live in the environment, not the committed config. Values are never logged.

post_rpc(t, id, msg)

Streamable-mode request/response: POST the message, parse the JSON or SSE reply, return {:ok, result, extra_msgs, t} where extra_msgs are any other JSON-RPC messages (server requests/notifications) seen on the stream.

send_msg(t, msg)

Fire-and-forget send (notifications, responses to server requests; legacy-mode requests).