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