Session management for streamable HTTP connections.
This module provides session management capabilities for MCP servers using streamable HTTP transports like 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: New sessions are created when SSE connections are established
- Event Storage: Events are buffered with unique IDs for potential replay
- Session Resumption: Clients can reconnect using Last-Event-ID header
- Session Termination: Sessions are terminated on explicit DELETE or timeout
Configuration
:max_events_per_session- Maximum events to buffer per session (default: 1000):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
Returns a specification to start this module under a supervisor.
Creates a new session with the given metadata.
Ensures a transport-issued session ID exists and 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_events_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(), status: :active | :terminated }
@type session_id() :: String.t()
Functions
Returns a specification to start this module under a supervisor.
See Supervisor.
@spec create_session(map()) :: session_id()
Creates a new session with the given metadata.
Returns a unique session ID that can be used for subsequent operations.
@spec ensure_session(session_id(), map()) :: :ok
Ensures a transport-issued session ID exists and refreshes its activity.
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 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}
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 SSE connections are explicitly closed or when DELETE requests are received.
@spec update_session(session_id(), map()) :: :ok | {:error, :session_not_found}
Updates session metadata or activity timestamp.