Add composable lifespans (#2828)

docs/docs.json

@@ -124,6 +124,7 @@
                   "servers/context",
                   "servers/elicitation",
                   "servers/icons",
+                  "servers/lifespan",
                   "servers/logging",
                   "servers/middleware",
                   "servers/progress",

docs/integrations/fastapi.mdx

@@ -410,41 +410,31 @@ If you need CORS on your own FastAPI routes, use the sub-app pattern: mount your
 
 ### Combining Lifespans
 
-If your FastAPI app already has a lifespan (for database connections, startup tasks, etc.), you can't simply replace it with the MCP lifespan. Instead, you need to create a new lifespan function that manages both contexts. This ensures that both your app's initialization logic and the MCP server's session manager run properly:
+If your FastAPI app already has a lifespan (for database connections, startup tasks, etc.), you can't simply replace it with the MCP lifespan. Use `combine_lifespans` to run both:
 
 ```python
-from contextlib import asynccontextmanager
 from fastapi import FastAPI
 from fastmcp import FastMCP
+from fastmcp.utilities.lifespan import combine_lifespans
+from contextlib import asynccontextmanager
 
 # Your existing lifespan
 @asynccontextmanager
 async def app_lifespan(app: FastAPI):
-    # Startup
     print("Starting up the app...")
-    # Initialize database, cache, etc.
     yield
-    # Shutdown
     print("Shutting down the app...")
 
 # Create MCP server
 mcp = FastMCP("Tools")
-mcp_app = mcp.http_app(path='/mcp')
+mcp_app = mcp.http_app()
 
 # Combine both lifespans
-@asynccontextmanager
-async def combined_lifespan(app: FastAPI):
-    # Run both lifespans
-    async with app_lifespan(app):
-        async with mcp_app.lifespan(app):
-            yield
-
-# Use the combined lifespan
-app = FastAPI(lifespan=combined_lifespan)
-app.mount("/mcp", mcp_app)
+app = FastAPI(lifespan=combine_lifespans(app_lifespan, mcp_app.lifespan))
+app.mount("/mcp", mcp_app)  # MCP endpoint at /mcp
 ```
 
-This pattern ensures both your app's initialization logic and the MCP server's session manager are properly managed. The key is using nested `async with` statements - the inner context (MCP) will be initialized after the outer context (your app), and cleaned up before it. This maintains the correct initialization and cleanup order for all your resources.
+`combine_lifespans` enters lifespans in order and exits in reverse order.
 
 ### Performance Tips
 

docs/servers/lifespan.mdx

@@ -0,0 +1,141 @@
+---
+title: Lifespans
+description: Server-level setup and teardown with composable lifespans
+---
+
+Lifespans let you run code once when the server starts and clean up when it stops. Unlike per-session handlers, lifespans run exactly once regardless of how many clients connect.
+
+## Basic Usage
+
+Use the `@lifespan` decorator to define a lifespan:
+
+```python
+from fastmcp import FastMCP
+from fastmcp.server.lifespan import lifespan
+
+@lifespan
+async def app_lifespan(server):
+    # Setup: runs once when server starts
+    print("Starting up...")
+    try:
+        yield {"started_at": "2024-01-01"}
+    finally:
+        # Teardown: runs when server stops
+        print("Shutting down...")
+
+mcp = FastMCP("MyServer", lifespan=app_lifespan)
+```
+
+The dict you yield becomes the **lifespan context**, accessible from tools.
+
+<Note>
+Always use `try/finally` for cleanup code to ensure it runs even if the server is cancelled.
+</Note>
+
+## Accessing Lifespan Context
+
+Access the lifespan context in tools via `ctx.lifespan_context`:
+
+```python
+from fastmcp import FastMCP, Context
+from fastmcp.server.lifespan import lifespan
+
+@lifespan
+async def app_lifespan(server):
+    # Initialize shared state
+    data = {"users": ["alice", "bob"]}
+    yield {"data": data}
+
+mcp = FastMCP("MyServer", lifespan=app_lifespan)
+
+@mcp.tool
+def list_users(ctx: Context) -> list[str]:
+    data = ctx.lifespan_context["data"]
+    return data["users"]
+```
+
+## Composing Lifespans
+
+Compose multiple lifespans with the `|` operator:
+
+```python
+from fastmcp import FastMCP
+from fastmcp.server.lifespan import lifespan
+
+@lifespan
+async def config_lifespan(server):
+    config = {"debug": True, "version": "1.0"}
+    yield {"config": config}
+
+@lifespan
+async def data_lifespan(server):
+    data = {"items": []}
+    yield {"data": data}
+
+# Compose with |
+mcp = FastMCP("MyServer", lifespan=config_lifespan | data_lifespan)
+```
+
+Composed lifespans:
+- Enter in order (left to right)
+- Exit in reverse order (right to left)
+- Merge their context dicts (later values overwrite earlier on conflict)
+
+## Backwards Compatibility
+
+Existing `@asynccontextmanager` lifespans still work when passed directly to FastMCP:
+
+```python
+from contextlib import asynccontextmanager
+from fastmcp import FastMCP
+
+@asynccontextmanager
+async def legacy_lifespan(server):
+    yield {"key": "value"}
+
+mcp = FastMCP("MyServer", lifespan=legacy_lifespan)
+```
+
+To compose an `@asynccontextmanager` function with `@lifespan` functions, wrap it with `ContextManagerLifespan`:
+
+```python
+from contextlib import asynccontextmanager
+from fastmcp.server.lifespan import lifespan, ContextManagerLifespan
+
+@asynccontextmanager
+async def legacy_lifespan(server):
+    yield {"legacy": True}
+
+@lifespan
+async def new_lifespan(server):
+    yield {"new": True}
+
+# Wrap the legacy lifespan explicitly for composition
+combined = ContextManagerLifespan(legacy_lifespan) | new_lifespan
+```
+
+## With FastAPI
+
+When mounting FastMCP into FastAPI, use `combine_lifespans` to run both your app's lifespan and the MCP server's lifespan:
+
+```python
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+from fastmcp import FastMCP
+from fastmcp.utilities.lifespan import combine_lifespans
+
+@asynccontextmanager
+async def app_lifespan(app):
+    print("FastAPI starting...")
+    yield
+    print("FastAPI shutting down...")
+
+mcp = FastMCP("Tools")
+mcp_app = mcp.http_app()
+
+app = FastAPI(lifespan=combine_lifespans(app_lifespan, mcp_app.lifespan))
+app.mount("/mcp", mcp_app)
+```
+
+See the [FastAPI integration guide](/integrations/fastapi#combining-lifespans) for full details.

docs/servers/server.mdx

@@ -60,8 +60,8 @@ The `FastMCP` constructor accepts several arguments:
   Authentication provider for securing HTTP-based transports. See [Authentication](/servers/auth/authentication) for configuration options
 </ParamField>
 
-<ParamField body="lifespan" type="AsyncContextManager | None">
-  An async context manager function for server startup and shutdown logic
+<ParamField body="lifespan" type="Lifespan | AsyncContextManager | None">
+  Server-level setup and teardown logic. See [Lifespans](/servers/lifespan) for composable lifespans
 </ParamField>
 
 <ParamField body="tools" type="list[Tool | Callable] | None">

src/fastmcp/server/context.py

@@ -248,6 +248,29 @@ class Context:
         except LookupError:
             return None
 
+    @property
+    def lifespan_context(self) -> dict[str, Any]:
+        """Access the server's lifespan context.
+
+        Returns the context dict yielded by the server's lifespan function.
+        Returns an empty dict if no lifespan was configured or if the MCP
+        session is not yet established.
+
+        Example:
+        ```python
+        @server.tool
+        def my_tool(ctx: Context) -> str:
+            db = ctx.lifespan_context.get("db")
+            if db:
+                return db.query("SELECT 1")
+            return "No database connection"
+        ```
+        """
+        rc = self.request_context
+        if rc is None:
+            return {}
+        return rc.lifespan_context
+
     async def report_progress(
         self, progress: float, total: float | None = None, message: str | None = None
     ) -> None:

src/fastmcp/server/lifespan.py

@@ -0,0 +1,198 @@
+"""Composable lifespans for FastMCP servers.
+
+This module provides a `@lifespan` decorator for creating composable server lifespans
+that can be combined using the `|` operator.
+
+Example:
+    ```python
+    from fastmcp import FastMCP
+    from fastmcp.server.lifespan import lifespan
+
+    @lifespan
+    async def db_lifespan(server):
+        conn = await connect_db()
+        yield {"db": conn}
+        await conn.close()
+
+    @lifespan
+    async def cache_lifespan(server):
+        cache = await connect_cache()
+        yield {"cache": cache}
+        await cache.close()
+
+    mcp = FastMCP("server", lifespan=db_lifespan | cache_lifespan)
+    ```
+
+To compose with existing `@asynccontextmanager` lifespans, wrap them explicitly:
+
+    ```python
+    from contextlib import asynccontextmanager
+    from fastmcp.server.lifespan import lifespan, ContextManagerLifespan
+
+    @asynccontextmanager
+    async def legacy_lifespan(server):
+        yield {"legacy": True}
+
+    @lifespan
+    async def new_lifespan(server):
+        yield {"new": True}
+
+    # Wrap the legacy lifespan explicitly
+    combined = ContextManagerLifespan(legacy_lifespan) | new_lifespan
+    ```
+"""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator, Callable
+from contextlib import AbstractAsyncContextManager, asynccontextmanager
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from fastmcp.server.server import FastMCP
+
+
+LifespanFn = Callable[["FastMCP[Any]"], AsyncIterator[dict[str, Any] | None]]
+LifespanContextManagerFn = Callable[
+    ["FastMCP[Any]"], AbstractAsyncContextManager[dict[str, Any] | None]
+]
+
+
+class Lifespan:
+    """Composable lifespan wrapper.
+
+    Wraps an async generator function and enables composition via the `|` operator.
+    The wrapped function should yield a dict that becomes part of the lifespan context.
+    """
+
+    def __init__(self, fn: LifespanFn) -> None:
+        """Initialize a Lifespan wrapper.
+
+        Args:
+            fn: An async generator function that takes a FastMCP server and yields
+                a dict for the lifespan context.
+        """
+        self._fn = fn
+
+    @asynccontextmanager
+    async def __call__(self, server: FastMCP[Any]) -> AsyncIterator[dict[str, Any]]:
+        """Execute the lifespan as an async context manager.
+
+        Args:
+            server: The FastMCP server instance.
+
+        Yields:
+            The lifespan context dict.
+        """
+        async with asynccontextmanager(self._fn)(server) as result:
+            yield result if result is not None else {}
+
+    def __or__(self, other: Lifespan) -> ComposedLifespan:
+        """Compose with another lifespan using the | operator.
+
+        Args:
+            other: Another Lifespan instance.
+
+        Returns:
+            A ComposedLifespan that runs both lifespans.
+
+        Raises:
+            TypeError: If other is not a Lifespan instance.
+        """
+        if not isinstance(other, Lifespan):
+            raise TypeError(
+                f"Cannot compose Lifespan with {type(other).__name__}. "
+                f"Use @lifespan decorator or wrap with ContextManagerLifespan()."
+            )
+        return ComposedLifespan(self, other)
+
+
+class ContextManagerLifespan(Lifespan):
+    """Lifespan wrapper for already-wrapped context manager functions.
+
+    Use this for functions already decorated with @asynccontextmanager.
+    """
+
+    _fn: LifespanContextManagerFn  # Override type for this subclass
+
+    def __init__(self, fn: LifespanContextManagerFn) -> None:
+        """Initialize with a context manager factory function."""
+        self._fn = fn  # type: ignore[assignment]
+
+    @asynccontextmanager
+    async def __call__(self, server: FastMCP[Any]) -> AsyncIterator[dict[str, Any]]:
+        """Execute the lifespan as an async context manager.
+
+        Args:
+            server: The FastMCP server instance.
+
+        Yields:
+            The lifespan context dict.
+        """
+        # self._fn is already a context manager factory, just call it
+        async with self._fn(server) as result:
+            yield result if result is not None else {}
+
+
+class ComposedLifespan(Lifespan):
+    """Two lifespans composed together.
+
+    Enters the left lifespan first, then the right. Exits in reverse order.
+    Results are shallow-merged into a single dict.
+    """
+
+    def __init__(self, left: Lifespan, right: Lifespan) -> None:
+        """Initialize a composed lifespan.
+
+        Args:
+            left: The first lifespan to enter.
+            right: The second lifespan to enter.
+        """
+        # Don't call super().__init__ since we override __call__
+        self._left = left
+        self._right = right
+
+    @asynccontextmanager
+    async def __call__(self, server: FastMCP[Any]) -> AsyncIterator[dict[str, Any]]:
+        """Execute both lifespans, merging their results.
+
+        Args:
+            server: The FastMCP server instance.
+
+        Yields:
+            The merged lifespan context dict from both lifespans.
+        """
+        async with (
+            self._left(server) as left_result,
+            self._right(server) as right_result,
+        ):
+            yield {**left_result, **right_result}
+
+
+def lifespan(fn: LifespanFn) -> Lifespan:
+    """Decorator to create a composable lifespan.
+
+    Use this decorator on an async generator function to make it composable
+    with other lifespans using the `|` operator.
+
+    Example:
+        ```python
+        @lifespan
+        async def my_lifespan(server):
+            # Setup
+            resource = await create_resource()
+            yield {"resource": resource}
+            # Teardown
+            await resource.close()
+
+        mcp = FastMCP("server", lifespan=my_lifespan | other_lifespan)
+        ```
+
+    Args:
+        fn: An async generator function that takes a FastMCP server and yields
+            a dict for the lifespan context.
+
+    Returns:
+        A composable Lifespan wrapper.
+    """
+    return Lifespan(fn)

src/fastmcp/server/server.py

@@ -76,6 +76,7 @@ from fastmcp.server.http import (
     create_sse_app,
     create_streamable_http_app,
 )
+from fastmcp.server.lifespan import Lifespan
 from fastmcp.server.low_level import LowLevelServer
 from fastmcp.server.middleware import Middleware, MiddlewareContext
 from fastmcp.server.providers import LocalProvider, Provider
@@ -203,7 +204,7 @@ class FastMCP(Generic[LifespanResultT]):
         auth: AuthProvider | None = None,
         middleware: Sequence[Middleware] | None = None,
         providers: Sequence[Provider] | None = None,
-        lifespan: LifespanCallable | None = None,
+        lifespan: LifespanCallable | Lifespan | None = None,
         mask_error_details: bool | None = None,
         tools: Sequence[Tool | Callable[..., Any]] | None = None,
         tool_transformations: Mapping[str, ToolTransformConfig] | None = None,
@@ -281,7 +282,11 @@ class FastMCP(Generic[LifespanResultT]):
             )
         self._tool_serializer: Callable[[Any], str] | None = tool_serializer
 
-        self._lifespan: LifespanCallable[LifespanResultT] = lifespan or default_lifespan
+        # Handle Lifespan instances (they're callable) or regular lifespan functions
+        if lifespan is not None:
+            self._lifespan: LifespanCallable[LifespanResultT] = lifespan
+        else:
+            self._lifespan = cast(LifespanCallable[LifespanResultT], default_lifespan)
         self._lifespan_result: LifespanResultT | None = None
         self._lifespan_result_set: bool = False
         self._started: asyncio.Event = asyncio.Event()

src/fastmcp/utilities/lifespan.py

@@ -0,0 +1,56 @@
+"""Lifespan utilities for combining async context manager lifespans."""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator, Callable
+from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager
+from typing import Any, TypeVar
+
+AppT = TypeVar("AppT")
+
+
+def combine_lifespans(
+    *lifespans: Callable[[AppT], AbstractAsyncContextManager[dict[str, Any] | None]],
+) -> Callable[[AppT], AbstractAsyncContextManager[dict[str, Any]]]:
+    """Combine multiple lifespans into a single lifespan.
+
+    Useful when mounting FastMCP into FastAPI and you need to run
+    both your app's lifespan and the MCP server's lifespan.
+
+    Works with both FastAPI-style lifespans (yield None) and FastMCP-style
+    lifespans (yield dict). Results are merged; later lifespans override
+    earlier ones on key conflicts.
+
+    Lifespans are entered in order and exited in reverse order (LIFO).
+
+    Example:
+        ```python
+        from fastmcp import FastMCP
+        from fastmcp.utilities.lifespan import combine_lifespans
+        from fastapi import FastAPI
+
+        mcp = FastMCP("Tools")
+        mcp_app = mcp.http_app()
+
+        app = FastAPI(lifespan=combine_lifespans(app_lifespan, mcp_app.lifespan))
+        app.mount("/mcp", mcp_app)  # MCP endpoint at /mcp
+        ```
+
+    Args:
+        *lifespans: Lifespan context manager factories to combine.
+
+    Returns:
+        A combined lifespan context manager factory.
+    """
+
+    @asynccontextmanager
+    async def combined(app: AppT) -> AsyncIterator[dict[str, Any]]:
+        merged: dict[str, Any] = {}
+        async with AsyncExitStack() as stack:
+            for ls in lifespans:
+                result = await stack.enter_async_context(ls(app))
+                if result is not None:
+                    merged.update(result)
+            yield merged
+
+    return combined

tests/server/test_server_lifespan.py

@@ -4,8 +4,12 @@ from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from typing import Any
 
+import pytest
+
 from fastmcp import Client, FastMCP
 from fastmcp.server.context import Context
+from fastmcp.server.lifespan import ContextManagerLifespan, lifespan
+from fastmcp.utilities.lifespan import combine_lifespans
 
 
 class TestServerLifespan:
@@ -62,10 +66,399 @@ class TestServerLifespan:
         @mcp.tool
         def get_db_info(ctx: Context) -> str:
             # Access the server lifespan context
-            assert ctx.request_context is not None
+            assert ctx.request_context is not None  # type narrowing for type checker
             lifespan_context = ctx.request_context.lifespan_context
             return lifespan_context.get("db_connection", "no_db")
 
         async with Client(mcp) as client:
             result = await client.call_tool("get_db_info", {})
             assert result.data == "mock_db"
+
+
+class TestComposableLifespans:
+    """Test composable lifespan functionality."""
+
+    async def test_lifespan_decorator_basic(self):
+        """Test that the @lifespan decorator works like @asynccontextmanager."""
+        events: list[str] = []
+
+        @lifespan
+        async def my_lifespan(server: FastMCP) -> AsyncIterator[dict[str, Any]]:
+            events.append("enter")
+            try:
+                yield {"key": "value"}
+            finally:
+                events.append("exit")
+
+        mcp = FastMCP("TestServer", lifespan=my_lifespan)
+
+        @mcp.tool
+        def get_info(ctx: Context) -> str:
+            assert ctx.request_context is not None
+            lifespan_context = ctx.request_context.lifespan_context
+            return lifespan_context.get("key", "missing")
+
+        assert events == []
+
+        async with Client(mcp) as client:
+            result = await client.call_tool("get_info", {})
+            assert result.data == "value"
+            assert events == ["enter"]
+
+        assert events == ["enter", "exit"]
+
+    async def test_lifespan_composition_two(self):
+        """Test composing two lifespans with |."""
+        events: list[str] = []
+
+        @lifespan
+        async def first_lifespan(server: FastMCP) -> AsyncIterator[dict[str, Any]]:
+            events.append("first_enter")
+            try:
+                yield {"first": "a"}
+            finally:
+                events.append("first_exit")
+
+        @lifespan
+        async def second_lifespan(server: FastMCP) -> AsyncIterator[dict[str, Any]]:
+            events.append("second_enter")
+            try:
+                yield {"second": "b"}
+            finally:
+                events.append("second_exit")
+
+        composed = first_lifespan | second_lifespan
+        mcp = FastMCP("TestServer", lifespan=composed)
+
+        @mcp.tool
+        def get_both(ctx: Context) -> dict:
+            assert ctx.request_context is not None
+            return dict(ctx.request_context.lifespan_context)
+
+        async with Client(mcp) as client:
+            result = await client.call_tool("get_both", {})
+            # Results should be merged
+            assert result.data == {"first": "a", "second": "b"}
+            # Should enter in order
+            assert events == ["first_enter", "second_enter"]
+
+        # Should exit in reverse order (LIFO)
+        assert events == ["first_enter", "second_enter", "second_exit", "first_exit"]
+
+    async def test_lifespan_composition_three(self):
+        """Test composing three lifespans with |."""
+        events: list[str] = []
+
+        @lifespan
+        async def ls_a(server: FastMCP) -> AsyncIterator[dict[str, Any]]:
+            events.append("a_enter")
+            try:
+                yield {"a": 1}
+            finally:
+                events.append("a_exit")
+
+        @lifespan
+        async def ls_b(server: FastMCP) -> AsyncIterator[dict[str, Any]]:
+            events.append("b_enter")
+            try:
+                yield {"b": 2}
+            finally:
+                events.append("b_exit")
+
+        @lifespan
+        async def ls_c(server: FastMCP) -> AsyncIterator[dict[str, Any]]:
+            events.append("c_enter")
+            try:
+                yield {"c": 3}
+            finally:
+                events.append("c_exit")
+
+        composed = ls_a | ls_b | ls_c
+        mcp = FastMCP("TestServer", lifespan=composed)
+
+        @mcp.tool
+        def get_all(ctx: Context) -> dict:
+            assert ctx.request_context is not None
+            return dict(ctx.request_context.lifespan_context)
+
+        async with Client(mcp) as client:
+            result = await client.call_tool("get_all", {})
+            assert result.data == {"a": 1, "b": 2, "c": 3}
+            assert events == ["a_enter", "b_enter", "c_enter"]
+
+        assert events == [
+            "a_enter",
+            "b_enter",
+            "c_enter",
+            "c_exit",
+            "b_exit",
+            "a_exit",
+        ]
+
+    async def test_lifespan_result_merge_later_wins(self):
+        """Test that later lifespans overwrite earlier ones on key conflict."""
+
+        @lifespan
+        async def first(server: FastMCP) -> AsyncIterator[dict[str, Any]]:
+            yield {"key": "first", "only_first": "yes"}
+
+        @lifespan
+        async def second(server: FastMCP) -> AsyncIterator[dict[str, Any]]:
+            yield {"key": "second", "only_second": "yes"}
+
+        composed = first | second
+        mcp = FastMCP("TestServer", lifespan=composed)
+
+        @mcp.tool
+        def get_context(ctx: Context) -> dict:
+            assert ctx.request_context is not None
+            return dict(ctx.request_context.lifespan_context)
+
+        async with Client(mcp) as client:
+            result = await client.call_tool("get_context", {})
+            # "key" should be overwritten by second
+            assert result.data == {
+                "key": "second",
+                "only_first": "yes",
+                "only_second": "yes",
+            }
+
+    async def test_lifespan_composition_with_context_manager_lifespan(self):
+        """Test composing with ContextManagerLifespan for @asynccontextmanager functions."""
+        events: list[str] = []
+
+        @asynccontextmanager
+        async def legacy_lifespan(server: FastMCP) -> AsyncIterator[dict[str, Any]]:
+            events.append("legacy_enter")
+            try:
+                yield {"legacy": True}
+            finally:
+                events.append("legacy_exit")
+
+        @lifespan
+        async def new_lifespan(server: FastMCP) -> AsyncIterator[dict[str, Any]]:
+            events.append("new_enter")
+            try:
+                yield {"new": True}
+            finally:
+                events.append("new_exit")
+
+        # Wrap the @asynccontextmanager function explicitly
+        composed = ContextManagerLifespan(legacy_lifespan) | new_lifespan
+        mcp = FastMCP("TestServer", lifespan=composed)
+
+        @mcp.tool
+        def get_context(ctx: Context) -> dict:
+            assert ctx.request_context is not None
+            return dict(ctx.request_context.lifespan_context)
+
+        async with Client(mcp) as client:
+            result = await client.call_tool("get_context", {})
+            assert result.data == {"legacy": True, "new": True}
+
+        assert events == [
+            "legacy_enter",
+            "new_enter",
+            "new_exit",
+            "legacy_exit",
+        ]
+
+    async def test_backwards_compatibility_asynccontextmanager(self):
+        """Test that existing @asynccontextmanager lifespans still work."""
+
+        @asynccontextmanager
+        async def old_style_lifespan(server: FastMCP) -> AsyncIterator[dict[str, Any]]:
+            yield {"old_style": True}
+
+        mcp = FastMCP("TestServer", lifespan=old_style_lifespan)
+
+        @mcp.tool
+        def get_context(ctx: Context) -> dict:
+            assert ctx.request_context is not None
+            return dict(ctx.request_context.lifespan_context)
+
+        async with Client(mcp) as client:
+            result = await client.call_tool("get_context", {})
+            assert result.data == {"old_style": True}
+
+    async def test_lifespan_or_requires_lifespan_instance(self):
+        """Test that | operator requires Lifespan instances and gives helpful error."""
+
+        @lifespan
+        async def my_lifespan(server: FastMCP) -> AsyncIterator[dict[str, Any]]:
+            yield {"key": "value"}
+
+        @asynccontextmanager
+        async def regular_lifespan(server: FastMCP) -> AsyncIterator[dict[str, Any]]:
+            yield {"regular": True}
+
+        # Composing with non-Lifespan should raise TypeError with helpful message
+        with pytest.raises(TypeError) as exc_info:
+            my_lifespan | regular_lifespan  # type: ignore[operator]
+
+        assert "ContextManagerLifespan" in str(exc_info.value)
+
+
+class TestCombineLifespans:
+    """Test combine_lifespans utility function."""
+
+    async def test_combine_lifespans_fastapi_style(self):
+        """Test combining lifespans that yield None (FastAPI-style)."""
+        events: list[str] = []
+
+        @asynccontextmanager
+        async def first_lifespan(app: Any) -> AsyncIterator[None]:
+            events.append("first_enter")
+            try:
+                yield
+            finally:
+                events.append("first_exit")
+
+        @asynccontextmanager
+        async def second_lifespan(app: Any) -> AsyncIterator[None]:
+            events.append("second_enter")
+            try:
+                yield
+            finally:
+                events.append("second_exit")
+
+        combined = combine_lifespans(first_lifespan, second_lifespan)
+
+        async with combined("mock_app") as result:
+            assert result == {}  # Empty dict when lifespans yield None
+            assert events == ["first_enter", "second_enter"]
+
+        # LIFO exit order
+        assert events == ["first_enter", "second_enter", "second_exit", "first_exit"]
+
+    async def test_combine_lifespans_fastmcp_style(self):
+        """Test combining lifespans that yield dicts (FastMCP-style)."""
+        events: list[str] = []
+
+        @asynccontextmanager
+        async def db_lifespan(app: Any) -> AsyncIterator[dict[str, Any]]:
+            events.append("db_enter")
+            try:
+                yield {"db": "connected"}
+            finally:
+                events.append("db_exit")
+
+        @asynccontextmanager
+        async def cache_lifespan(app: Any) -> AsyncIterator[dict[str, Any]]:
+            events.append("cache_enter")
+            try:
+                yield {"cache": "ready"}
+            finally:
+                events.append("cache_exit")
+
+        combined = combine_lifespans(db_lifespan, cache_lifespan)
+
+        async with combined("mock_app") as result:
+            assert result == {"db": "connected", "cache": "ready"}
+            assert events == ["db_enter", "cache_enter"]
+
+        assert events == ["db_enter", "cache_enter", "cache_exit", "db_exit"]
+
+    async def test_combine_lifespans_mixed_styles(self):
+        """Test combining FastAPI-style (yield None) and FastMCP-style (yield dict)."""
+        events: list[str] = []
+
+        @asynccontextmanager
+        async def fastapi_lifespan(app: Any) -> AsyncIterator[None]:
+            events.append("fastapi_enter")
+            try:
+                yield  # FastAPI-style: yield None
+            finally:
+                events.append("fastapi_exit")
+
+        @asynccontextmanager
+        async def fastmcp_lifespan(app: Any) -> AsyncIterator[dict[str, Any]]:
+            events.append("fastmcp_enter")
+            try:
+                yield {"mcp": "initialized"}  # FastMCP-style: yield dict
+            finally:
+                events.append("fastmcp_exit")
+
+        combined = combine_lifespans(fastapi_lifespan, fastmcp_lifespan)
+
+        async with combined("mock_app") as result:
+            # Only the dict from fastmcp_lifespan should be present
+            assert result == {"mcp": "initialized"}
+            assert events == ["fastapi_enter", "fastmcp_enter"]
+
+        assert events == [
+            "fastapi_enter",
+            "fastmcp_enter",
+            "fastmcp_exit",
+            "fastapi_exit",
+        ]
+
+    async def test_combine_lifespans_result_merge_later_wins(self):
+        """Test that later lifespans overwrite earlier ones on key conflict."""
+
+        @asynccontextmanager
+        async def first(app: Any) -> AsyncIterator[dict[str, Any]]:
+            yield {"key": "first", "only_first": "yes"}
+
+        @asynccontextmanager
+        async def second(app: Any) -> AsyncIterator[dict[str, Any]]:
+            yield {"key": "second", "only_second": "yes"}
+
+        combined = combine_lifespans(first, second)
+
+        async with combined("mock_app") as result:
+            assert result == {
+                "key": "second",  # Overwritten by later lifespan
+                "only_first": "yes",
+                "only_second": "yes",
+            }
+
+    async def test_combine_lifespans_three(self):
+        """Test combining three lifespans."""
+        events: list[str] = []
+
+        @asynccontextmanager
+        async def ls_a(app: Any) -> AsyncIterator[dict[str, Any]]:
+            events.append("a_enter")
+            try:
+                yield {"a": 1}
+            finally:
+                events.append("a_exit")
+
+        @asynccontextmanager
+        async def ls_b(app: Any) -> AsyncIterator[dict[str, Any]]:
+            events.append("b_enter")
+            try:
+                yield {"b": 2}
+            finally:
+                events.append("b_exit")
+
+        @asynccontextmanager
+        async def ls_c(app: Any) -> AsyncIterator[dict[str, Any]]:
+            events.append("c_enter")
+            try:
+                yield {"c": 3}
+            finally:
+                events.append("c_exit")
+
+        combined = combine_lifespans(ls_a, ls_b, ls_c)
+
+        async with combined("mock_app") as result:
+            assert result == {"a": 1, "b": 2, "c": 3}
+            assert events == ["a_enter", "b_enter", "c_enter"]
+
+        assert events == [
+            "a_enter",
+            "b_enter",
+            "c_enter",
+            "c_exit",
+            "b_exit",
+            "a_exit",
+        ]
+
+    async def test_combine_lifespans_empty(self):
+        """Test combining zero lifespans."""
+        combined = combine_lifespans()
+
+        async with combined("mock_app") as result:
+            assert result == {}