Legacy session management for Streamable HTTP connections.
This module supports the initialize-based MCP 2025-03-26 through
2025-11-25 transport lifecycle. MCP 2026-07-28 HTTP is stateless and does
not use this manager, Mcp-Session-Id, Last-Event-ID, GET streams, or
DELETE termination.
For legacy connections, this module provides session management for MCP servers using Streamable HTTP and Server-Sent Events (SSE). It handles session lifecycle, event buffering, and session resumption through Last-Event-ID support.
Features
- Session lifecycle management (create, update, terminate)
- Event buffering and replay for connection resumption
- Last-Event-ID support for seamless reconnection
- Session expiration and cleanup
- Memory-efficient event storage with configurable limits
- Session health monitoring and metrics
Session Lifecycle
- Session Creation: Legacy Streamable HTTP creates a server-issued ID
only for
initialize; deprecated HTTP+SSE retains its endpoint handshake. - Initialization: Exactly one monitored request may initialize a session; success immutably binds the negotiated version and failure terminates it.
- Event Storage: Events are buffered with unique IDs for potential replay.
- Session Resumption: Initialized clients reconnect using the issued ID
and may resume events with
Last-Event-ID. - Session Termination: Sessions terminate on explicit DELETE, abandoned initialization, or timeout.
Configuration
:max_events_per_session- Maximum events to buffer per session (default: 1000):max_sessions- Maximum active/retained sessions (default: 10,000):max_request_ids- Maximum claimed JSON-RPC request IDs per session (default: 10,000):max_event_bytes- Maximum JSON-encoded bytes for one replay event (default: 1 MiB):max_replay_bytes_per_session- Maximum JSON-encoded replay bytes retained per session (default: 8 MiB):session_ttl_seconds- Session TTL in seconds (default: 3600):cleanup_interval_ms- Cleanup interval in milliseconds (default: 60000):storage_backend- Storage backend (:etsor:persistent_term, default::ets)
Usage
# Start the session manager
{:ok, _pid} = ExMCP.SessionManager.start_link([])
# Create a new session
session_id = ExMCP.SessionManager.create_session(%{
transport: :sse,
client_info: %{user_agent: "my-client/1.0"}
})
# Store an event
ExMCP.SessionManager.store_event(session_id, %{
id: "event-123",
type: "notification",
data: %{message: "Hello"},
timestamp: System.system_time(:microsecond)
})
# Replay events after a specific event ID
events = ExMCP.SessionManager.replay_events_after(session_id, "event-122")
# Terminate session
ExMCP.SessionManager.terminate_session(session_id)Summary
Functions
Atomically appends an event using a store-owned, monotonically increasing ID.
Returns a specification to start this module under a supervisor.
Atomically claims the one initialization attempt allowed for an active
legacy session. This prevents concurrent or repeated initialize requests
from racing into application handlers.
Atomically claims a JSON-RPC request ID for an active session.
Completes a previously claimed initialization and immutably binds its negotiated protocol version to the session.
Creates a new session with the given metadata.
Atomically verifies that a session is active, identity-bound to the caller, and has completed initialization, then refreshes its activity.
Verifies that a server-issued session ID exists, is active, and is bound to the same authorization identity, then refreshes its activity.
Gets session information.
Gets session statistics.
Lists all active sessions.
Replays events for a session after the given event ID.
Replays events for a session after the given event ID and sends them to the handler.
Starts the session manager with optional configuration.
Stores an event for the given session.
Terminates a session and cleans up its events.
Updates session metadata or activity timestamp.
Types
@type config() :: %{ max_sessions: pos_integer(), max_request_ids: pos_integer(), max_events_per_session: pos_integer(), max_event_bytes: pos_integer(), max_replay_bytes_per_session: pos_integer(), session_ttl_seconds: pos_integer(), cleanup_interval_ms: pos_integer(), storage_backend: :ets | :persistent_term }
@type event_data() :: %{ id: event_id(), session_id: session_id(), type: String.t(), data: term(), timestamp: integer() }
@type event_id() :: String.t()
@type session_data() :: %{ id: session_id(), transport: atom(), client_info: map(), created_at: integer(), last_activity: integer(), event_count: non_neg_integer(), replay_bytes: non_neg_integer(), request_id_count: non_neg_integer(), status: :active | :terminated, initialized: boolean(), initialization_claimed: boolean(), protocol_version: String.t() | nil, principal_id: String.t() | nil, tenant_id: String.t() | nil, issuer: String.t() | nil, audience: String.t() | [String.t()] | nil }
@type session_id() :: String.t()
Functions
@spec append_event(session_id(), String.t(), term()) :: {:ok, event_data()} | {:error, :session_not_found | :event_too_large | :event_not_json_encodable}
Atomically appends an event using a store-owned, monotonically increasing ID.
Legacy SSE delivery uses this function before writing to the connection so events remain replayable when the write races a disconnect. The returned event is the public representation retained by the session manager.
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec claim_initialization(session_id()) :: :ok | {:error, :session_not_found | :session_already_initialized | :initialization_in_progress}
Atomically claims the one initialization attempt allowed for an active
legacy session. This prevents concurrent or repeated initialize requests
from racing into application handlers.
@spec claim_request_id(session_id(), String.t() | integer()) :: :ok | {:error, :session_not_found | :duplicate_request_id | :request_id_limit_exceeded}
Atomically claims a JSON-RPC request ID for an active session.
IDs remain claimed for the session lifetime. The configured
:max_request_ids bound fails closed once reached.
@spec complete_initialization(session_id(), String.t()) :: :ok | {:error, :session_not_found | :initialization_not_claimed | :initialization_owner_mismatch | :session_protocol_version_mismatch}
Completes a previously claimed initialization and immutably binds its negotiated protocol version to the session.
@spec create_session(map()) :: session_id() | {:error, :session_limit_exceeded}
Creates a new session with the given metadata.
Returns a unique session ID that can be used for subsequent operations.
@spec ensure_initialized_session(session_id(), map()) :: :ok | {:error, :session_not_found | :session_identity_mismatch | :session_not_initialized}
Atomically verifies that a session is active, identity-bound to the caller, and has completed initialization, then refreshes its activity.
@spec ensure_session(session_id(), map()) :: :ok | {:error, :session_not_found | :session_identity_mismatch}
Verifies that a server-issued session ID exists, is active, and is bound to the same authorization identity, then refreshes its activity.
Legacy Streamable HTTP issues the ID on POST before the client opens its SSE channel. Retaining that exact ID associates later subscriptions and notifications with one client across both channels.
@spec get_session(session_id()) :: {:ok, session_data()} | {:error, :session_not_found}
Gets session information.
@spec get_stats() :: %{ total_sessions: non_neg_integer(), active_sessions: non_neg_integer(), total_events: non_neg_integer(), memory_usage: non_neg_integer() }
Gets session statistics.
@spec list_sessions() :: [session_data()]
Lists all active sessions.
@spec replay_events_after(session_id(), event_id() | nil) :: [event_data()] | {:error, :session_not_found}
Replays events for a session after the given event ID.
This is used when legacy clients reconnect with a Last-Event-ID header to resume from where they left off.
@spec replay_events_after(session_id(), event_id() | nil, pid()) :: :ok | {:error, :session_not_found}
Replays events for a session after the given event ID and sends them to the handler.
This is the callback function referenced in SSEHandler for session replay.
@spec start_link(keyword()) :: GenServer.on_start()
Starts the session manager with optional configuration.
@spec store_event(session_id(), event_data()) :: :ok | {:error, :session_not_found | :event_too_large | :event_not_json_encodable}
Stores an event for the given session.
Events are stored with their ID, type, data, and timestamp for potential replay during session resumption.
@spec terminate_session(session_id()) :: :ok
Terminates a session and cleans up its events.
This should be called when a session is explicitly deleted or permanently abandoned. A transient SSE disconnect alone does not terminate the session, because its events must remain available for Last-Event-ID replay.
@spec update_session(session_id(), map()) :: :ok | {:error, :session_not_found | :session_identity_mismatch | :session_protocol_version_mismatch}
Updates session metadata or activity timestamp.