Add session-scoped state persistence

docs/development/upgrade-guide.mdx

@@ -256,6 +256,50 @@ The environment variable for controlling the server banner has been renamed:
 
 This change reflects that the setting now applies to all server startup methods, not just the CLI. The banner is now suppressed when running `python server.py` directly, not just when using `fastmcp run`.
 
+### Context State Methods Are Async
+
+<Warning>
+**Breaking Change:** `ctx.set_state()` and `ctx.get_state()` are now async methods. Synchronous calls will fail.
+</Warning>
+
+Context state has changed from request-scoped to session-scoped, persisting across multiple tool calls within the same MCP session. The methods are now async because they interact with a pluggable storage backend.
+
+<CodeGroup>
+```python Before
+@mcp.tool
+def my_tool(ctx: Context) -> str:
+    ctx.set_state("key", "value")
+    value = ctx.get_state("key")
+    return value
+```
+
+```python After
+@mcp.tool
+async def my_tool(ctx: Context) -> str:
+    await ctx.set_state("key", "value")
+    value = await ctx.get_state("key")
+    return value
+```
+</CodeGroup>
+
+**What changed:**
+- State now persists across requests within a session (not just within a single request)
+- Different clients have isolated state (keyed by session ID)
+- State expires after 1 day to prevent unbounded memory growth
+- New method: `await ctx.delete_state(key)`
+
+**Custom storage backends:**
+
+By default, state uses an in-memory store. For distributed deployments, provide a custom backend:
+
+```python
+from key_value.aio.stores.redis import RedisStore
+
+mcp = FastMCP("server", session_state_store=RedisStore(...))
+```
+
+See [Session State](/servers/context#session-state) for full documentation.
+
 ## v2.14.0
 
 ### OpenAPI Parser Promotion

docs/development/v3-notes/v3-features.mdx

@@ -150,6 +150,37 @@ Documentation: `docs/servers/providers/transforms.mdx`, `docs/servers/visibility
 
 ---
 
+## Session-Scoped State
+
+v3.0 changes context state from request-scoped to session-scoped. State now persists across multiple tool calls within the same MCP session.
+
+```python
+@mcp.tool
+async def increment_counter(ctx: Context) -> int:
+    count = await ctx.get_state("counter") or 0
+    await ctx.set_state("counter", count + 1)
+    return count + 1
+```
+
+State is automatically keyed by session ID, ensuring isolation between different clients. The implementation uses [pykeyvalue](https://github.com/strawgate/py-key-value) for pluggable storage backends:
+
+```python
+from key_value.aio.stores.redis import RedisStore
+
+# Use Redis for distributed deployments
+mcp = FastMCP("server", session_state_store=RedisStore(...))
+```
+
+**Key details:**
+- Methods are now async: `await ctx.get_state()`, `await ctx.set_state()`, `await ctx.delete_state()`
+- State expires after 1 day (TTL) to prevent unbounded memory growth
+- Works during `on_initialize` middleware when using the same session object
+- For distributed HTTP, session identity comes from the `mcp-session-id` header
+
+Documentation: `docs/servers/context.mdx`
+
+---
+
 ## Visibility System
 
 Components can be dynamically enabled/disabled at runtime using the visibility system (`src/fastmcp/server/transforms/visibility.py`).
@@ -537,3 +568,19 @@ See `docs/development/v3-notes/auth-provider-env-vars.mdx` for rationale.
 `FASTMCP_SHOW_CLI_BANNER` → `FASTMCP_SHOW_SERVER_BANNER`
 
 Now applies to all server startup methods, not just the CLI.
+
+### Context State Methods Are Async
+
+`ctx.set_state()` and `ctx.get_state()` are now async and session-scoped:
+
+```python
+# v2.x
+ctx.set_state("key", "value")
+value = ctx.get_state("key")
+
+# v3.0
+await ctx.set_state("key", "value")
+value = await ctx.get_state("key")
+```
+
+State now persists across requests within a session. See "Session-Scoped State" above.

docs/servers/context.mdx

@@ -22,7 +22,7 @@ The `Context` object provides a clean interface to access MCP features within yo
 - **Prompt Access**: List and retrieve prompts registered with the server
 - **LLM Sampling**: Request the client's LLM to generate text based on provided messages
 - **User Elicitation**: Request structured input from users during tool execution
-- **State Management**: Store and share data between middleware and the handler within a single request
+- **Session State**: Store data that persists across requests within an MCP session
 - **Request Information**: Access metadata about the current request
 - **Server Access**: When needed, access the underlying FastMCP server instance
 
@@ -209,55 +209,57 @@ messages = result.messages
 - **`ctx.list_prompts() -> list[MCPPrompt]`**: Returns list of all available prompts
 - **`ctx.get_prompt(name: str, arguments: dict[str, Any] | None = None) -> GetPromptResult`**: Get a specific prompt with optional arguments
 
-### State Management
+### Session State
 
-<VersionBadge version="2.11.0" />
+<VersionBadge version="3.0.0" />
 
-Store and share data between middleware and handlers within a single MCP request. Each MCP request (such as calling a tool, reading a resource, listing tools, or listing resources) receives its own context object with isolated state. Context state is particularly useful for passing information from [middleware](/servers/middleware) to your handlers.
+Store data that persists across multiple requests within the same MCP session. Session state is automatically keyed by the client's session, ensuring isolation between different clients.
 
-To store a value in the context state, use `ctx.set_state(key, value)`. To retrieve a value, use `ctx.get_state(key)`.
+```python
+from fastmcp import FastMCP, Context
 
-<Warning>
-Context state is scoped to a single MCP request. Each operation (tool call, resource read, list operation, etc.) receives a new context object. State set during one request will not be available in subsequent requests. For persistent data storage across requests, use external storage mechanisms like databases, files, or in-memory caches.
-</Warning>
+mcp = FastMCP("stateful-app")
 
-This simplified example shows how to use MCP middleware to store user info in the context state, and how to access that state in a tool:
+@mcp.tool
+async def increment_counter(ctx: Context) -> int:
+    """Increment a counter that persists across tool calls."""
+    count = await ctx.get_state("counter") or 0
+    await ctx.set_state("counter", count + 1)
+    return count + 1
 
-```python {7-8, 16-17}
-from fastmcp.server.middleware import Middleware, MiddlewareContext
+@mcp.tool
+async def get_counter(ctx: Context) -> int:
+    """Get the current counter value."""
+    return await ctx.get_state("counter") or 0
+```
 
-class UserAuthMiddleware(Middleware):
-    async def on_call_tool(self, context: MiddlewareContext, call_next):
+Each client session has its own isolated state—two different clients calling `increment_counter` will each have their own counter.
 
-        # Middleware stores user info in context state
-        context.fastmcp_context.set_state("user_id", "user_123")
-        context.fastmcp_context.set_state("permissions", ["read", "write"])
+**Method signatures:**
+- **`await ctx.set_state(key: str, value: Any) -> None`**: Store a value in session state
+- **`await ctx.get_state(key: str) -> Any`**: Retrieve a value (returns None if not found)
+- **`await ctx.delete_state(key: str) -> None`**: Remove a value from session state
 
-        return await call_next(context)
+<Note>
+State methods are async and require `await`. State expires after 1 day to prevent unbounded memory growth.
+</Note>
 
-@mcp.tool
-async def secure_operation(data: str, ctx: Context) -> str:
-    """Tool can access state set by middleware."""
+#### Custom Storage Backends
 
-    user_id = ctx.get_state("user_id")  # "user_123"
-    permissions = ctx.get_state("permissions")  # ["read", "write"]
-    
-    if "write" not in permissions:
-        return "Access denied"
-    
-    return f"Processing {data} for user {user_id}"
+By default, session state uses an in-memory store suitable for single-server deployments. For distributed or serverless deployments, provide a custom storage backend:
+
+```python
+from key_value.aio.stores.redis import RedisStore
+
+# Use Redis for distributed state
+mcp = FastMCP("distributed-app", session_state_store=RedisStore(...))
 ```
 
-**Method signatures:**
-- **`ctx.set_state(key: str, value: Any) -> None`**: Store a value in the context state
-- **`ctx.get_state(key: str) -> Any`**: Retrieve a value from the context state (returns None if not found)
+Any backend compatible with the [py-key-value-aio](https://github.com/strawgate/py-key-value) `AsyncKeyValue` protocol works. See [Storage Backends](/servers/storage-backends) for more options including Redis, DynamoDB, and MongoDB.
 
-**State Inheritance:**
-When a new context is created (nested contexts), it inherits a copy of its parent's state. This ensures that:
-- State set on a child context never affects the parent context
-- State set on a parent context after the child context is initialized is not propagated to the child context
+#### State During Initialization
 
-This makes state management predictable and prevents unexpected side effects between nested operations.
+State set during `on_initialize` middleware persists to subsequent tool calls when using the same session object (STDIO, SSE, single-server HTTP). For distributed/serverless HTTP deployments where different machines handle init and tool calls, state is isolated by the `mcp-session-id` header.
 
 ### Change Notifications
 
@@ -345,7 +347,7 @@ async def request_info(ctx: Context) -> dict:
 
 - **`ctx.request_id -> str`**: Get the unique ID for the current MCP request
 - **`ctx.client_id -> str | None`**: Get the ID of the client making the request, if provided during initialization
-- **`ctx.session_id -> str | None`**: Get the MCP session ID for session-based data sharing (HTTP transports only)
+- **`ctx.session_id -> str`**: Get the MCP session ID for session-based data sharing. Raises `RuntimeError` if the MCP session is not yet established.
 
 #### Request Context Availability
 

examples/persistent_state/README.md

@@ -0,0 +1,51 @@
+# Persistent Session State
+
+This example demonstrates session-scoped state that persists across tool calls within the same MCP session.
+
+## What it shows
+
+- State set in one tool call is readable in subsequent calls
+- Different clients have isolated state (same keys, different values)
+- Reconnecting creates a new session with fresh state
+
+## Running
+
+**HTTP transport:**
+
+```bash
+# Terminal 1: Start the server
+uv run python server.py
+
+# Terminal 2: Run the client
+uv run python client.py
+```
+
+**STDIO transport (in-process):**
+
+```bash
+uv run python client_stdio.py
+```
+
+## Example output
+
+```text
+Each line below is a separate tool call
+
+Alice connects
+  session a9f6eaa3
+  set user = Alice
+  set secret = alice-password
+  get user → Alice
+  get secret → alice-password
+
+Bob connects (different session)
+  session 0c3bffc5
+  get user → not found
+  get secret → not found
+  set user = Bob
+  get user → Bob
+
+Alice reconnects (new session)
+  session e39640e3
+  get user → not found
+```

examples/persistent_state/client.py

@@ -0,0 +1,85 @@
+"""Client for testing persistent state.
+
+Run the server first:
+    uv run python examples/persistent_state/server.py
+
+Then run this client:
+    uv run python examples/persistent_state/client.py
+"""
+
+import asyncio
+
+from rich.console import Console
+
+from fastmcp import Client
+from fastmcp.client.transports import StreamableHttpTransport
+
+URL = "http://127.0.0.1:8000/mcp"
+console = Console()
+
+
+async def main() -> None:
+    console.print()
+    console.print("[dim italic]Each line below is a separate tool call[/dim italic]")
+    console.print()
+
+    # --- Alice's session ---
+    console.print("[dim]Alice connects[/dim]")
+
+    transport1 = StreamableHttpTransport(url=URL)
+    async with Client(transport=transport1) as alice:
+        result = await alice.call_tool("list_session_info", {})
+        console.print(f"  session [cyan]{result.data['session_id'][:8]}[/cyan]")
+
+        await alice.call_tool("set_value", {"key": "user", "value": "Alice"})
+        console.print("  set [white]user[/white] = [green]Alice[/green]")
+
+        await alice.call_tool("set_value", {"key": "secret", "value": "alice-password"})
+        console.print("  set [white]secret[/white] = [green]alice-password[/green]")
+
+        result = await alice.call_tool("get_value", {"key": "user"})
+        console.print("  get [white]user[/white] → [green]Alice[/green]")
+
+        result = await alice.call_tool("get_value", {"key": "secret"})
+        console.print("  get [white]secret[/white] → [green]alice-password[/green]")
+
+    console.print()
+
+    # --- Bob's session ---
+    console.print("[dim]Bob connects (different session)[/dim]")
+
+    transport2 = StreamableHttpTransport(url=URL)
+    async with Client(transport=transport2) as bob:
+        result = await bob.call_tool("list_session_info", {})
+        console.print(f"  session [cyan]{result.data['session_id'][:8]}[/cyan]")
+
+        await bob.call_tool("get_value", {"key": "user"})
+        console.print("  get [white]user[/white] → [dim]not found[/dim]")
+
+        await bob.call_tool("get_value", {"key": "secret"})
+        console.print("  get [white]secret[/white] → [dim]not found[/dim]")
+
+        await bob.call_tool("set_value", {"key": "user", "value": "Bob"})
+        console.print("  set [white]user[/white] = [green]Bob[/green]")
+
+        await bob.call_tool("get_value", {"key": "user"})
+        console.print("  get [white]user[/white] → [green]Bob[/green]")
+
+    console.print()
+
+    # --- Alice reconnects ---
+    console.print("[dim]Alice reconnects (new session)[/dim]")
+
+    transport3 = StreamableHttpTransport(url=URL)
+    async with Client(transport=transport3) as alice_again:
+        result = await alice_again.call_tool("list_session_info", {})
+        console.print(f"  session [cyan]{result.data['session_id'][:8]}[/cyan]")
+
+        await alice_again.call_tool("get_value", {"key": "user"})
+        console.print("  get [white]user[/white] → [dim]not found[/dim]")
+
+    console.print()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())