Anubis.
Copy Markdown
StreamableHTTP transport implementation for MCP servers.
This module manages SSE (Server-Sent Events) connections for server-to-client communication. In the refactored architecture, request handling is done directly by Session processes - this module only manages SSE handlers and notifications.
Features
- SSE handler registration for server-to-client push
- Automatic handler cleanup on disconnect
- Keepalive messages to maintain connections
- Notification broadcasting to connected clients
- Cross-node delivery: SSE handlers join a
:pggroup keyed by session id, soroute_to_session/3reaches a handler whose GET stream landed on a different node than the session owner (see "Cluster routing" below) - Optional resumability: when an
:event_storeis configured, messages on a session's standalone stream are recorded with monotonic ids and replayed on reconnect viaLast-Event-ID. Messages fired while no handler is attached are still recorded (bounded by:stream_grace), so they survive reconnect gaps. SeeAnubis.Server.Transport.StreamableHTTP.EventStore.
Cluster routing
Each transport registers its SSE handlers in a node-local map AND in a :pg
group (scope derived from the transport name) keyed by session id. When
route_to_session/3 finds no local handler, it falls back to the :pg
group and delivers {:sse_message, message} directly to the remote handler
pid, mirroring how Anubis.Server.Registry.PG routes inbound requests.
This removes the sticky-session requirement for server-to-client traffic:
the SSE GET may terminate on any node, regardless of where the session
process lives. Handler pids are monitored by :pg, so crashed or
disconnected handlers disappear from the group automatically.
Notes:
- The scope is derived from the transport's registered name, so cluster
routing requires an atom name (the default naming used by
Anubis.Server.Supervisor). Transports registered with a:viaor:globalname keep node-local delivery only. - The broadcast path (
send_message/3without a:session_id) remains node-local. - Resumability across nodes additionally requires a shared
:event_store; with the default node-local store, events emitted while the handler lives on another node are delivered live but not recorded.
Usage
StreamableHTTP is typically started through the server supervisor:
Anubis.Server.start_link(MyServer, [],
transport: :streamable_http,
streamable_http: [port: 4000]
)For integration with existing Phoenix/Plug applications:
# In your router
forward "/mcp", Anubis.Server.Transport.StreamableHTTP.Plug,
server: MyApp.MCPServerSummary
Functions
Returns a specification to start this module under a supervisor.
Closes a session's resumable stream and drops its recorded events. Called on
DELETE (explicit session termination). No-op when resumability is disabled.
Returns the SSE handler pid for a session, or nil if none is connected.
Returns the number of connected SSE handlers.
Returns the number of connected SSE handlers whose metadata satisfies selector.
Registers the calling process as the SSE handler for a session.
Registers the calling process as the SSE handler for a session, attaching an
opaque metadata map.
Returns the resumability config for this transport as {event_store, retry},
where event_store is {module, name} or nil and retry is the SSE
retry: value in milliseconds or nil. Read by the Plug when opening an SSE
stream.
Routes a message to a specific session's SSE handler for server-to-client push.
Sends a message to every connected SSE handler whose metadata satisfies selector.
Unregisters the SSE handler for a session. Called when the SSE connection closes.
Types
@type option() :: {:server, GenServer.server()} | {:name, GenServer.name()} | GenServer.option()
StreamableHTTP transport options
:server- The server module (required):name- Name for registering the GenServer (required)
@type t() :: GenServer.server()
Functions
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec close_session_stream(GenServer.server(), String.t()) :: :ok | {:error, term()}
Closes a session's resumable stream and drops its recorded events. Called on
DELETE (explicit session termination). No-op when resumability is disabled.
Fails closed: when the event store cannot delete the recorded events the
error is returned and the stream stays open in transport state, so a retried
DELETE can complete the cleanup.
@spec get_sse_handler(GenServer.server(), String.t()) :: pid() | nil
Returns the SSE handler pid for a session, or nil if none is connected.
@spec handler_count(GenServer.server()) :: non_neg_integer()
Returns the number of connected SSE handlers.
@spec handler_count(GenServer.server(), (map() -> as_boolean(term()))) :: non_neg_integer()
Returns the number of connected SSE handlers whose metadata satisfies selector.
selector receives each handler's opaque metadata map (see
register_sse_handler/3) and returns a truthy value to count that handler.
@spec register_sse_handler(GenServer.server(), String.t()) :: :ok | {:error, term()}
Registers the calling process as the SSE handler for a session.
Called by the Plug when establishing an SSE connection. Equivalent to
register_sse_handler/3 with empty metadata.
@spec register_sse_handler(GenServer.server(), String.t(), map()) :: :ok | {:error, term()}
Registers the calling process as the SSE handler for a session, attaching an
opaque metadata map.
The transport stores metadata verbatim and never interprets it. Hosts use it
to tag a subscriber with application-defined attributes (tenant, user, feature
scope, ...) so later send_message_to_subscribers/4 and handler_count/2 calls
can select on them. The Plug populates it from its :subscriber_metadata
callback; direct callers may pass any map.
@spec resumability_config(GenServer.server()) :: {term() | nil, non_neg_integer() | nil}
Returns the resumability config for this transport as {event_store, retry},
where event_store is {module, name} or nil and retry is the SSE
retry: value in milliseconds or nil. Read by the Plug when opening an SSE
stream.
@spec route_to_session(GenServer.server(), String.t(), binary()) :: :ok | {:error, term()}
Routes a message to a specific session's SSE handler for server-to-client push.
When resumability is enabled the message is also recorded on the session's
stream so it can be replayed on reconnect. The message is recorded even if no
handler is currently attached, in which case :ok is still returned.
@spec send_message_to_subscribers( GenServer.server(), (map() -> as_boolean(term())), binary(), keyword() ) :: :ok | {:error, term()}
Sends a message to every connected SSE handler whose metadata satisfies selector.
selector receives each handler's opaque metadata map (see
register_sse_handler/3) and returns a truthy value for the subscribers that
should receive message. This complements route_to_session/3 (a single
session) and send_message/3 (broadcast to all handlers) with delivery to an
arbitrary, application-defined subset.
opts accepts :timeout (default 5000).
@spec unregister_sse_handler(GenServer.server(), String.t(), pid() | nil) :: :ok
Unregisters the SSE handler for a session. Called when the SSE connection closes.