MCP cleanup exits AnyIO cancel scopes from a different asyncio task

[hertznsk]

Summary

MCPManager keeps MCP stdio and session contexts in one shared AsyncExitStack. A connection may be opened from a parallel-agent or for-each worker task, while provider shutdown later closes that stack from the workflow/root task.

AnyIO cancel scopes are task-affine, so cleanup can fail with:

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

The exception is currently caught and logged by MCPManager.close(), which means shutdown continues but MCP sessions/processes may not be cleaned up correctly.

Public main references

Verified against origin/main at c8fad5c95fdae0e748455041ed1eca706329989d:

• The manager owns one shared AsyncExitStack:

  conductor/src/conductor/mcp/manager.py

  Lines 122 to 127 in c8fad5c

  	self.sessions: dict[str, ClientSession] = {}
  	self.tools: dict[str, list[dict[str, Any]]] = {} # server -> tools
  	self.tool_to_server: dict[str, str] = {} # prefixed_name -> server
  	self._exit_stack = AsyncExitStack()
  	self._initialized = False
  	self._tool_output = tool_output or ToolOutputConfig()

• connect_server() enters both AnyIO-backed contexts in the calling task:

  conductor/src/conductor/mcp/manager.py

  Lines 177 to 186 in c8fad5c

  	try:
  	# Enter the stdio_client context
  	transport = await self._exit_stack.enter_async_context(stdio_client(server_params))
  	read_stream, write_stream = transport
  	# Create and initialize session
  	session = await self._exit_stack.enter_async_context(
  	ClientSession(read_stream, write_stream)
  	)
  	await session.initialize()

• close() exits the stack in whichever task performs provider cleanup:

  conductor/src/conductor/mcp/manager.py

  Lines 494 to 508 in c8fad5c

  	async def close(self) -> None:
  	"""Close all server connections and clean up resources.
  	This method should be called when the manager is no longer needed.
  	It properly closes all stdio connections and cleans up internal state.
  	"""
  	if not self._initialized:
  	return
  	logger.debug(f"Closing {len(self.sessions)} MCP server connection(s)")
  	try:
  	await self._exit_stack.aclose()
  	except Exception as e:
  	logger.warning(f"Error closing MCP connections: {e}")

• Claude pools managers used by concurrent agents:

  conductor/src/conductor/providers/claude.py

  Lines 627 to 634 in c8fad5c

  	async def _get_mcp_manager_for_cwd(self, resolved_cwd: str) -> MCPManager | None:
  	"""Return the pooled MCPManager for ``resolved_cwd``, connecting on first use.
  	Each distinct working directory gets its own MCPManager so stdio MCP
  	servers are spawned with that directory as their ``cwd``. The lazy
  	connect is guarded by a per-cwd ``asyncio.Lock`` so parallel agents
  	resolving the same cwd observe exactly one manager (no duplicate
  	spawns), while agents with different cwds proceed concurrently.

• Provider shutdown closes those managers later:

  conductor/src/conductor/providers/claude.py

  Lines 755 to 770 in c8fad5c

  	async def close(self) -> None:
  	"""Release provider resources and close connections.
  	Shuts down every pooled MCPManager (one per distinct working
  	directory). Idempotent: a second call is a no-op.
  	"""
  	# Close MCP connections first (all pool entries).
  	if self._mcp_managers:
  	for cwd, manager in self._mcp_managers.items():
  	try:
  	await manager.close()
  	except Exception as e:
  	logger.warning(f"Error closing MCP manager for cwd={cwd}: {e}")
  	self._mcp_managers.clear()
  	self._mcp_manager_locks.clear()
  	logger.debug("All pooled MCP managers closed")

Minimal reproduction of the lifecycle pattern

This does not require an API call or Pydantic AI; it isolates the same task-ownership mismatch:

import asyncio
from contextlib import AsyncExitStack, asynccontextmanager

import anyio


@asynccontextmanager
async def task_bound_context():
    with anyio.CancelScope():
        yield


async def main():
    stack = AsyncExitStack()

    async def connect():
        await stack.enter_async_context(task_bound_context())

    await asyncio.create_task(connect())
    await stack.aclose()


asyncio.run(main())

Observed with Python 3.12.13 and AnyIO 4.12.1:

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

In Conductor, parallel and for-each execution provide the worker-task side of this pattern; provider/registry cleanup provides the root-task side.

Expected behavior

MCP connections should close cleanly regardless of which workflow task first requested the manager, including parallel and for-each runs. No task-affinity exception should be logged, and all sessions/transports/processes should be released.

Suggested direction

Give each MCPManager a persistent lifecycle-owner task and marshal both context entry and context exit through that same task. Regression coverage should include:

1. connect from a worker task, close from its parent task;
2. failed connection followed by close;
3. cancelled connection followed by close;
4. multiple pooled managers created by parallel agents.

Closing only from the original caller is insufficient because that task may already have completed by provider shutdown.