How to actually build session persistence in streamable http MCP server?

[karthich]

I have a question here regarding how to build session persistence into an MCP server.

I am running my mcp server (FastMCP) in an ecs service with multiple tasks. I have built a redis event store to allow for resumability. The service sits behind an ALB that directs HTTPS traffic to it. Inspite of all of it, Im still observing intermittent Bad Request: No valid session ID provided 400 errors when I connect the mcpinspector.

My suspicion is StreamableHTTPSessionManager doesn't seem to be able to store session-id data in an external data source. Therefore traffic from a client has to always hit the same task since the session id is tied to a server.

In an AWS architecture, how do I configure my server and client to accomplish this. Is using sticky sessions the only route here? Are streamable http sessions not horizontally scalable?

[lorenzocesconetto]

Hi @karthich,

The sessions are currently stored in memory by the SDK. This is a limitation of this implementation, not a limitation of the protocol itself.

For a quick fix, if your use case permits, you could run the server in stateless mode.

If you're interested in knowing how this is implemented internally:

• This dictionary stores the sessions
• Then it's used to check incoming session ids here.

One way to fix this: SDK should (IMHO) allow the user to provide a custom MemoryManager class which implements some MemoryManagerInterface (methods could be something like this: storeSession, retrieveSession, deleteSession). Then users of the sdk could provide their own implementation of the interface, for instance storing it in some DB. I could work on this over the weekend if some core maintainers agree it's a good idea (cc: @ihrpr @jerome3o-anthropic ).

Disclaimer: I am not a core maintainer, but found this issue to be interesting.

[karthich]

Yes. I was hoping there was something that would let me do that but the implementation is pretty strong about what it is and there's no good way to communicate to the SDK to use a different way to store the sessions. I know its not part of this project but I was able to implement a Redis backed implementation of the EventStore class but alas no such thing for managing the session themselves.

[yadavnikhil]

I'm using sampling which only works in stateful mode as i noticed.
Session persistence is a new need now.

+1 for this

[yadavnikhil]

As a temporary WA and checks, I did successfully make multi-workers for fastapi uvicorn to persist session id to redis by overriding the _handle_stateful_request() method in StreamableHTTPSessionManager by creating a custom class and creating a new session with same session id and save to dictionary if request goes to new worker, though duplicating server_instances dict in different workers this seems to work for tools, prompts and resources at client, but sampling still didn't work.

Override class:

class PersistentSessionManager(StreamableHTTPSessionManager):
    """
    StreamableHTTPSessionManager with session ID persistence in Redis.
    """

    def __init__(
        self, *args, redis_url="redis://localhost:6379/0", **kwargs
    ):
        super().__init__(*args, **kwargs)
        self.redis = aioredis.from_url(redis_url, decode_responses=True)

    async def _create_and_start_session(
        self, session_id: str
    ) -> StreamableHTTPServerTransport:
        http_transport = StreamableHTTPServerTransport(
            mcp_session_id=session_id,
            is_json_response_enabled=self.json_response,
            event_store=self.event_store,
        )
        assert http_transport.mcp_session_id is not None
        self._server_instances[http_transport.mcp_session_id] = (
            http_transport
        )

        async def run_server(
            *,
            task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORED,
        ):
            async with http_transport.connect() as streams:
                read_stream, write_stream = streams
                task_status.started()
                await self.app.run(
                    read_stream,
                    write_stream,
                    self.app.create_initialization_options(),
                    stateless=False,  # Stateful mode
                )

        assert self._task_group is not None
        await self._task_group.start(run_server)
        return http_transport

    async def _handle_stateful_request(
        self,
        scope,
        receive,
        send,
    ):
        request = Request(scope, receive)
        request_mcp_session_id = request.headers.get(MCP_SESSION_ID_HEADER)

        # Try in-memory first
        if (
            request_mcp_session_id is not None
            and request_mcp_session_id in self._server_instances
        ):
            transport = self._server_instances[request_mcp_session_id]
            LOGGER.debug(
                "Session found in memory, handling request directly"
            )
            await transport.handle_request(scope, receive, send)
            return

        # Try Redis for session persistence
        if request_mcp_session_id is not None:
            session_exists = await self.redis.exists(
                f"mcp_session:{request_mcp_session_id}"
            )
            if session_exists:
                LOGGER.info(
                    f"Session {request_mcp_session_id} found in Redis (not in memory), recreating transport."
                )
                async with self._session_creation_lock:
                    http_transport = await self._create_and_start_session(
                        request_mcp_session_id
                    )
                    await http_transport.handle_request(
                        scope, receive, send
                    )
                    return

        # If no session ID provided, create a new session and save to Redis
        if request_mcp_session_id is None:
            async with self._session_creation_lock:
                new_session_id = uuid4().hex
                http_transport = await self._create_and_start_session(
                    new_session_id
                )
                await self.save_session_id(new_session_id)
                LOGGER.info(
                    f"Created new session and saved to Redis: {new_session_id}"
                )
                await http_transport.handle_request(scope, receive, send)
                return

        # Invalid session ID (not in memory, not in Redis)
        response = Response(
            "Bad Request: No valid session ID provided",
            status_code=400,
        )
        await response(scope, receive, send)

    async def save_session_id(self, session_id: str):
        """
        Save the session ID to Redis.
        """
        await self.redis.set(f"mcp_session:{session_id}", "1", ex=360)

But sampling now get stuck at with error:

Request stream _GET_stream not found for message. Still processing message as the client might reconnect and replay.

This seems may not be simple implementation of persisting just session id in redis and recreating session with same id in different workers may not work for sampling atleast OR may be some fix needed for sampling to work.

[qwertyuiop17u]

I've identified that StreamableHTTPSessionManager supports custom EventStore configurations, which opens up possibilities for implementing persistence solutions.

Proposed Solution

We can develop a custom EventStore implementation to handle session persistence. I've begun exploring a MongoDB-based approach as a reference implementation.

class MongoEventStore(EventStore):
    """
    MongoEventStore provides a MongoDB-based implementation of the EventStore interface for MCP session resumability.
    By implementing the EventStore interface, MCP can support various types of event stores for session persistence.
    To use a different backend, simply implement the EventStore interface accordingly.

    This class handles storing and replaying events for a given stream using MongoDB as the backend.
    It ensures connection management, event serialization, and efficient querying via indexes.
    """
    def __init__(
        self,
        connection_string: str = "mongodb://localhost:27017",
        database_name: str = "mcp_sessions",
        collection_name: str = "events"
    ):
        self.connection_string = connection_string
        self.database_name = database_name
        self.collection_name = collection_name
        self._client: Optional[AsyncIOMotorClient] = None
        self._collection: Optional[AsyncIOMotorCollection] = None
        self._event_counter = 0

    async def _ensure_connection(self):
        """Ensure MongoDB connection is established."""
        if self._client is None:
            self._client = AsyncIOMotorClient(self.connection_string)
            self._collection = self._client[self.database_name][self.collection_name]

            # Create indexes to optimize query performance
            await self._collection.create_index([
                ("stream_id", 1),
                ("event_id", 1)
            ])
            await self._collection.create_index([
                ("stream_id", 1),
                ("timestamp", 1)
            ])

    async def store_event(
        self,
        stream_id: StreamId,
        message: JSONRPCMessage
    ) -> EventId:
        """Store an event into MongoDB."""
        await self._ensure_connection()

        # Generate a unique event ID
        event_id = str(ObjectId())

        # Serialize JSONRPCMessage
        message_dict = message.model_dump(by_alias=True, exclude_none=True)

        # Create event document
        event_doc = {
            "_id": ObjectId(event_id),
            "stream_id": stream_id,
            "event_id": event_id,
            "message": message_dict,
            "timestamp": datetime.utcnow(),
            "message_type": message.root.__class__.__name__
        }

        # Insert into MongoDB
        await self._collection.insert_one(event_doc)

        return event_id

    async def replay_events_after(
        self,
        last_event_id: EventId,
        send_callback: EventCallback,
    ) -> StreamId | None:
        """Replay events after the specified event ID."""
        await self._ensure_connection()

        # Find the timestamp of the last event
        last_event = await self._collection.find_one(
            {"event_id": last_event_id}
        )

        if last_event:
            # Replay events after this timestamp
            query = {
                "stream_id": last_event["stream_id"],
                "timestamp": {"$gt": last_event["timestamp"]}
            }
        else:
            # If event ID not found, cannot replay
            # Need stream_id to replay from the beginning; return None for now
            return None

        # Query events in chronological order
        cursor = self._collection.find(query).sort("timestamp", 1)

        stream_id = None
        async for event_doc in cursor:
            # Reconstruct JSONRPCMessage
            message_dict = event_doc["message"]
            message = JSONRPCMessage.model_validate(message_dict)

            # Send event
            event_message = EventMessage(message, event_doc["event_id"])
            await send_callback(event_message)

            # Record stream_id
            if stream_id is None:
                stream_id = event_doc["stream_id"]

        return stream_id

    async def close(self):
        """Close MongoDB connection."""
        if self._client:
            self._client.close()

Current Status

• With the MongoEventStore, I've written each session to MongoDB

• For the session resuming, the implementation is currently blocked by an upstream issue in the Python SDK:
  Issue: Resumption of streamable HTTP session has potential for deadlock #860

It causes deadlock when resumes session, so I can not test the session resuming from Mongo.

Next Steps

Priority: I would try to resolve the session resume functionality first
Follow-up: Complete the MongoEventStore implementation and testing once the blocking issue is resolved

[karthich]

@qwertyuiop17u I built an event store backed by Redis as well. This gives me the ability to resume events across multiple app instances. However I cannot resume those events due to the in-memory storage of the sessions themselves. I'm really curious to hear what the maintainers think we should do here. It seems like stateless MCP server is the only option for resilient deployments.

[ranpariyachetan]

@yadavnikhil It is a good attempt creating custom implementation of StreamableHTTPSessionManager with redis as backend.

Creating a new plain object of StreamableHTTPServerTransport may not work as it seems _request_streams dictionary in StreamableHTTPServerTransport has significance on handling and process request.

The new object of StreamableHTTPServerTransport clearly has an empty dictionary for _request_streams and that's why you see the error

Request stream _GET_stream not found for message. Still processing message as the client might reconnect and replay.

Hence I believe, not only "1" but also the state of StreamableHTTPServerTransport which is represented by _request_streams should be stored in redis and re-assigned to the new object of StreamableHTTPServerTransport.

And that leads to the fact that _server_instances will not be needed any more. Each request can re-create the transport object with the state retrieved from redis, handle the request, store the updated state back to redis, and dispose the transport object.

So in the end it is not just a simple implementation of persisting just session id in redis and recreating session with same id in different workers.

[karthich]

Correct. Do the maintainers have a plan to introduce the ability to store the session object in a backend at all? I have not seen them chime in yet.