Repository: modelcontextprotocol/python-sdk

Issue:
Client hangs on valid server responses
**Describe the bug**
1. The client hangs when the server send a log notification after the initialization phase.
2. The client hangs when the server sends a response having the id as string.

Both of the actions operations are valid as per specification.

**To Reproduce**
[This repository](https://github.com/peter-jerry-ye/mcp-py-sdk-report) contains the files. 

By modifying `server.py` then run `uv run client.py` should give you the behavior.

It also has a client implemented with typescript sdk as comparison.

**Expected behavior**
The client should perform normally, as the client written with typescript sdk.

**Screenshots**
If applicable, add screenshots to help explain your problem.

**Desktop (please complete the following information):**
 - OS: MacOS
 - Browser [e.g. chrome, safari]: n/a
 - Version [e.g. 22]: n/a

**Smartphone (please complete the following information):**
 - Device: [e.g. iPhone6]: n/a
 - OS: [e.g. iOS8.1]: n/a
 - Browser [e.g. stock browser, safari]: n/a
 - Version [e.g. 22]: n/a

**Additional context**
n/a

Hints from issue comments:
@g-votte: I investigated this issue and found:
* The hang problem itself can be resolved by adding a cast in [`session.py`](https://github.com/modelcontextprotocol/python-sdk/blob/main/src/mcp/shared/session.py)
* However, before proceeding with any fix, I would like to hear maintainers' thoughts on whether this issue relates to python-sdk, typescript-sdk, or the specification itself

Here is my analysis to share what I found.

## Analysis of Hang Problem

Direct cause of the hang seems in those two lines:
https://github.com/modelcontextprotocol/python-sdk/blob/f10665db4c2f676da1131617ad67715952258712/src/mcp/shared/session.py#L218 
https://github.com/modelcontextprotocol/python-sdk/blob/f10665db4c2f676da1131617ad67715952258712/src/mcp/shared/session.py#L337 

What is happening?
* `_response_streams` identifies a request by the "integer" request id
* When receiving the response, the request id in the message (message.root.id) is not cast to integer
* Thus, if the request id is passed as a str, the program cannot identify the corresponding request id in the dictionary. Hence, the stream is not sent

Those logics make the following line keep waiting forever:
https://github.com/modelcontextprotocol/python-sdk/blob/f10665db4c2f676da1131617ad67715952258712/src/mcp/shared/session.py#L236

## Discussion

Following the typescript-sdk approach, we can resolve the hang by adding a cast.
- detail: https://github.com/modelcontextprotocol/typescript-sdk/blob/13c3eea3b80dd658d91cc0422e0711078306c7c7/src/shared/protocol.ts#L299 

Specifically, changing `self._response_streams.pop(message.root.id, None)` to `self._response_streams.pop(int(message.root.id), None)` in this line would fix the hang, allowing @peter-jerry-ye's example program to run without issues:
https://github.com/modelcontextprotocol/python-sdk/blob/f10665db4c2f676da1131617ad67715952258712/src/mcp/shared/session.py#L337 

However, I wonder about the intended design in the [MCP spec](https://github.com/modelcontextprotocol/specification/blob/3d4877e69cbc9921e1b511a90cdf17d42483036b/schema/2024-11-05/schema.ts#L56) - why is `RequestId` allowed to be either `string` (`str`) or `number` (`int`)?

There are three possibilities:

* If the schema is designed for developer experience by allowing incrementing integer IDs to be passed as strings, we should add integer casting
  * In this case, the current python-sdk behavior would be considered a bug, and we could create a PR to add the casting along with test cases to verify this behavior
* If the schema is designed to support non-numeric IDs (e.g., UUIDs), integer casting would be inappropriate
  * In this case, the typescript-sdk behavior would need revision rather than the python-sdk
* If this design direction is not yet finalized, neither SDK implementation can be considered definitively correct or incorrect
  * This would indicate a specification issue that should be addressed in the [specification repository](https://github.com/modelcontextprotocol/specification), as ambiguity in `RequestId` handling could cause interoperability issues

@dsp-ant Could you help guide our understanding of this issue? The above analysis might be missing important aspects of the MCP design, so we'd appreciate your insights on how to best address this.@peter-jerry-ye: > However, I wonder about the intended design in the [MCP spec](https://github.com/modelcontextprotocol/specification/blob/3d4877e69cbc9921e1b511a90cdf17d42483036b/schema/2024-11-05/schema.ts#L56) - why is `RequestId` allowed to be either `string` (`str`) or `number` (`int`)?

The MCP is based on [JSON RPC 2.0](https://www.jsonrpc.org/specification), where 

> An identifier established by the Client that MUST contain a String, Number, or NULL value if included. If it is not included it is assumed to be a notification. The value SHOULD normally not be Null and Numbers SHOULD NOT contain fractional parts
@dsp-ant: Thank you for the analysis. This is super helpful. We should fix this. We will stick to the JSON RPC standard and hence allow string or a number. Now we should NOT cast to integer but rather allow the _response_streams, since a string can contain a uuid or any other format that cannot be casted.

Now I am not even sure I fully follow WHY the bug is happening. It seems to me that the _response_stream dictionary uses `RequestId` as key, so it should be able to find any request id. `message.root.id` is also a `RequestId`. I have to do more diggig.

If any of you have time to see you can produce a unit tests for this, this would be massive help. There are some examples for creating a test with a client and a server in the repo.@peter-jerry-ye: > Now we should NOT cast to integer but rather allow the _response_streams, since a string can contain a uuid or any other format that cannot be casted.

I see, then I misunderstood the JSON rpc protocol. It seems that the id needs to be exactly the same in the sense that they should also be the same type: number or string. Then this might have been a misimplementation on typescript sdk's side.

Closing as this is the expected behavior.

@g-votte: @dsp-ant 
Thank you for your guidance!

Let me clarify the situation:
- Actually, the python-sdk behavior is correct - it properly handles `RequestId` allowing both string (such as UUID) and number.
  - The hang occurs in python-sdk when uncommenting [this line](https://github.com/peter-jerry-ye/mcp-py-sdk-report/blob/main/server.py#L17) of @peter-jerry-ye's code and using a string `RequestId` (`"1"` instead of `1`), which is the expected behavior.
- On the other hand, the hang does not happen in the typescript-sdk, which might be a bug in the typescript-sdk.
  - The typescript-sdk forcibly casts `RequestId` to Number [here](https://github.com/modelcontextprotocol/typescript-sdk/blob/13c3eea3b80dd658d91cc0422e0711078306c7c7/src/shared/protocol.ts#L299), even if the `RequestId` is a string.

Solution patch:
diff --git a/.git-blame-ignore-revs b/.git-blame-ignore-revs
new file mode 100644
index 000000000..e69de29bb
diff --git a/.gitignore b/.gitignore
index bb25f9e46..54006f93f 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,4 +1,5 @@
 .DS_Store
+scratch/
 
 # Byte-compiled / optimized / DLL files
 __pycache__/
diff --git a/README.md b/README.md
index 310bb35b5..370b4f334 100644
--- a/README.md
+++ b/README.md
@@ -128,6 +128,9 @@ The [Model Context Protocol (MCP)](https://modelcontextprotocol.io) lets you bui
 The FastMCP server is your core interface to the MCP protocol. It handles connection management, protocol compliance, and message routing:
 
 ```python
+# Add lifespan support for startup/shutdown with strong typing
+from dataclasses import dataclass
+from typing import AsyncIterator
 from mcp.server.fastmcp import FastMCP
 
 # Create a named server
@@ -135,6 +138,31 @@ mcp = FastMCP("My App")
 
 # Specify dependencies for deployment and development
 mcp = FastMCP("My App", dependencies=["pandas", "numpy"])
+
+@dataclass
+class AppContext:
+    db: Database  # Replace with your actual DB type
+
+@asynccontextmanager
+async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
+    """Manage application lifecycle with type-safe context"""
+    try:
+        # Initialize on startup
+        await db.connect()
+        yield AppContext(db=db)
+    finally:
+        # Cleanup on shutdown
+        await db.disconnect()
+
+# Pass lifespan to server
+mcp = FastMCP("My App", lifespan=app_lifespan)
+
+# Access type-safe lifespan context in tools
+@mcp.tool()
+def query_db(ctx: Context) -> str:
+    """Tool that uses initialized resources"""
+    db = ctx.request_context.lifespan_context["db"]
+    return db.query()
 ```
 
 ### Resources
@@ -334,7 +362,38 @@ def query_data(sql: str) -> str:
 
 ### Low-Level Server
 
-For more control, you can use the low-level server implementation directly. This gives you full access to the protocol and allows you to customize every aspect of your server:
+For more control, you can use the low-level server implementation directly. This gives you full access to the protocol and allows you to customize every aspect of your server, including lifecycle management through the lifespan API:
+
+```python
+from contextlib import asynccontextmanager
+from typing import AsyncIterator
+
+@asynccontextmanager
+async def server_lifespan(server: Server) -> AsyncIterator[dict]:
+    """Manage server startup and shutdown lifecycle."""
+    try:
+        # Initialize resources on startup
+        await db.connect()
+        yield {"db": db}
+    finally:
+        # Clean up on shutdown
+        await db.disconnect()
+
+# Pass lifespan to server
+server = Server("example-server", lifespan=server_lifespan)
+
+# Access lifespan context in handlers
+@server.call_tool()
+async def query_db(name: str, arguments: dict) -> list:
+    ctx = server.request_context
+    db = ctx.lifespan_context["db"]
+    return await db.query(arguments["query"])
+```
+
+The lifespan API provides:
+- A way to initialize resources when the server starts and clean them up when it stops
+- Access to initialized resources through the request context in handlers
+- Type-safe context passing between lifespan and request handlers
 
 ```python
 from mcp.server.lowlevel import Server, NotificationOptions
diff --git a/src/mcp/server/fastmcp/server.py b/src/mcp/server/fastmcp/server.py
index aa7c79bcb..5ae30a5ca 100644
--- a/src/mcp/server/fastmcp/server.py
+++ b/src/mcp/server/fastmcp/server.py
@@ -3,8 +3,13 @@
 import inspect
 import json
 import re
+from collections.abc import AsyncIterator
+from contextlib import (
+    AbstractAsyncContextManager,
+    asynccontextmanager,
+)
 from itertools import chain
-from typing import Any, Callable, Literal, Sequence
+from typing import Any, Callable, Generic, Literal, Sequence
 
 import anyio
 import pydantic_core
@@ -19,8 +24,16 @@
 from mcp.server.fastmcp.tools import ToolManager
 from mcp.server.fastmcp.utilities.logging import configure_logging, get_logger
 from mcp.server.fastmcp.utilities.types import Image
-from mcp.server.lowlevel import Server as MCPServer
 from mcp.server.lowlevel.helper_types import ReadResourceContents
+from mcp.server.lowlevel.server import (
+    LifespanResultT,
+)
+from mcp.server.lowlevel.server import (
+    Server as MCPServer,
+)
+from mcp.server.lowlevel.server import (
+    lifespan as default_lifespan,
+)
 from mcp.server.sse import SseServerTransport
 from mcp.server.stdio import stdio_server
 from mcp.shared.context import RequestContext
@@ -50,7 +63,7 @@
 logger = get_logger(__name__)
 
 
-class Settings(BaseSettings):
+class Settings(BaseSettings, Generic[LifespanResultT]):
     """FastMCP server settings.
 
     All settings can be configured via environment variables with the prefix FASTMCP_.
@@ -85,13 +98,36 @@ class Settings(BaseSettings):
         description="List of dependencies to install in the server environment",
     )
 
+    lifespan: (
+        Callable[["FastMCP"], AbstractAsyncContextManager[LifespanResultT]] | None
+    ) = Field(None, description="Lifespan context manager")
+
+
+def lifespan_wrapper(
+    app: "FastMCP",
+    lifespan: Callable[["FastMCP"], AbstractAsyncContextManager[LifespanResultT]],
+) -> Callable[[MCPServer], AbstractAsyncContextManager[object]]:
+    @asynccontextmanager
+    async def wrap(s: MCPServer) -> AsyncIterator[object]:
+        async with lifespan(app) as context:
+            yield context
+
+    return wrap
+
 
 class FastMCP:
     def __init__(
         self, name: str | None = None, instructions: str | None = None, **settings: Any
     ):
         self.settings = Settings(**settings)
-        self._mcp_server = MCPServer(name=name or "FastMCP", instructions=instructions)
+
+        self._mcp_server = MCPServer(
+            name=name or "FastMCP",
+            instructions=instructions,
+            lifespan=lifespan_wrapper(self, self.settings.lifespan)
+            if self.settings.lifespan
+            else default_lifespan,
+        )
         self._tool_manager = ToolManager(
             warn_on_duplicate_tools=self.settings.warn_on_duplicate_tools
         )
diff --git a/src/mcp/server/lowlevel/server.py b/src/mcp/server/lowlevel/server.py
index 3d9172260..c0008b329 100644
--- a/src/mcp/server/lowlevel/server.py
+++ b/src/mcp/server/lowlevel/server.py
@@ -68,8 +68,10 @@ async def main():
 import logging
 import warnings
 from collections.abc import Awaitable, Callable
-from typing import Any, Sequence
+from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager
+from typing import Any, AsyncIterator, Generic, Sequence, TypeVar
 
+import anyio
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 from pydantic import AnyUrl
 
@@ -84,7 +86,10 @@ async def main():
 
 logger = logging.getLogger(__name__)
 
-request_ctx: contextvars.ContextVar[RequestContext[ServerSession]] = (
+LifespanResultT = TypeVar("LifespanResultT")
+
+# This will be properly typed in each Server instance's context
+request_ctx: contextvars.ContextVar[RequestContext[ServerSession, Any]] = (
     contextvars.ContextVar("request_ctx")
 )
 
@@ -101,13 +106,33 @@ def __init__(
         self.tools_changed = tools_changed
 
 
-class Server:
+@asynccontextmanager
+async def lifespan(server: "Server") -> AsyncIterator[object]:
+    """Default lifespan context manager that does nothing.
+
+    Args:
+        server: The server instance this lifespan is managing
+
+    Returns:
+        An empty context object
+    """
+    yield {}
+
+
+class Server(Generic[LifespanResultT]):
     def __init__(
-        self, name: str, version: str | None = None, instructions: str | None = None
+        self,
+        name: str,
+        version: str | None = None,
+        instructions: str | None = None,
+        lifespan: Callable[
+            ["Server"], AbstractAsyncContextManager[LifespanResultT]
+        ] = lifespan,
     ):
         self.name = name
         self.version = version
         self.instructions = instructions
+        self.lifespan = lifespan
         self.request_handlers: dict[
             type, Callable[..., Awaitable[types.ServerResult]]
         ] = {
@@ -188,7 +213,7 @@ def get_capabilities(
         )
 
     @property
-    def request_context(self) -> RequestContext[ServerSession]:
+    def request_context(self) -> RequestContext[ServerSession, LifespanResultT]:
         """If called outside of a request context, this will raise a LookupError."""
         return request_ctx.get()
 
@@ -445,36 +470,54 @@ async def run(
         # in-process servers.
         raise_exceptions: bool = False,
     ):
-        with warnings.catch_warnings(record=True) as w:
-            async with ServerSession(
-                read_stream, write_stream, initialization_options
-            ) as session:
+        async with AsyncExitStack() as stack:
+            lifespan_context = await stack.enter_async_context(self.lifespan(self))
+            session = await stack.enter_async_context(
+                ServerSession(read_stream, write_stream, initialization_options)
+            )
+
+            async with anyio.create_task_group() as tg:
                 async for message in session.incoming_messages:
                     logger.debug(f"Received message: {message}")
 
-                    match message:
-                        case (
-                            RequestResponder(
-                                request=types.ClientRequest(root=req)
-                            ) as responder
-                        ):
-                            with responder:
-                                await self._handle_request(
-                                    message, req, session, raise_exceptions
-                                )
-                        case types.ClientNotification(root=notify):
-                            await self._handle_notification(notify)
-
-                    for warning in w:
-                        logger.info(
-                            f"Warning: {warning.category.__name__}: {warning.message}"
+                    tg.start_soon(
+                        self._handle_message,
+                        message,
+                        session,
+                        lifespan_context,
+                        raise_exceptions,
+                    )
+
+    async def _handle_message(
+        self,
+        message: RequestResponder[types.ClientRequest, types.ServerResult]
+        | types.ClientNotification
+        | Exception,
+        session: ServerSession,
+        lifespan_context: LifespanResultT,
+        raise_exceptions: bool = False,
+    ):
+        with warnings.catch_warnings(record=True) as w:
+            match message:
+                case (
+                    RequestResponder(request=types.ClientRequest(root=req)) as responder
+                ):
+                    with responder:
+                        await self._handle_request(
+                            message, req, session, lifespan_context, raise_exceptions
                         )
+                case types.ClientNotification(root=notify):
+                    await self._handle_notification(notify)
+
+            for warning in w:
+                logger.info(f"Warning: {warning.category.__name__}: {warning.message}")
 
     async def _handle_request(
         self,
         message: RequestResponder,
         req: Any,
         session: ServerSession,
+        lifespan_context: LifespanResultT,
         raise_exceptions: bool,
     ):
         logger.info(f"Processing request of type {type(req).__name__}")
@@ -491,6 +534,7 @@ async def _handle_request(
                         message.request_id,
                         message.request_meta,
                         session,
+                        lifespan_context,
                     )
                 )
                 response = await handler(req)
diff --git a/src/mcp/shared/context.py b/src/mcp/shared/context.py
index 760d55877..a45fdacd4 100644
--- a/src/mcp/shared/context.py
+++ b/src/mcp/shared/context.py
@@ -5,10 +5,12 @@
 from mcp.types import RequestId, RequestParams
 
 SessionT = TypeVar("SessionT", bound=BaseSession)
+LifespanContextT = TypeVar("LifespanContextT")
 
 
 @dataclass
-class RequestContext(Generic[SessionT]):
+class RequestContext(Generic[SessionT, LifespanContextT]):
     request_id: RequestId
     meta: RequestParams.Meta | None
     session: SessionT
+    lifespan_context: LifespanContextT


Test patch:
diff --git a/tests/issues/test_176_progress_token.py b/tests/issues/test_176_progress_token.py
index ed8ab128a..7f9131a1e 100644
--- a/tests/issues/test_176_progress_token.py
+++ b/tests/issues/test_176_progress_token.py
@@ -20,7 +20,10 @@ async def test_progress_token_zero_first_call():
     mock_meta.progressToken = 0  # This is the key test case - token is 0
 
     request_context = RequestContext(
-        request_id="test-request", session=mock_session, meta=mock_meta
+        request_id="test-request",
+        session=mock_session,
+        meta=mock_meta,
+        lifespan_context=None,
     )
 
     # Create context with our mocks
diff --git a/tests/issues/test_188_concurrency.py b/tests/issues/test_188_concurrency.py
new file mode 100644
index 000000000..a56c0d300
--- /dev/null
+++ b/tests/issues/test_188_concurrency.py
@@ -0,0 +1,49 @@
+import anyio
+from pydantic import AnyUrl
+
+from mcp.server.fastmcp import FastMCP
+from mcp.shared.memory import (
+    create_connected_server_and_client_session as create_session,
+)
+
+_sleep_time_seconds = 0.01
+_resource_name = "slow://slow_resource"
+
+
+async def test_messages_are_executed_concurrently():
+    server = FastMCP("test")
+
+    @server.tool("sleep")
+    async def sleep_tool():
+        await anyio.sleep(_sleep_time_seconds)
+        return "done"
+
+    @server.resource(_resource_name)
+    async def slow_resource():
+        await anyio.sleep(_sleep_time_seconds)
+        return "slow"
+
+    async with create_session(server._mcp_server) as client_session:
+        start_time = anyio.current_time()
+        async with anyio.create_task_group() as tg:
+            for _ in range(10):
+                tg.start_soon(client_session.call_tool, "sleep")
+                tg.start_soon(client_session.read_resource, AnyUrl(_resource_name))
+
+        end_time = anyio.current_time()
+
+        duration = end_time - start_time
+        assert duration < 3 * _sleep_time_seconds
+        print(duration)
+
+
+def main():
+    anyio.run(test_messages_are_executed_concurrently)
+
+
+if __name__ == "__main__":
+    import logging
+
+    logging.basicConfig(level=logging.DEBUG)
+
+    main()
diff --git a/tests/issues/test_192_request_id.py b/tests/issues/test_192_request_id.py
new file mode 100644
index 000000000..628f00f99
--- /dev/null
+++ b/tests/issues/test_192_request_id.py
@@ -0,0 +1,88 @@
+import anyio
+import pytest
+
+from mcp.server.lowlevel import NotificationOptions, Server
+from mcp.server.models import InitializationOptions
+from mcp.types import (
+    LATEST_PROTOCOL_VERSION,
+    ClientCapabilities,
+    Implementation,
+    InitializeRequestParams,
+    JSONRPCMessage,
+    JSONRPCNotification,
+    JSONRPCRequest,
+    NotificationParams,
+)
+
+
+@pytest.mark.anyio
+async def test_request_id_match() -> None:
+    """Test that the server preserves request IDs in responses."""
+    server = Server("test")
+    custom_request_id = "test-123"
+
+    # Create memory streams for communication
+    client_writer, client_reader = anyio.create_memory_object_stream(1)
+    server_writer, server_reader = anyio.create_memory_object_stream(1)
+
+    # Server task to process the request
+    async def run_server():
+        async with client_reader, server_writer:
+            await server.run(
+                client_reader,
+                server_writer,
+                InitializationOptions(
+                    server_name="test",
+                    server_version="1.0.0",
+                    capabilities=server.get_capabilities(
+                        notification_options=NotificationOptions(),
+                        experimental_capabilities={},
+                    ),
+                ),
+                raise_exceptions=True,
+            )
+
+    # Start server task
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(run_server)
+
+        # Send initialize request
+        init_req = JSONRPCRequest(
+            id="init-1",
+            method="initialize",
+            params=InitializeRequestParams(
+                protocolVersion=LATEST_PROTOCOL_VERSION,
+                capabilities=ClientCapabilities(),
+                clientInfo=Implementation(name="test-client", version="1.0.0"),
+            ).model_dump(by_alias=True, exclude_none=True),
+            jsonrpc="2.0",
+        )
+
+        await client_writer.send(JSONRPCMessage(root=init_req))
+        await server_reader.receive()  # Get init response but don't need to check it
+
+        # Send initialized notification
+        initialized_notification = JSONRPCNotification(
+            method="notifications/initialized",
+            params=NotificationParams().model_dump(by_alias=True, exclude_none=True),
+            jsonrpc="2.0",
+        )
+        await client_writer.send(JSONRPCMessage(root=initialized_notification))
+
+        # Send ping request with custom ID
+        ping_request = JSONRPCRequest(
+            id=custom_request_id, method="ping", params={}, jsonrpc="2.0"
+        )
+
+        await client_writer.send(JSONRPCMessage(root=ping_request))
+
+        # Read response
+        response = await server_reader.receive()
+
+        # Verify response ID matches request ID
+        assert (
+            response.root.id == custom_request_id
+        ), "Response ID should match request ID"
+
+        # Cancel server task
+        tg.cancel_scope.cancel()
diff --git a/tests/server/fastmcp/test_func_metadata.py b/tests/server/fastmcp/test_func_metadata.py
index b68fb9025..6461648eb 100644
--- a/tests/server/fastmcp/test_func_metadata.py
+++ b/tests/server/fastmcp/test_func_metadata.py
@@ -236,7 +236,7 @@ async def check_call(args):
 
 def test_complex_function_json_schema():
     """Test JSON schema generation for complex function arguments.
-    
+
     Note: Different versions of pydantic output slightly different
     JSON Schema formats for model fields with defaults. The format changed in 2.9.0:
 
@@ -245,16 +245,16 @@ def test_complex_function_json_schema():
          "allOf": [{"$ref": "#/$defs/Model"}],
          "default": {}
        }
-       
+
     2. Since 2.9.0:
        {
          "$ref": "#/$defs/Model",
          "default": {}
        }
-       
+
     Both formats are valid and functionally equivalent. This test accepts either format
     to ensure compatibility across our supported pydantic versions.
-    
+
     This change in format does not affect runtime behavior since:
     1. Both schemas validate the same way
     2. The actual model classes and validation logic are unchanged
@@ -262,17 +262,17 @@ def test_complex_function_json_schema():
     """
     meta = func_metadata(complex_arguments_fn)
     actual_schema = meta.arg_model.model_json_schema()
-    
+
     # Create a copy of the actual schema to normalize
     normalized_schema = actual_schema.copy()
-    
+
     # Normalize the my_model_a_with_default field to handle both pydantic formats
-    if 'allOf' in actual_schema['properties']['my_model_a_with_default']:
-        normalized_schema['properties']['my_model_a_with_default'] = {
-            '$ref': '#/$defs/SomeInputModelA',
-            'default': {}
+    if "allOf" in actual_schema["properties"]["my_model_a_with_default"]:
+        normalized_schema["properties"]["my_model_a_with_default"] = {
+            "$ref": "#/$defs/SomeInputModelA",
+            "default": {},
         }
-    
+
     assert normalized_schema == {
         "$defs": {
             "InnerModel": {
diff --git a/tests/server/test_lifespan.py b/tests/server/test_lifespan.py
new file mode 100644
index 000000000..14afb6b06
--- /dev/null
+++ b/tests/server/test_lifespan.py
@@ -0,0 +1,207 @@
+"""Tests for lifespan functionality in both low-level and FastMCP servers."""
+
+from contextlib import asynccontextmanager
+from typing import AsyncIterator
+
+import anyio
+import pytest
+from pydantic import TypeAdapter
+
+from mcp.server.fastmcp import Context, FastMCP
+from mcp.server.lowlevel.server import NotificationOptions, Server
+from mcp.server.models import InitializationOptions
+from mcp.types import (
+    ClientCapabilities,
+    Implementation,
+    InitializeRequestParams,
+    JSONRPCMessage,
+    JSONRPCNotification,
+    JSONRPCRequest,
+)
+
+
+@pytest.mark.anyio
+async def test_lowlevel_server_lifespan():
+    """Test that lifespan works in low-level server."""
+
+    @asynccontextmanager
+    async def test_lifespan(server: Server) -> AsyncIterator[dict]:
+        """Test lifespan context that tracks startup/shutdown."""
+        context = {"started": False, "shutdown": False}
+        try:
+            context["started"] = True
+            yield context
+        finally:
+            context["shutdown"] = True
+
+    server = Server("test", lifespan=test_lifespan)
+
+    # Create memory streams for testing
+    send_stream1, receive_stream1 = anyio.create_memory_object_stream(100)
+    send_stream2, receive_stream2 = anyio.create_memory_object_stream(100)
+
+    # Create a tool that accesses lifespan context
+    @server.call_tool()
+    async def check_lifespan(name: str, arguments: dict) -> list:
+        ctx = server.request_context
+        assert isinstance(ctx.lifespan_context, dict)
+        assert ctx.lifespan_context["started"]
+        assert not ctx.lifespan_context["shutdown"]
+        return [{"type": "text", "text": "true"}]
+
+    # Run server in background task
+    async with anyio.create_task_group() as tg:
+
+        async def run_server():
+            await server.run(
+                receive_stream1,
+                send_stream2,
+                InitializationOptions(
+                    server_name="test",
+                    server_version="0.1.0",
+                    capabilities=server.get_capabilities(
+                        notification_options=NotificationOptions(),
+                        experimental_capabilities={},
+                    ),
+                ),
+                raise_exceptions=True,
+            )
+
+        tg.start_soon(run_server)
+
+        # Initialize the server
+        params = InitializeRequestParams(
+            protocolVersion="2024-11-05",
+            capabilities=ClientCapabilities(),
+            clientInfo=Implementation(name="test-client", version="0.1.0"),
+        )
+        await send_stream1.send(
+            JSONRPCMessage(
+                root=JSONRPCRequest(
+                    jsonrpc="2.0",
+                    id=1,
+                    method="initialize",
+                    params=TypeAdapter(InitializeRequestParams).dump_python(params),
+                )
+            )
+        )
+        response = await receive_stream2.receive()
+
+        # Send initialized notification
+        await send_stream1.send(
+            JSONRPCMessage(
+                root=JSONRPCNotification(
+                    jsonrpc="2.0",
+                    method="notifications/initialized",
+                )
+            )
+        )
+
+        # Call the tool to verify lifespan context
+        await send_stream1.send(
+            JSONRPCMessage(
+                root=JSONRPCRequest(
+                    jsonrpc="2.0",
+                    id=2,
+                    method="tools/call",
+                    params={"name": "check_lifespan", "arguments": {}},
+                )
+            )
+        )
+
+        # Get response and verify
+        response = await receive_stream2.receive()
+        assert response.root.result["content"][0]["text"] == "true"
+
+        # Cancel server task
+        tg.cancel_scope.cancel()
+
+
+@pytest.mark.anyio
+async def test_fastmcp_server_lifespan():
+    """Test that lifespan works in FastMCP server."""
+
+    @asynccontextmanager
+    async def test_lifespan(server: FastMCP) -> AsyncIterator[dict]:
+        """Test lifespan context that tracks startup/shutdown."""
+        context = {"started": False, "shutdown": False}
+        try:
+            context["started"] = True
+            yield context
+        finally:
+            context["shutdown"] = True
+
+    server = FastMCP("test", lifespan=test_lifespan)
+
+    # Create memory streams for testing
+    send_stream1, receive_stream1 = anyio.create_memory_object_stream(100)
+    send_stream2, receive_stream2 = anyio.create_memory_object_stream(100)
+
+    # Add a tool that checks lifespan context
+    @server.tool()
+    def check_lifespan(ctx: Context) -> bool:
+        """Tool that checks lifespan context."""
+        assert isinstance(ctx.request_context.lifespan_context, dict)
+        assert ctx.request_context.lifespan_context["started"]
+        assert not ctx.request_context.lifespan_context["shutdown"]
+        return True
+
+    # Run server in background task
+    async with anyio.create_task_group() as tg:
+
+        async def run_server():
+            await server._mcp_server.run(
+                receive_stream1,
+                send_stream2,
+                server._mcp_server.create_initialization_options(),
+                raise_exceptions=True,
+            )
+
+        tg.start_soon(run_server)
+
+        # Initialize the server
+        params = InitializeRequestParams(
+            protocolVersion="2024-11-05",
+            capabilities=ClientCapabilities(),
+            clientInfo=Implementation(name="test-client", version="0.1.0"),
+        )
+        await send_stream1.send(
+            JSONRPCMessage(
+                root=JSONRPCRequest(
+                    jsonrpc="2.0",
+                    id=1,
+                    method="initialize",
+                    params=TypeAdapter(InitializeRequestParams).dump_python(params),
+                )
+            )
+        )
+        response = await receive_stream2.receive()
+
+        # Send initialized notification
+        await send_stream1.send(
+            JSONRPCMessage(
+                root=JSONRPCNotification(
+                    jsonrpc="2.0",
+                    method="notifications/initialized",
+                )
+            )
+        )
+
+        # Call the tool to verify lifespan context
+        await send_stream1.send(
+            JSONRPCMessage(
+                root=JSONRPCRequest(
+                    jsonrpc="2.0",
+                    id=2,
+                    method="tools/call",
+                    params={"name": "check_lifespan", "arguments": {}},
+                )
+            )
+        )
+
+        # Get response and verify
+        response = await receive_stream2.receive()
+        assert response.root.result["content"][0]["text"] == "true"
+
+        # Cancel server task
+        tg.cancel_scope.cancel()