feat(server): lifespan redesign with server-scoped and session-scoped lifetimes

README.v2.md

@@ -276,7 +276,7 @@ mcp = MCPServer("My App", lifespan=app_lifespan)
 @mcp.tool()
 def query_db(ctx: Context[AppContext]) -> str:
     """Tool that uses initialized resources."""
-    db = ctx.request_context.lifespan_context.db
+    db = ctx.request_context.session_lifespan_context.db
     return db.query()
 ```
 
@@ -1125,35 +1125,48 @@ async def notify_data_update(resource_uri: str, ctx: Context) -> str:
 
 The request context accessible via `ctx.request_context` contains request-specific information and resources:
 
-- `ctx.request_context.lifespan_context` - Access to resources initialized during server startup
-  - Database connections, configuration objects, shared services
-  - Type-safe access to resources defined in your server's lifespan function
+- `ctx.request_context.session_lifespan_context` - Access to resources from the session lifespan (runs per-client connection)
+  - User authentication context, per-client state, session IDs
+  - Type-safe access to resources defined in your server's session lifespan function
+- `ctx.request_context.server_lifespan_context` - Access to resources from the server lifespan (runs once at server startup)
+  - Database connection pools, ML models, shared caches, global configuration
+  - Type-safe access to resources defined in your server's server lifespan function
+  - **Note:** When using MCPServer with `lifespan` parameter, this is populated with that context
 - `ctx.request_context.meta` - Request metadata from the client including:
   - `progressToken` - Token for progress notifications
   - Other client-provided metadata
 - `ctx.request_context.request` - The original MCP request object for advanced processing
 - `ctx.request_context.request_id` - Unique identifier for this request
 
 ```python
-# Example with typed lifespan context
+# Example with typed contexts
 @dataclass
-class AppContext:
+class ServerContext:
     db: Database
     config: AppConfig
 
+@dataclass
+class SessionContext:
+    user_id: str
+    session_id: str
+
 @mcp.tool()
 def query_with_config(query: str, ctx: Context) -> str:
-    """Execute a query using shared database and configuration."""
-    # Access typed lifespan context
-    app_ctx: AppContext = ctx.request_context.lifespan_context
+    """Execute a query using shared database and per-session user context."""
+    # Access server-level context (shared across all clients)
+    server_ctx: ServerContext = ctx.request_context.server_lifespan_context
+
+    # Access session-level context (per-client)
+    session_ctx: SessionContext = ctx.request_context.session_lifespan_context
 
-    # Use shared resources
-    connection = app_ctx.db
-    settings = app_ctx.config
+    # Use resources from both contexts
+    connection = server_ctx.db
+    settings = server_ctx.config
+    user = session_ctx.user_id
 
-    # Execute query with configuration
-    result = connection.execute(query, timeout=settings.query_timeout)
-    return str(result)
+    # Execute query with configuration and user context
+    result = connection.execute(query, timeout=settings.query_timeout, user=user)
+    return f"User {user}: {str(result)}"
 ```
 
 _Full lifespan example: [examples/snippets/servers/lifespan_example.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/lifespan_example.py)_
@@ -1643,6 +1656,7 @@ uv run examples/snippets/servers/lowlevel/lifespan.py
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from typing import TypedDict
+from uuid import uuid4
 
 import mcp.server.stdio
 from mcp import types
@@ -1653,71 +1667,141 @@ from mcp.server import Server, ServerRequestContext
 class Database:
     """Mock database class for example."""
 
+    connections: int = 0
+
     @classmethod
     async def connect(cls) -> "Database":
         """Connect to database."""
-        print("Database connected")
+        cls.connections += 1
+        print(f"Database connected (total connections: {cls.connections})")
         return cls()
 
     async def disconnect(self) -> None:
         """Disconnect from database."""
-        print("Database disconnected")
+        self.connections -= 1
+        print(f"Database disconnected (total connections: {self.connections})")
 
     async def query(self, query_str: str) -> list[dict[str, str]]:
         """Execute a query."""
         # Simulate database query
         return [{"id": "1", "name": "Example", "query": query_str}]
 
 
-class AppContext(TypedDict):
+class ServerContext(TypedDict):
+    """Server-level context (shared across all clients)."""
+
     db: Database
 
 
+class SessionContext(TypedDict):
+    """Session-level context (per-client connection)."""
+
+    session_id: str
+
+
 @asynccontextmanager
-async def server_lifespan(_server: Server[AppContext]) -> AsyncIterator[AppContext]:
-    """Manage server startup and shutdown lifecycle."""
+async def server_lifespan(_server: Server) -> AsyncIterator[ServerContext]:
+    """Manage server startup and shutdown lifecycle.
+
+    This runs ONCE when the server process starts, before any clients connect.
+    Use this for resources that should be shared across all client connections:
+    - Database connection pools
+    - Machine learning models
+    - Shared caches
+    - Global configuration
+    """
+    print("[SERVER LIFESPAN] Starting server...")
     db = await Database.connect()
     try:
+        print("[SERVER LIFESPAN] Server started, database connected")
         yield {"db": db}
     finally:
         await db.disconnect()
+        print("[SERVER LIFESPAN] Server stopped, database disconnected")
+
+
+@asynccontextmanager
+async def session_lifespan(_server: Server) -> AsyncIterator[SessionContext]:
+    """Manage per-client session lifecycle.
+
+    This runs FOR EACH CLIENT that connects to the server.
+    Use this for resources that are specific to a single client connection:
+    - User authentication context
+    - Per-client transaction state
+    - Client-specific caches
+    - Session identifiers
+    """
+    session_id = str(uuid4())
+    print(f"[SESSION LIFESPAN] Session {session_id} started")
+    try:
+        yield {"session_id": session_id}
+    finally:
+        print(f"[SESSION LIFESPAN] Session {session_id} stopped")
 
 
 async def handle_list_tools(
-    ctx: ServerRequestContext[AppContext], params: types.PaginatedRequestParams | None
+    ctx: ServerRequestContext[ServerContext, SessionContext],
+    params: types.PaginatedRequestParams | None,
 ) -> types.ListToolsResult:
     """List available tools."""
     return types.ListToolsResult(
         tools=[
             types.Tool(
                 name="query_db",
-                description="Query the database",
+                description="Query the database (uses shared server connection)",
                 input_schema={
                     "type": "object",
                     "properties": {"query": {"type": "string", "description": "SQL query to execute"}},
                     "required": ["query"],
                 },
-            )
+            ),
+            types.Tool(
+                name="get_session_info",
+                description="Get information about the current session",
+                input_schema={
+                    "type": "object",
+                    "properties": {},
+                },
+            ),
         ]
     )
 
 
 async def handle_call_tool(
-    ctx: ServerRequestContext[AppContext], params: types.CallToolRequestParams
+    ctx: ServerRequestContext[ServerContext, SessionContext],
+    params: types.CallToolRequestParams,
 ) -> types.CallToolResult:
-    """Handle database query tool call."""
-    if params.name != "query_db":
-        raise ValueError(f"Unknown tool: {params.name}")
+    """Handle tool calls."""
+    if params.name == "query_db":
+        # Access server-level resource (shared database connection)
+        db = ctx.server_lifespan_context["db"]
+        results = await db.query((params.arguments or {})["query"])
+
+        return types.CallToolResult(
+            content=[
+                types.TextContent(
+                    type="text",
+                    text=f"Query results (session {ctx.session_lifespan_context['session_id']}): {results}",
+                )
+            ]
+        )
 
-    db = ctx.lifespan_context["db"]
-    results = await db.query((params.arguments or {})["query"])
+    if params.name == "get_session_info":
+        # Access session-level resource (session ID)
+        session_id = ctx.session_lifespan_context["session_id"]
 
-    return types.CallToolResult(content=[types.TextContent(type="text", text=f"Query results: {results}")])
+        return types.CallToolResult(
+            content=[types.TextContent(type="text", text=f"Your session ID: {session_id}")]
+        )
+
+    raise ValueError(f"Unknown tool: {params.name}")
 
 
+# Create server with BOTH server and session lifespans
 server = Server(
     "example-server",
-    lifespan=server_lifespan,
+    server_lifespan=server_lifespan,  # Runs once at server startup
+    session_lifespan=session_lifespan,  # Runs per-client connection
     on_list_tools=handle_list_tools,
     on_call_tool=handle_call_tool,
 )

docs/migration.md

@@ -798,6 +798,89 @@ params = CallToolRequestParams(
 )
 ```
 
+### Lifespan redesign: Server-scoped and Session-scoped lifetimes
+
+The single `lifespan` parameter has been replaced with two separate parameters: `server_lifespan` and `session_lifespan`. This fixes bugs where server-level resources (like database pools) were being initialized per-client connection instead of once at server startup.
+
+**Before (v1):**
+
+```python
+from mcp.server import Server
+
+@asynccontextmanager
+async def lifespan(server):
+    # This ran PER-CLIENT, causing bugs #1300 and #1304
+    db_pool = await create_db_pool()
+    try:
+        yield {"db": db_pool}
+    finally:
+        await db_pool.close()
+
+server = Server("my-server", lifespan=lifespan)
+```
+
+**After (v2):**
+
+```python
+from mcp.server import Server
+
+@asynccontextmanager
+async def server_lifespan(server):
+    # Runs ONCE at server startup
+    # Use for: database pools, ML models, shared caches
+    db_pool = await create_db_pool()
+    try:
+        yield {"db": db_pool}
+    finally:
+        await db_pool.close()
+
+@asynccontextmanager
+async def session_lifespan(server):
+    # Runs PER-CLIENT connection
+    # Use for: user auth, per-client state
+    session_id = str(uuid4())
+    try:
+        yield {"session_id": session_id}
+    finally:
+        pass
+
+server = Server(
+    "my-server",
+    server_lifespan=server_lifespan,  # Server-scoped
+    session_lifespan=session_lifespan,  # Session-scoped
+)
+
+# Handlers can access both contexts
+async def handle_tool(ctx, params):
+    db = ctx.server_lifespan_context["db"]  # Shared resource
+    session_id = ctx.session_lifespan_context["session_id"]  # Per-client resource
+    ...
+```
+
+**Key differences:**
+
+| v1 (`lifespan`) | v2 (`server_lifespan` / `session_lifespan`) |
+|-----------------|---------------------------------------------------|
+| Ran per-client connection | `server_lifespan` runs once at startup |
+| No separation of concerns | `session_lifespan` runs per-client |
+| `ctx.lifespan_context` | `ctx.server_lifespan_context` and `ctx.session_lifespan_context` |
+| Database pools connected on first client | Database pools connected at server startup |
+| Bug: resources re-initialized unnecessarily | Fixed: proper resource lifecycle |
+
+**When to use each:**
+
+- **`server_lifespan`**: Server-level resources that persist across all clients
+  - Database connection pools
+  - Machine learning models
+  - Shared caches
+  - Global configuration
+
+- **`session_lifespan`**: Client-specific resources
+  - User authentication context
+  - Per-client transaction state
+  - Session identifiers
+  - Client-specific caches
+
 ## New Features
 
 ### `streamable_http_app()` available on lowlevel Server