Server-side SSE session manager for bidirectional MCP communication.
Manages the lifecycle of SSE sessions where the server needs to send requests to the client (elicitation, sampling) and receive responses.
Architecture
Each MCP session has:
- A POST endpoint for client→server requests
- A GET SSE stream for server→client messages (notifications, requests)
- A pending request tracker for correlating server requests with client responses
Usage
# Initialize session state (call once at server startup)
SSESession.init()
# Register a GET SSE stream for a session
SSESession.register_sse_stream(session_id)
# Send a request to the client and wait for response
{:ok, result} = SSESession.send_request(session_id, "elicitation/create", params)
# Route a client response to the waiting request handler
SSESession.handle_response(request_id, {:ok, result})Summary
Functions
Wait until exactly one live SSE stream exists, then return its session id.
Wait until a live SSE stream is registered for session_id.
Clean up session state for an SSE stream registration.
Route a client response back to the waiting request handler.
Check if a session has an active (live) SSE stream.
Initialize the session ETS table.
Register the calling process as the SSE stream for a session.
Run the SSE event loop for a GET connection.
Send a JSON-RPC request to the client via the GET SSE stream.
If exactly one live SSE stream is registered, return its session id.
Functions
@spec await_sole_live_session_id(non_neg_integer()) :: {:ok, String.t()} | {:error, :timeout | :ambiguous}
Wait until exactly one live SSE stream exists, then return its session id.
Handles the race where tools/call POST arrives slightly before the client's
GET SSE stream is registered (and the POST may omit mcp-session-id).
@spec await_sse_stream(String.t(), non_neg_integer()) :: :ok | {:error, :timeout}
Wait until a live SSE stream is registered for session_id.
Returns :ok when ready, or {:error, :timeout} if not registered within
timeout_ms (default 3000ms).
@spec cleanup(String.t()) :: :ok
Clean up session state for an SSE stream registration.
Route a client response back to the waiting request handler.
Called when the server receives a POST with a JSON-RPC response
(has id + result/error, no method).
Accepts integer or string request ids (JSON number vs string echo).
Check if a session has an active (live) SSE stream.
@spec init() :: :ok
Initialize the session ETS table.
@spec register_sse_stream(String.t()) :: :ok
Register the calling process as the SSE stream for a session.
Replaces any previous registration for the same session id (including dead pids).
@spec run_sse_loop(Plug.Conn.t(), String.t(), keyword()) :: Plug.Conn.t()
Run the SSE event loop for a GET connection.
Forwards {:sse_send, data} messages as SSE events.
Returns when :sse_close is received or timeout expires.
Always clears this process's registration for session_id on exit.
Send a JSON-RPC request to the client via the GET SSE stream.
Options:
:timeout— wait for client response (default 10000ms):wait_for_stream— wait for SSE registration before send (default 3000ms)
Blocks until the client responds or timeout is reached.
Returns {:ok, result} or {:error, reason}.
@spec sole_live_session_id() :: {:ok, String.t()} | :error
If exactly one live SSE stream is registered, return its session id.
Used as a last-resort session resolution for bidirectional tools when the
client omits mcp-session-id on tools/call.