FastMCP: State and Memory Management
Relevant source files
The following files were used as context for generating this wiki page:
State and memory management within the python-sdk primarily revolves around two key areas: providing in-memory communication channels for testing purposes and managing event history for client session resumability. The core of the testing framework relies on an in-memory session utility that bypasses network transports, allowing for direct, efficient, and reliable testing of FastMCP server logic.
This document outlines the architecture and usage of the in-memory session management for testing and touches upon the InMemoryEventStore used for stateful session resumption in the StreamableHTTP transport.
In-Memory Session for Testing
The primary mechanism for testing FastMCP servers is the create_connected_server_and_client_session utility. It establishes a direct, in-memory connection between a server instance and a client session, which is ideal for unit and integration tests as it eliminates network latency and potential transport-layer issues.
Sources: src/mcp/shared/memory.py, tests/server/fastmcp/test_server.py
Core Components
| Component | Description | Source File |
|---|---|---|
create_connected_server_and_client_session | An async context manager that sets up a fully connected ClientSession to a Server instance using in-memory object streams. This is the main entry point for testing. | src/mcp/shared/memory.py:16-43 |
anyio.create_memory_object_stream | The underlying anyio utility used to create pairs of in-memory send/receive streams that act as the communication channel between the server and client sessions. | src/mcp/shared/memory.py:27 |
ServerSession | The server-side session handler that is spawned in a background task to process incoming requests from the client stream. | src/mcp/shared/memory.py:32 |
ClientSession | The client-side session object that is yielded by the context manager. Test code uses this object to interact with the server. | src/mcp/shared/memory.py:42 |
Connection Flow
The following diagram illustrates the setup process when create_connected_server_and_client_session is invoked.
This diagram shows how the context manager orchestrates the creation of memory streams and connects the server and client sessions for testing.
Sources: src/mcp/shared/memory.py:16-43
Usage in Tests
This utility is used extensively throughout the project's test suite. The common pattern is to instantiate a FastMCP server, and then wrap a call to the utility in an async with block to get a connected client.
Here is a typical usage example for testing a tool:
# tests/server/fastmcp/test_server.py:202-208
@pytest.mark.anyio
async def test_call_tool(self):
mcp = FastMCP()
mcp.add_tool(tool_fn)
async with client_session(mcp._mcp_server) as client:
result = await client.call_tool("my_tool", {"arg1": "value"})
assert not hasattr(result, "error")
assert len(result.content) > 0
In this snippet, client_session is an alias for create_connected_server_and_client_session. The test creates a server, adds a tool, and then immediately gets a client to test the tool's functionality without any network setup.
Sources: tests/server/fastmcp/test_server.py:202-208, src/mcp/shared/memory.py:12
A similar pattern is used for testing resources:
# tests/server/fastmcp/servers/test_file_server.py:82-89
@pytest.mark.anyio
async def test_read_resource_dir(mcp: FastMCP):
res_iter = await mcp.read_resource("dir://test_dir")
res_list = list(res_iter)
assert len(res_list) == 1
res = res_list[0]
assert res.mime_type == "text/plain"
This demonstrates how the in-memory connection facilitates testing of resource-providing functions.
Sources: tests/server/fastmcp/servers/test_file_server.py:82-89
State Management for Resumability
A different form of in-memory state management is used in the StreamableHTTP transport to support client session resumability.
InMemoryEventStore
The simple-streamablehttp server example utilizes an InMemoryEventStore. This component's purpose is to store events generated during a client's session in memory. If a client disconnects and then reconnects with a Last-Event-ID header, the server can use the InMemoryEventStore to replay the missed events, allowing the client to resume its session from where it left off.
Key features of this approach:
- Event Storage: Events are stored in memory.
- Event Replay: Replays events to reconnecting clients.
- Session Resumption: Enables clients to recover from network interruptions without losing state.
This implementation is intended for demonstration purposes. A production system would likely require a persistent storage solution instead of an in-memory one.
Sources: examples/servers/simple-streamablehttp/README.md
Summary
The python-sdk employs in-memory data structures and utilities for two distinct but important purposes. The create_connected_server_and_client_session function is a cornerstone of the testing strategy, providing a fast and reliable way to test FastMCP server logic by simulating a client-server connection entirely in memory. Separately, the concept of an InMemoryEventStore demonstrates how state can be managed in memory to provide advanced features like session resumability for specific transports like StreamableHTTP. Both highlight a pragmatic approach to state and memory management tailored to the specific needs of testing and feature implementation.