Repository: modelcontextprotocol/python-sdk

Issue:
Random error thrown on response
**Describe the bug**
Sometimes, I see a stacktrace printed in the logs of my mcp server. Claude eventually succeeds to response but I think its good to investigate it.

**To Reproduce**
Its hard to reproduce as it does not always happen. The code in my codebase that caused it to happen is this one:

```
    def run_tool(self, args: dict) -> Sequence[TextContent | ImageContent | EmbeddedResource]:

        user_id = args.get(toolhandler.USER_ID_ARG)
        if not user_id:
            raise RuntimeError(f"Missing required argument: {toolhandler.USER_ID_ARG}")

        gmail_service = gmail.GmailService(user_id=user_id)
        query = args.get('query')
        max_results = args.get('max_results', 100)
        emails = gmail_service.query_emails(query=query, max_results=max_results)

        return [
            TextContent(
                type="text",
                text=json.dumps(emails, indent=2)
            )
        ]
```

Called in this context

```
@app.call_tool()
async def call_tool(name: str, arguments: Any) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
    """Handle tool calls for command line run."""
    
    if not isinstance(arguments, dict):
        raise RuntimeError("arguments must be dictionary")
    
    if toolhandler.USER_ID_ARG not in arguments:
        raise RuntimeError("user_id argument is missing in dictionary.")

    setup_oauth2(user_id=arguments.get(toolhandler.USER_ID_ARG, ""))

    tool_handler = get_tool_handler(name)
    if not tool_handler:
        raise ValueError(f"Unknown tool: {name}")

    try:
        return tool_handler.run_tool(arguments)
    except Exception as e:
        logger.error(str(e))
        raise RuntimeError(f"Caught Exception. Error: {str(e)}")
```

The RunTime Error is NOT thrown

**Expected behavior**
No stacktrace?

**Desktop (please complete the following information):**
- MAC OSX
- Python 3.13
- mcp 1.1.0

**Additional context**

Here is a full log
```
+ Exception Group Traceback (most recent call last):
  |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/.venv/lib/python3.13/site-packages/mcp/server/stdio.py", line 83, in stdio_server
  |     yield read_stream, write_stream
  |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/src/mcp_gsuite/server.py", line 161, in main
  |     await app.run(
  |     ...<3 lines>...
  |     )
  |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/.venv/lib/python3.13/site-packages/mcp/server/__init__.py", line 423, in run
  |     async with ServerSession(
  |                ~~~~~~~~~~~~~^
  |         read_stream, write_stream, initialization_options
  |         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |     ) as session:
  |     ^
  |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/.venv/lib/python3.13/site-packages/mcp/shared/session.py", line 122, in __aexit__
  |     return await self._task_group.__aexit__(exc_type, exc_val, exc_tb)
  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/.venv/lib/python3.13/site-packages/anyio/_backends/_asyncio.py", line 763, in __aexit__
  |     raise BaseExceptionGroup(
  |         "unhandled errors in a TaskGroup", self._exceptions
  |     )
  | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
  +-+---------------- 1 ----------------
    | Traceback (most recent call last):
    |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/.venv/lib/python3.13/site-packages/mcp/shared/session.py", line 235, in _receive_loop
    |     notification = self._receive_notification_type.model_validate(
    |         message.root.model_dump(
    |             by_alias=True, mode="json", exclude_none=True
    |         )
    |     )
    |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/.venv/lib/python3.13/site-packages/pydantic/main.py", line 627, in model_validate
    |     return cls.__pydantic_validator__.validate_python(
    |            ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
    |         obj, strict=strict, from_attributes=from_attributes, context=context
    |         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    |     )
    |     ^
    | pydantic_core._pydantic_core.ValidationError: 5 validation errors for ClientNotification
    | ProgressNotification.method
    |   Input should be 'notifications/progress' [type=literal_error, input_value='cancelled', input_type=str]
    |     For further information visit https://errors.pydantic.dev/2.10/v/literal_error
    | ProgressNotification.params.progressToken
    |   Field required [type=missing, input_value={'requestId': 20, 'reason... -2: Request timed out'}, input_type=dict]
    |     For further information visit https://errors.pydantic.dev/2.10/v/missing
    | ProgressNotification.params.progress
    |   Field required [type=missing, input_value={'requestId': 20, 'reason... -2: Request timed out'}, input_type=dict]
    |     For further information visit https://errors.pydantic.dev/2.10/v/missing
    | InitializedNotification.method
    |   Input should be 'notifications/initialized' [type=literal_error, input_value='cancelled', input_type=str]
    |     For further information visit https://errors.pydantic.dev/2.10/v/literal_error
    | RootsListChangedNotification.method
    |   Input should be 'notifications/roots/list_changed' [type=literal_error, input_value='cancelled', input_type=str]
    |     For further information visit https://errors.pydantic.dev/2.10/v/literal_error
    +------------------------------------

During handling of the above exception, another exception occurred:

  + Exception Group Traceback (most recent call last):
  |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/.venv/bin/mcp-gsuite", line 8, in <module>
  |     sys.exit(main())
  |              ~~~~^^
  |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/src/mcp_gsuite/__init__.py", line 6, in main
  |     asyncio.run(server.main())
  |     ~~~~~~~~~~~^^^^^^^^^^^^^^^
  |   File "/opt/homebrew/Cellar/python@3.13/3.13.0_1/Frameworks/Python.framework/Versions/3.13/lib/python3.13/asyncio/runners.py", line 194, in run
  |     return runner.run(main)
  |            ~~~~~~~~~~^^^^^^
  |   File "/opt/homebrew/Cellar/python@3.13/3.13.0_1/Frameworks/Python.framework/Versions/3.13/lib/python3.13/asyncio/runners.py", line 118, in run
  |     return self._loop.run_until_complete(task)
  |            ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^^^^^^
  |   File "/opt/homebrew/Cellar/python@3.13/3.13.0_1/Frameworks/Python.framework/Versions/3.13/lib/python3.13/asyncio/base_events.py", line 721, in run_until_complete
  |     return future.result()
  |            ~~~~~~~~~~~~~^^
  |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/src/mcp_gsuite/server.py", line 160, in main
  |     async with stdio_server() as (read_stream, write_stream):
  |                ~~~~~~~~~~~~^^
  |   File "/opt/homebrew/Cellar/python@3.13/3.13.0_1/Frameworks/Python.framework/Versions/3.13/lib/python3.13/contextlib.py", line 235, in __aexit__
  |     await self.gen.athrow(value)
  |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/.venv/lib/python3.13/site-packages/mcp/server/stdio.py", line 80, in stdio_server
  |     async with anyio.create_task_group() as tg:
  |                ~~~~~~~~~~~~~~~~~~~~~~~^^
  |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/.venv/lib/python3.13/site-packages/anyio/_backends/_asyncio.py", line 763, in __aexit__
  |     raise BaseExceptionGroup(
  |         "unhandled errors in a TaskGroup", self._exceptions
  |     )
  | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
  +-+---------------- 1 ----------------
    | Traceback (most recent call last):
    |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/.venv/lib/python3.13/site-packages/anyio/streams/memory.py", line 242, in send
    |     self.send_nowait(item)
    |     ~~~~~~~~~~~~~~~~^^^^^^
    |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/.venv/lib/python3.13/site-packages/anyio/streams/memory.py", line 225, in send_nowait
    |     raise WouldBlock
    | anyio.WouldBlock
    | 
    | During handling of the above exception, another exception occurred:
    | 
    | Traceback (most recent call last):
    |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/.venv/lib/python3.13/site-packages/mcp/server/__init__.py", line 467, in run
    |     await message.respond(
    |     ...<4 lines>...
    |     )
    |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/.venv/lib/python3.13/site-packages/mcp/shared/session.py", line 58, in respond
    |     await self._session._send_response(
    |         request_id=self.request_id, response=response
    |     )
    |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/.venv/lib/python3.13/site-packages/mcp/shared/session.py", line 196, in _send_response
    |     await self._write_stream.send(JSONRPCMessage(jsonrpc_error))
    |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/.venv/lib/python3.13/site-packages/anyio/streams/memory.py", line 248, in send
    |     await send_event.wait()
    |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/.venv/lib/python3.13/site-packages/anyio/_backends/_asyncio.py", line 1747, in wait
    |     await self._event.wait()
    |   File "/opt/homebrew/Cellar/python@3.13/3.13.0_1/Frameworks/Python.framework/Versions/3.13/lib/python3.13/asyncio/locks.py", line 213, in wait
    |     await fut
    | asyncio.exceptions.CancelledError: Cancelled by cancel scope 1040d7c50
    | 
    | During handling of the above exception, another exception occurred:
    | 
    | Exception Group Traceback (most recent call last):
    |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/.venv/lib/python3.13/site-packages/mcp/server/stdio.py", line 83, in stdio_server
    |     yield read_stream, write_stream
    |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/src/mcp_gsuite/server.py", line 161, in main
    |     await app.run(
    |     ...<3 lines>...
    |     )
    |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/.venv/lib/python3.13/site-packages/mcp/server/__init__.py", line 423, in run
    |     async with ServerSession(
    |                ~~~~~~~~~~~~~^
    |         read_stream, write_stream, initialization_options
    |         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    |     ) as session:
    |     ^
    |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/.venv/lib/python3.13/site-packages/mcp/shared/session.py", line 122, in __aexit__
    |     return await self._task_group.__aexit__(exc_type, exc_val, exc_tb)
    |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/.venv/lib/python3.13/site-packages/anyio/_backends/_asyncio.py", line 763, in __aexit__
    |     raise BaseExceptionGroup(
    |         "unhandled errors in a TaskGroup", self._exceptions
    |     )
    | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
    +-+---------------- 1 ----------------
      | Traceback (most recent call last):
      |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/.venv/lib/python3.13/site-packages/mcp/shared/session.py", line 235, in _receive_loop
      |     notification = self._receive_notification_type.model_validate(
      |         message.root.model_dump(
      |             by_alias=True, mode="json", exclude_none=True
      |         )
      |     )
      |   File "/Users/markus/experiments/claude-mvp/mcp-gsuite/.venv/lib/python3.13/site-packages/pydantic/main.py", line 627, in model_validate
      |     return cls.__pydantic_validator__.validate_python(
      |            ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
      |         obj, strict=strict, from_attributes=from_attributes, context=context
      |         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
      |     )
      |     ^
      | pydantic_core._pydantic_core.ValidationError: 5 validation errors for ClientNotification
      | ProgressNotification.method
      |   Input should be 'notifications/progress' [type=literal_error, input_value='cancelled', input_type=str]
      |     For further information visit https://errors.pydantic.dev/2.10/v/literal_error
      | ProgressNotification.params.progressToken
      |   Field required [type=missing, input_value={'requestId': 20, 'reason... -2: Request timed out'}, input_type=dict]
      |     For further information visit https://errors.pydantic.dev/2.10/v/missing
      | ProgressNotification.params.progress
      |   Field required [type=missing, input_value={'requestId': 20, 'reason... -2: Request timed out'}, input_type=dict]
      |     For further information visit https://errors.pydantic.dev/2.10/v/missing
      | InitializedNotification.method
      |   Input should be 'notifications/initialized' [type=literal_error, input_value='cancelled', input_type=str]
      |     For further information visit https://errors.pydantic.dev/2.10/v/literal_error
      | RootsListChangedNotification.method
      |   Input should be 'notifications/roots/list_changed' [type=literal_error, input_value='cancelled', input_type=str]
      |     For further information visit https://errors.pydantic.dev/2.10/v/literal_error
```


Discussion:
EduPonz: This doesn't seem to solve #88 for me. I can reproduce triggering a call tool from claude with a call tool handle that simply waits for 200 s and returns a valid message. Clause receives the response but the server crashes

```python
@server.call_tool()
async def handle_call_tool(
    name: str, arguments: dict | None,
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    await asyncio.sleep(200)
    response = {"msg": "success"}
    return [types.TextContent(type="text", text=f"{response}")]
```

Patch:
diff --git a/.gitignore b/.gitignore
index bb25f9e..f27f895 100644
--- a/.gitignore
+++ b/.gitignore
@@ -162,6 +162,3 @@ cython_debug/
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 #.idea/
-
-# vscode
-.vscode/
\ No newline at end of file
diff --git a/src/mcp/client/session.py b/src/mcp/client/session.py
index 4858ede..27ca74d 100644
--- a/src/mcp/client/session.py
+++ b/src/mcp/client/session.py
@@ -120,17 +120,6 @@ class ClientSession(
             types.ListResourcesResult,
         )
 
-    async def list_resource_templates(self) -> types.ListResourceTemplatesResult:
-        """Send a resources/templates/list request."""
-        return await self.send_request(
-            types.ClientRequest(
-                types.ListResourceTemplatesRequest(
-                    method="resources/templates/list",
-                )
-            ),
-            types.ListResourceTemplatesResult,
-        )
-
     async def read_resource(self, uri: AnyUrl) -> types.ReadResourceResult:
         """Send a resources/read request."""
         return await self.send_request(
diff --git a/src/mcp/server/fastmcp/server.py b/src/mcp/server/fastmcp/server.py
index aa7c79b..06b16e0 100644
--- a/src/mcp/server/fastmcp/server.py
+++ b/src/mcp/server/fastmcp/server.py
@@ -25,7 +25,6 @@ from mcp.server.sse import SseServerTransport
 from mcp.server.stdio import stdio_server
 from mcp.shared.context import RequestContext
 from mcp.types import (
-    AnyFunction,
     EmbeddedResource,
     GetPromptResult,
     ImageContent,
@@ -166,7 +165,7 @@ class FastMCP:
         return Context(request_context=request_context, fastmcp=self)
 
     async def call_tool(
-        self, name: str, arguments: dict[str, Any]
+        self, name: str, arguments: dict
     ) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
         """Call a tool by name with arguments."""
         context = self.get_context()
@@ -215,7 +214,7 @@ class FastMCP:
 
     def add_tool(
         self,
-        fn: AnyFunction,
+        fn: Callable,
         name: str | None = None,
         description: str | None = None,
     ) -> None:
@@ -231,9 +230,7 @@ class FastMCP:
         """
         self._tool_manager.add_tool(fn, name=name, description=description)
 
-    def tool(
-        self, name: str | None = None, description: str | None = None
-    ) -> Callable[[AnyFunction], AnyFunction]:
+    def tool(self, name: str | None = None, description: str | None = None) -> Callable:
         """Decorator to register a tool.
 
         Tools can optionally request a Context object by adding a parameter with the
@@ -266,7 +263,7 @@ class FastMCP:
                 "Did you forget to call it? Use @tool() instead of @tool"
             )
 
-        def decorator(fn: AnyFunction) -> AnyFunction:
+        def decorator(fn: Callable) -> Callable:
             self.add_tool(fn, name=name, description=description)
             return fn
 
@@ -287,7 +284,7 @@ class FastMCP:
         name: str | None = None,
         description: str | None = None,
         mime_type: str | None = None,
-    ) -> Callable[[AnyFunction], AnyFunction]:
+    ) -> Callable:
         """Decorator to register a function as a resource.
 
         The function will be called when the resource is read to generate its content.
@@ -331,7 +328,7 @@ class FastMCP:
                 "Did you forget to call it? Use @resource('uri') instead of @resource"
             )
 
-        def decorator(fn: AnyFunction) -> AnyFunction:
+        def decorator(fn: Callable) -> Callable:
             # Check if this should be a template
             has_uri_params = "{" in uri and "}" in uri
             has_func_params = bool(inspect.signature(fn).parameters)
@@ -379,7 +376,7 @@ class FastMCP:
 
     def prompt(
         self, name: str | None = None, description: str | None = None
-    ) -> Callable[[AnyFunction], AnyFunction]:
+    ) -> Callable:
         """Decorator to register a prompt.
 
         Args:
@@ -420,7 +417,7 @@ class FastMCP:
                 "Did you forget to call it? Use @prompt() instead of @prompt"
             )
 
-        def decorator(func: AnyFunction) -> AnyFunction:
+        def decorator(func: Callable) -> Callable:
             prompt = Prompt.from_function(func, name=name, description=description)
             self.add_prompt(prompt)
             return func
diff --git a/src/mcp/server/lowlevel/server.py b/src/mcp/server/lowlevel/server.py
index 13d4fd9..3d91722 100644
--- a/src/mcp/server/lowlevel/server.py
+++ b/src/mcp/server/lowlevel/server.py
@@ -453,10 +453,15 @@ class Server:
                     logger.debug(f"Received message: {message}")
 
                     match message:
-                        case RequestResponder(request=types.ClientRequest(root=req)):
-                            await self._handle_request(
-                                message, req, session, raise_exceptions
-                            )
+                        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)
 
diff --git a/src/mcp/server/session.py b/src/mcp/server/session.py
index b71b372..d918b98 100644
--- a/src/mcp/server/session.py
+++ b/src/mcp/server/session.py
@@ -126,19 +126,20 @@ class ServerSession(
             case types.InitializeRequest(params=params):
                 self._initialization_state = InitializationState.Initializing
                 self._client_params = params
-                await responder.respond(
-                    types.ServerResult(
-                        types.InitializeResult(
-                            protocolVersion=types.LATEST_PROTOCOL_VERSION,
-                            capabilities=self._init_options.capabilities,
-                            serverInfo=types.Implementation(
-                                name=self._init_options.server_name,
-                                version=self._init_options.server_version,
-                            ),
-                            instructions=self._init_options.instructions,
+                with responder:
+                    await responder.respond(
+                        types.ServerResult(
+                            types.InitializeResult(
+                                protocolVersion=types.LATEST_PROTOCOL_VERSION,
+                                capabilities=self._init_options.capabilities,
+                                serverInfo=types.Implementation(
+                                    name=self._init_options.server_name,
+                                    version=self._init_options.server_version,
+                                ),
+                                instructions=self._init_options.instructions,
+                            )
                         )
                     )
-                )
             case _:
                 if self._initialization_state != InitializationState.Initialized:
                     raise RuntimeError(
diff --git a/src/mcp/shared/session.py b/src/mcp/shared/session.py
index 5e114ec..3d3988c 100644
--- a/src/mcp/shared/session.py
+++ b/src/mcp/shared/session.py
@@ -1,6 +1,7 @@
+import logging
 from contextlib import AbstractAsyncContextManager
 from datetime import timedelta
-from typing import Generic, TypeVar
+from typing import Any, Callable, Generic, TypeVar
 
 import anyio
 import anyio.lowlevel
@@ -10,6 +11,7 @@ from pydantic import BaseModel
 
 from mcp.shared.exceptions import McpError
 from mcp.types import (
+    CancelledNotification,
     ClientNotification,
     ClientRequest,
     ClientResult,
@@ -38,27 +40,98 @@ RequestId = str | int
 
 
 class RequestResponder(Generic[ReceiveRequestT, SendResultT]):
+    """Handles responding to MCP requests and manages request lifecycle.
+
+    This class MUST be used as a context manager to ensure proper cleanup and
+    cancellation handling:
+
+    Example:
+        with request_responder as resp:
+            await resp.respond(result)
+
+    The context manager ensures:
+    1. Proper cancellation scope setup and cleanup
+    2. Request completion tracking
+    3. Cleanup of in-flight requests
+    """
+
     def __init__(
         self,
         request_id: RequestId,
         request_meta: RequestParams.Meta | None,
         request: ReceiveRequestT,
         session: "BaseSession",
+        on_complete: Callable[["RequestResponder[ReceiveRequestT, SendResultT]"], Any],
     ) -> None:
         self.request_id = request_id
         self.request_meta = request_meta
         self.request = request
         self._session = session
-        self._responded = False
+        self._completed = False
+        self._cancel_scope = anyio.CancelScope()
+        self._on_complete = on_complete
+        self._entered = False  # Track if we're in a context manager
+
+    def __enter__(self) -> "RequestResponder[ReceiveRequestT, SendResultT]":
+        """Enter the context manager, enabling request cancellation tracking."""
+        self._entered = True
+        self._cancel_scope = anyio.CancelScope()
+        self._cancel_scope.__enter__()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
+        """Exit the context manager, performing cleanup and notifying completion."""
+        try:
+            if self._completed:
+                self._on_complete(self)
+        finally:
+            self._entered = False
+            if not self._cancel_scope:
+                raise RuntimeError("No active cancel scope")
+            self._cancel_scope.__exit__(exc_type, exc_val, exc_tb)
 
     async def respond(self, response: SendResultT | ErrorData) -> None:
-        assert not self._responded, "Request already responded to"
-        self._responded = True
+        """Send a response for this request.
+
+        Must be called within a context manager block.
+        Raises:
+            RuntimeError: If not used within a context manager
+            AssertionError: If request was already responded to
+        """
+        if not self._entered:
+            raise RuntimeError("RequestResponder must be used as a context manager")
+        assert not self._completed, "Request already responded to"
+
+        if not self.cancelled:
+            self._completed = True
+
+            await self._session._send_response(
+                request_id=self.request_id, response=response
+            )
+
+    async def cancel(self) -> None:
+        """Cancel this request and mark it as completed."""
+        if not self._entered:
+            raise RuntimeError("RequestResponder must be used as a context manager")
+        if not self._cancel_scope:
+            raise RuntimeError("No active cancel scope")
 
+        self._cancel_scope.cancel()
+        self._completed = True  # Mark as completed so it's removed from in_flight
+        # Send an error response to indicate cancellation
         await self._session._send_response(
-            request_id=self.request_id, response=response
+            request_id=self.request_id,
+            response=ErrorData(code=0, message="Request cancelled", data=None),
         )
 
+    @property
+    def in_flight(self) -> bool:
+        return not self._completed and not self.cancelled
+
+    @property
+    def cancelled(self) -> bool:
+        return self._cancel_scope is not None and self._cancel_scope.cancel_called
+
 
 class BaseSession(
     AbstractAsyncContextManager,
@@ -82,6 +155,7 @@ class BaseSession(
         RequestId, MemoryObjectSendStream[JSONRPCResponse | JSONRPCError]
     ]
     _request_id: int
+    _in_flight: dict[RequestId, RequestResponder[ReceiveRequestT, SendResultT]]
 
     def __init__(
         self,
@@ -99,6 +173,7 @@ class BaseSession(
         self._receive_request_type = receive_request_type
         self._receive_notification_type = receive_notification_type
         self._read_timeout_seconds = read_timeout_seconds
+        self._in_flight = {}
 
         self._incoming_message_stream_writer, self._incoming_message_stream_reader = (
             anyio.create_memory_object_stream[
@@ -219,6 +294,7 @@ class BaseSession(
                             by_alias=True, mode="json", exclude_none=True
                         )
                     )
+
                     responder = RequestResponder(
                         request_id=message.root.id,
                         request_meta=validated_request.root.params.meta
@@ -226,20 +302,37 @@ class BaseSession(
                         else None,
                         request=validated_request,
                         session=self,
+                        on_complete=lambda r: self._in_flight.pop(r.request_id, None),
                     )
 
+                    self._in_flight[responder.request_id] = responder
                     await self._received_request(responder)
-                    if not responder._responded:
+                    if not responder._completed:
                         await self._incoming_message_stream_writer.send(responder)
+
                 elif isinstance(message.root, JSONRPCNotification):
-                    notification = self._receive_notification_type.model_validate(
-                        message.root.model_dump(
-                            by_alias=True, mode="json", exclude_none=True
+                    try:
+                        notification = self._receive_notification_type.model_validate(
+                            message.root.model_dump(
+                                by_alias=True, mode="json", exclude_none=True
+                            )
+                        )
+                        # Handle cancellation notifications
+                        if isinstance(notification.root, CancelledNotification):
+                            cancelled_id = notification.root.params.requestId
+                            if cancelled_id in self._in_flight:
+                                await self._in_flight[cancelled_id].cancel()
+                        else:
+                            await self._received_notification(notification)
+                            await self._incoming_message_stream_writer.send(
+                                notification
+                            )
+                    except Exception as e:
+                        # For other validation errors, log and continue
+                        logging.warning(
+                            f"Failed to validate notification: {e}. "
+                            f"Message was: {message.root}"
                         )
-                    )
-
-                    await self._received_notification(notification)
-                    await self._incoming_message_stream_writer.send(notification)
                 else:  # Response or error
                     stream = self._response_streams.pop(message.root.id, None)
                     if stream:
diff --git a/src/mcp/types.py b/src/mcp/types.py
index d1157aa..a4fa254 100644
--- a/src/mcp/types.py
+++ b/src/mcp/types.py
@@ -1,4 +1,4 @@
-from typing import Annotated, Any, Callable, Generic, Literal, TypeAlias, TypeVar
+from typing import Annotated, Any, Generic, Literal, TypeVar
 
 from pydantic import BaseModel, ConfigDict, Field, FileUrl, RootModel
 from pydantic.networks import AnyUrl
@@ -27,7 +27,6 @@ ProgressToken = str | int
 Cursor = str
 Role = Literal["user", "assistant"]
 RequestId = str | int
-AnyFunction: TypeAlias = Callable[..., Any]
 
 
 class RequestParams(BaseModel):


Test patch:
diff --git a/tests/issues/test_141_resource_templates.py b/tests/issues/test_141_resource_templates.py
deleted file mode 100644
index d6526e9..0000000
--- a/tests/issues/test_141_resource_templates.py
+++ /dev/null
@@ -1,118 +0,0 @@
-import pytest
-from pydantic import AnyUrl
-
-from mcp.server.fastmcp import FastMCP
-from mcp.shared.memory import (
-    create_connected_server_and_client_session as client_session,
-)
-from mcp.types import (
-    ListResourceTemplatesResult,
-    TextResourceContents,
-)
-
-
-@pytest.mark.anyio
-async def test_resource_template_edge_cases():
-    """Test server-side resource template validation"""
-    mcp = FastMCP("Demo")
-
-    # Test case 1: Template with multiple parameters
-    @mcp.resource("resource://users/{user_id}/posts/{post_id}")
-    def get_user_post(user_id: str, post_id: str) -> str:
-        return f"Post {post_id} by user {user_id}"
-
-    # Test case 2: Template with optional parameter (should fail)
-    with pytest.raises(ValueError, match="Mismatch between URI parameters"):
-
-        @mcp.resource("resource://users/{user_id}/profile")
-        def get_user_profile(user_id: str, optional_param: str | None = None) -> str:
-            return f"Profile for user {user_id}"
-
-    # Test case 3: Template with mismatched parameters
-    with pytest.raises(ValueError, match="Mismatch between URI parameters"):
-
-        @mcp.resource("resource://users/{user_id}/profile")
-        def get_user_profile_mismatch(different_param: str) -> str:
-            return f"Profile for user {different_param}"
-
-    # Test case 4: Template with extra function parameters
-    with pytest.raises(ValueError, match="Mismatch between URI parameters"):
-
-        @mcp.resource("resource://users/{user_id}/profile")
-        def get_user_profile_extra(user_id: str, extra_param: str) -> str:
-            return f"Profile for user {user_id}"
-
-    # Test case 5: Template with missing function parameters
-    with pytest.raises(ValueError, match="Mismatch between URI parameters"):
-
-        @mcp.resource("resource://users/{user_id}/profile/{section}")
-        def get_user_profile_missing(user_id: str) -> str:
-            return f"Profile for user {user_id}"
-
-    # Verify valid template works
-    result = await mcp.read_resource("resource://users/123/posts/456")
-    assert result.content == "Post 456 by user 123"
-    assert result.mime_type == "text/plain"
-
-    # Verify invalid parameters raise error
-    with pytest.raises(ValueError, match="Unknown resource"):
-        await mcp.read_resource("resource://users/123/posts")  # Missing post_id
-
-    with pytest.raises(ValueError, match="Unknown resource"):
-        await mcp.read_resource(
-            "resource://users/123/posts/456/extra"
-        )  # Extra path component
-
-
-@pytest.mark.anyio
-async def test_resource_template_client_interaction():
-    """Test client-side resource template interaction"""
-    mcp = FastMCP("Demo")
-
-    # Register some templated resources
-    @mcp.resource("resource://users/{user_id}/posts/{post_id}")
-    def get_user_post(user_id: str, post_id: str) -> str:
-        return f"Post {post_id} by user {user_id}"
-
-    @mcp.resource("resource://users/{user_id}/profile")
-    def get_user_profile(user_id: str) -> str:
-        return f"Profile for user {user_id}"
-
-    async with client_session(mcp._mcp_server) as session:
-        # Initialize the session
-        await session.initialize()
-
-        # List available resources
-        resources = await session.list_resource_templates()
-        assert isinstance(resources, ListResourceTemplatesResult)
-        assert len(resources.resourceTemplates) == 2
-
-        # Verify resource templates are listed correctly
-        templates = [r.uriTemplate for r in resources.resourceTemplates]
-        assert "resource://users/{user_id}/posts/{post_id}" in templates
-        assert "resource://users/{user_id}/profile" in templates
-
-        # Read a resource with valid parameters
-        result = await session.read_resource(AnyUrl("resource://users/123/posts/456"))
-        contents = result.contents[0]
-        assert isinstance(contents, TextResourceContents)
-        assert contents.text == "Post 456 by user 123"
-        assert contents.mimeType == "text/plain"
-
-        # Read another resource with valid parameters
-        result = await session.read_resource(AnyUrl("resource://users/789/profile"))
-        contents = result.contents[0]
-        assert isinstance(contents, TextResourceContents)
-        assert contents.text == "Profile for user 789"
-        assert contents.mimeType == "text/plain"
-
-        # Verify invalid resource URIs raise appropriate errors
-        with pytest.raises(Exception):  # Specific exception type may vary
-            await session.read_resource(
-                AnyUrl("resource://users/123/posts")
-            )  # Missing post_id
-
-        with pytest.raises(Exception):  # Specific exception type may vary
-            await session.read_resource(
-                AnyUrl("resource://users/123/invalid")
-            )  # Invalid template
diff --git a/tests/issues/test_88_random_error.py b/tests/issues/test_88_random_error.py
new file mode 100644
index 0000000..8609c20
--- /dev/null
+++ b/tests/issues/test_88_random_error.py
@@ -0,0 +1,111 @@
+"""Test to reproduce issue #88: Random error thrown on response."""
+
+from datetime import timedelta
+from pathlib import Path
+from typing import Sequence
+
+import anyio
+import pytest
+
+from mcp.client.session import ClientSession
+from mcp.server.lowlevel import Server
+from mcp.shared.exceptions import McpError
+from mcp.types import (
+    EmbeddedResource,
+    ImageContent,
+    TextContent,
+)
+
+
+@pytest.mark.anyio
+async def test_notification_validation_error(tmp_path: Path):
+    """Test that timeouts are handled gracefully and don't break the server.
+
+    This test verifies that when a client request times out:
+    1. The server task stays alive
+    2. The server can still handle new requests
+    3. The client can make new requests
+    4. No resources are leaked
+    """
+
+    server = Server(name="test")
+    request_count = 0
+    slow_request_started = anyio.Event()
+    slow_request_complete = anyio.Event()
+
+    @server.call_tool()
+    async def slow_tool(
+        name: str, arg
+    ) -> Sequence[TextContent | ImageContent | EmbeddedResource]:
+        nonlocal request_count
+        request_count += 1
+
+        if name == "slow":
+            # Signal that slow request has started
+            slow_request_started.set()
+            # Long enough to ensure timeout
+            await anyio.sleep(0.2)
+            # Signal completion
+            slow_request_complete.set()
+            return [TextContent(type="text", text=f"slow {request_count}")]
+        elif name == "fast":
+            # Fast enough to complete before timeout
+            await anyio.sleep(0.01)
+            return [TextContent(type="text", text=f"fast {request_count}")]
+        return [TextContent(type="text", text=f"unknown {request_count}")]
+
+    async def server_handler(read_stream, write_stream):
+        await server.run(
+            read_stream,
+            write_stream,
+            server.create_initialization_options(),
+            raise_exceptions=True,
+        )
+
+    async def client(read_stream, write_stream):
+        # Use a timeout that's:
+        # - Long enough for fast operations (>10ms)
+        # - Short enough for slow operations (<200ms)
+        # - Not too short to avoid flakiness
+        async with ClientSession(
+            read_stream, write_stream, read_timeout_seconds=timedelta(milliseconds=50)
+        ) as session:
+            await session.initialize()
+
+            # First call should work (fast operation)
+            result = await session.call_tool("fast")
+            assert result.content == [TextContent(type="text", text="fast 1")]
+            assert not slow_request_complete.is_set()
+
+            # Second call should timeout (slow operation)
+            with pytest.raises(McpError) as exc_info:
+                await session.call_tool("slow")
+            assert "Timed out while waiting" in str(exc_info.value)
+
+            # Wait for slow request to complete in the background
+            with anyio.fail_after(1):  # Timeout after 1 second
+                await slow_request_complete.wait()
+
+            # Third call should work (fast operation),
+            # proving server is still responsive
+            result = await session.call_tool("fast")
+            assert result.content == [TextContent(type="text", text="fast 3")]
+
+    # Run server and client in separate task groups to avoid cancellation
+    server_writer, server_reader = anyio.create_memory_object_stream(1)
+    client_writer, client_reader = anyio.create_memory_object_stream(1)
+
+    server_ready = anyio.Event()
+
+    async def wrapped_server_handler(read_stream, write_stream):
+        server_ready.set()
+        await server_handler(read_stream, write_stream)
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(wrapped_server_handler, server_reader, client_writer)
+        # Wait for server to start and initialize
+        with anyio.fail_after(1):  # Timeout after 1 second
+            await server_ready.wait()
+        # Run client in a separate task to avoid cancellation
+        async with anyio.create_task_group() as client_tg:
+            client_tg.start_soon(client, client_reader, server_writer)
diff --git a/tests/shared/test_session.py b/tests/shared/test_session.py
new file mode 100644
index 0000000..65cf061
--- /dev/null
+++ b/tests/shared/test_session.py
@@ -0,0 +1,126 @@
+from typing import AsyncGenerator
+
+import anyio
+import pytest
+
+import mcp.types as types
+from mcp.client.session import ClientSession
+from mcp.server.lowlevel.server import Server
+from mcp.shared.exceptions import McpError
+from mcp.shared.memory import create_connected_server_and_client_session
+from mcp.types import (
+    CancelledNotification,
+    CancelledNotificationParams,
+    ClientNotification,
+    ClientRequest,
+    EmptyResult,
+)
+
+
+@pytest.fixture
+def mcp_server() -> Server:
+    return Server(name="test server")
+
+
+@pytest.fixture
+async def client_connected_to_server(
+    mcp_server: Server,
+) -> AsyncGenerator[ClientSession, None]:
+    async with create_connected_server_and_client_session(mcp_server) as client_session:
+        yield client_session
+
+
+@pytest.mark.anyio
+async def test_in_flight_requests_cleared_after_completion(
+    client_connected_to_server: ClientSession,
+):
+    """Verify that _in_flight is empty after all requests complete."""
+    # Send a request and wait for response
+    response = await client_connected_to_server.send_ping()
+    assert isinstance(response, EmptyResult)
+
+    # Verify _in_flight is empty
+    assert len(client_connected_to_server._in_flight) == 0
+
+
+@pytest.mark.anyio
+async def test_request_cancellation():
+    """Test that requests can be cancelled while in-flight."""
+    # The tool is already registered in the fixture
+
+    ev_tool_called = anyio.Event()
+    ev_cancelled = anyio.Event()
+    request_id = None
+
+    # Start the request in a separate task so we can cancel it
+    def make_server() -> Server:
+        server = Server(name="TestSessionServer")
+
+        # Register the tool handler
+        @server.call_tool()
+        async def handle_call_tool(name: str, arguments: dict | None) -> list:
+            nonlocal request_id, ev_tool_called
+            if name == "slow_tool":
+                request_id = server.request_context.request_id
+                ev_tool_called.set()
+                await anyio.sleep(10)  # Long enough to ensure we can cancel
+                return []
+            raise ValueError(f"Unknown tool: {name}")
+
+        # Register the tool so it shows up in list_tools
+        @server.list_tools()
+        async def handle_list_tools() -> list[types.Tool]:
+            return [
+                types.Tool(
+                    name="slow_tool",
+                    description="A slow tool that takes 10 seconds to complete",
+                    inputSchema={},
+                )
+            ]
+
+        return server
+
+    async def make_request(client_session):
+        nonlocal ev_cancelled
+        try:
+            await client_session.send_request(
+                ClientRequest(
+                    types.CallToolRequest(
+                        method="tools/call",
+                        params=types.CallToolRequestParams(
+                            name="slow_tool", arguments={}
+                        ),
+                    )
+                ),
+                types.CallToolResult,
+            )
+            pytest.fail("Request should have been cancelled")
+        except McpError as e:
+            # Expected - request was cancelled
+            assert "Request cancelled" in str(e)
+            ev_cancelled.set()
+
+    async with create_connected_server_and_client_session(
+        make_server()
+    ) as client_session:
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(make_request, client_session)
+
+            # Wait for the request to be in-flight
+            with anyio.fail_after(1):  # Timeout after 1 second
+                await ev_tool_called.wait()
+
+            # Send cancellation notification
+            assert request_id is not None
+            await client_session.send_notification(
+                ClientNotification(
+                    CancelledNotification(
+                        method="notifications/cancelled",
+                        params=CancelledNotificationParams(requestId=request_id),
+                    )
+                )
+            )
+
+            # Give cancellation time to process
+            with anyio.fail_after(1):
+                await ev_cancelled.wait()