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
- 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.
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.