RuntimeError: Attempted to exit cancel scope in a different task than it was entered in

[npkanaka]

Description

When attempting to stop an SSE client connection created with mcp.client.sse.sse_client(), the application throws a RuntimeError related to AnyIO cancel scopes being exited in different tasks than they were entered in. This prevents proper application shutdown and can lead to resource leaks.

Steps to Reproduce

Set up an SSE client using the mcp.client.sse.sse_client() function
Establish a connection using the aenter() method
Attempt to close the connection using the aexit() method during application shutdown

The issue appears to be that the sse_client function in mcp.client.sse module creates AnyIO task groups or cancel scopes that are entered in one task but exited in another during cleanup. AnyIO requires that cancel scopes must be entered and exited in the same task context.
The problem is likely in the implementation of the async generator pattern in the sse_client function. When the connection is closed or an error occurs, the task group's __aexit__ method is being called in a different task than the one where __aenter__ was called, violating AnyIO's constraints.

Expected Behavior

The SSE connection should close cleanly without errors.

Actual Behavior

The application throws the following error:

RuntimeError: Attempted to exit cancel scope in a different task than it was entered in

Code Example

from mcp.client.sse import sse_client

async def connect_and_disconnect():
    # Set up the client
    streams_context = sse_client(url="http://example.com/sse", headers={"Authorization": "Bearer token"})
    streams = await streams_context.__aenter__()
    
    # Use the streams...
    
    # When shutting down - this throws the RuntimeError
    await streams_context.__aexit__(None, None, None)

Environment

• Python version: 3.10
• mcp package version: 1.6.0

[jingyugao]

i met the same problem

[npkanaka]

The issue appears to be that the sse_client function in mcp.client.sse module creates AnyIO task groups or cancel scopes that are entered in one task but exited in another during cleanup. AnyIO requires that cancel scopes must be entered and exited in the same task context.

I've used a different approach for managing the async resources that doesn't rely on AnyIO's cancel scopes.
My current setup creates clients for each request in the same task context. Used an async context manager pattern for proper lifecycle handling.

[LEON-REIN]

@npkanaka Hello! Do you mean we need to use async with to manage a mcp-session?

[christoph-ender]

Related:
#79

[npkanaka]

Just checking if there are any plans to address the RuntimeError when exiting AsyncExitStack across tasks.

[nigiva]

Just checking if there are any plans to address the RuntimeError when exiting AsyncExitStack across tasks.

@npkanaka As I explained here, I found a solution to close the connection from another task using an asyncio.Event signal.

Hope this helps ! ;)

[nickovs]

It's worth noting that this bug gets triggered when writing asynchronous pytest fixtures for testing MCP services. As a result it is much harder to write clear test code that properly cleans up resources when the tests are done.

Below is some minimal test code for testing a service; the tests themselves run just fine but pytest fails during the cleanup with this same RuntimeError. Note that the example below is using stdio rather than sse, so the bug appears to exist in both paths.

import asyncio
import pytest
import pytest_asyncio

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client


@pytest_asyncio.fixture(scope="class", loop_scope="class")
async def session_context():
    server_params = StdioServerParameters(command="mcp-file-reader")

    async with stdio_client(server_params) as stdio_transport:
        stdio, write = stdio_transport
        async with ClientSession(stdio, write) as session:
            await session.initialize()
            yield session

class TestMCPClientNew:
    loop: asyncio.AbstractEventLoop

    @pytest.mark.asyncio(loop_scope="class")
    async def test_list_tools(self, session_context):
        print("Testing list tools")
        response = await session_context.list_tools()
        print(response)
        tool_names = set(tool.name for tool in response.tools)
        expected_tools = {"list_allowed_directories", "read_file_content"}
        assert tool_names == expected_tools