Potential MCP issue (asyncio.exceptions.CancelledError)

[Eddydpyl]

Question

I've been having a recurring error, where running an Agent with MCP toolsets causes it to error, but it does not happen all the time. It happens very frequently when the Agent is run while another is already running, and they both use the same MCP toolset. That said, it also happens without that being the case, just less frequently.

Here's a stacktrace. Another thing that's suspicious, is that the Agent run is being called when another Agent's run is ending (in this particular case). Maybe one is closing the MCP connection, while the other is opening it, and that causes this issue?

Exception Group Traceback (most recent call last):
  |   File "/app/.venv/lib/python3.12/site-packages/opentelemetry/trace/__init__.py", line 589, in use_span
  |     yield span
  |   File "/app/.venv/lib/python3.12/site-packages/pydantic_graph/graph.py", line 259, in iter
  |     yield GraphRun[StateT, DepsT, RunEndT](
  |   File "/app/.venv/lib/python3.12/site-packages/pydantic_ai/agent/__init__.py", line 659, in iter
  |     yield agent_run
  |   File "/app/ocean/hooked/agent.py", line 258, in iter
  |     yield self._last_run
  |   File "/app/.venv/lib/python3.12/site-packages/pydantic_ai/agent/abstract.py", line 222, in run
  |     async for node in agent_run:
  |   File "/app/ocean/hooked/agent.py", line 366, in __anext__
  |     node = await self._apply_end_hooks(node)
  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/app/ocean/hooked/agent.py", line 429, in _apply_end_hooks
  |     node = await hook.end_hook(ctx, node)
  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/app/ocean/hooked/hooks/final_check.py", line 79, in end_hook
  |     return await self._validate_attachments(ctx, node)
  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/app/ocean/hooked/hooks/final_check.py", line 120, in _validate_attachments
  |     validate_result = await validate_attachments(abridge_result.output, last_call.tool_args.get("attachments"), self.base_dir)
  |                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/app/ocean/agents/validator.py", line 73, in validate_attachments
  |     return await agent.run(prompt, toolsets=[FunctionToolset(tools)] + excel.get_toolsets())
  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/app/.venv/lib/python3.12/site-packages/pydantic_ai/agent/abstract.py", line 211, in run
  |     async with self.iter(
  |                ^^^^^^^^^^
  |   File "/usr/local/lib/python3.12/contextlib.py", line 210, in __aenter__
  |     return await anext(self.gen)
  |            ^^^^^^^^^^^^^^^^^^^^^
  |   File "/app/.venv/lib/python3.12/site-packages/pydantic_ai/agent/__init__.py", line 650, in iter
  |     async with toolset:
  |                ^^^^^^^
  |   File "/app/.venv/lib/python3.12/site-packages/pydantic_ai/toolsets/combined.py", line 56, in __aenter__
  |     await exit_stack.enter_async_context(toolset)
  |   File "/usr/local/lib/python3.12/contextlib.py", line 659, in enter_async_context
  |     result = await _enter(cm)
  |              ^^^^^^^^^^^^^^^^
  |   File "/app/.venv/lib/python3.12/site-packages/pydantic_ai/toolsets/combined.py", line 56, in __aenter__
  |     await exit_stack.enter_async_context(toolset)
  |   File "/usr/local/lib/python3.12/contextlib.py", line 659, in enter_async_context
  |     result = await _enter(cm)
  |              ^^^^^^^^^^^^^^^^
  |   File "/app/.venv/lib/python3.12/site-packages/pydantic_ai/mcp.py", line 244, in __aenter__
  |     async with AsyncExitStack() as exit_stack:
  |                ^^^^^^^^^^^^^^^^
  |   File "/usr/local/lib/python3.12/contextlib.py", line 754, in __aexit__
  |     raise exc_details[1]
  |   File "/usr/local/lib/python3.12/contextlib.py", line 737, in __aexit__
  |     cb_suppress = await cb(*exc_details)
  |                   ^^^^^^^^^^^^^^^^^^^^^^
  |   File "/usr/local/lib/python3.12/contextlib.py", line 231, in __aexit__
  |     await self.gen.athrow(value)
  |   File "/app/.venv/lib/python3.12/site-packages/pydantic_ai/mcp.py", line 512, in client_streams
  |     async with stdio_client(server=server) as (read_stream, write_stream):
  |                ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/usr/local/lib/python3.12/contextlib.py", line 231, in __aexit__
  |     await self.gen.athrow(value)
  |   File "/app/.venv/lib/python3.12/site-packages/mcp/client/stdio/__init__.py", line 181, in stdio_client
  |     anyio.create_task_group() as tg,
  |     ^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/app/.venv/lib/python3.12/site-packages/anyio/_backends/_asyncio.py", line 772, in __aexit__
  |     raise BaseExceptionGroup(
  | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
  +-+---------------- 1 ----------------
    | Exception Group Traceback (most recent call last):
    |   File "/app/.venv/lib/python3.12/site-packages/mcp/client/stdio/__init__.py", line 187, in stdio_client
    |     yield read_stream, write_stream
    |   File "/app/.venv/lib/python3.12/site-packages/pydantic_ai/mcp.py", line 513, in client_streams
    |     yield read_stream, write_stream
    |   File "/usr/local/lib/python3.12/contextlib.py", line 737, in __aexit__
    |     cb_suppress = await cb(*exc_details)
    |                   ^^^^^^^^^^^^^^^^^^^^^^
    |   File "/app/.venv/lib/python3.12/site-packages/mcp/shared/session.py", line 218, in __aexit__
    |     return await self._task_group.__aexit__(exc_type, exc_val, exc_tb)
    |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    |   File "/app/.venv/lib/python3.12/site-packages/anyio/_backends/_asyncio.py", line 772, in __aexit__
    |     raise BaseExceptionGroup(
    | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
    +-+---------------- 1 ----------------
      | Traceback (most recent call last):
      |   File "/app/.venv/lib/python3.12/site-packages/anyio/streams/memory.py", line 111, in receive
      |     return self.receive_nowait()
      |            ^^^^^^^^^^^^^^^^^^^^^
      |   File "/app/.venv/lib/python3.12/site-packages/anyio/streams/memory.py", line 106, in receive_nowait
      |     raise WouldBlock
      | anyio.WouldBlock
      | 
      | During handling of the above exception, another exception occurred:
      | 
      | Traceback (most recent call last):
      |   File "/app/.venv/lib/python3.12/site-packages/anyio/_core/_tasks.py", line 115, in fail_after
      |     yield cancel_scope
      |   File "/app/.venv/lib/python3.12/site-packages/pydantic_ai/mcp.py", line 256, in __aenter__
      |     await self._client.initialize()
      |   File "/app/.venv/lib/python3.12/site-packages/mcp/client/session.py", line 151, in initialize
      |     result = await self.send_request(
      |              ^^^^^^^^^^^^^^^^^^^^^^^^
      |   File "/app/.venv/lib/python3.12/site-packages/mcp/shared/session.py", line 272, in send_request
      |     response_or_error = await response_stream_reader.receive()
      |                         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
      |   File "/app/.venv/lib/python3.12/site-packages/anyio/streams/memory.py", line 119, in receive
      |     await receive_event.wait()
      |   File "/app/.venv/lib/python3.12/site-packages/anyio/_backends/_asyncio.py", line 1774, in wait
      |     await self._event.wait()
      |   File "/usr/local/lib/python3.12/asyncio/locks.py", line 212, in wait
      |     await fut
      | asyncio.exceptions.CancelledError: Cancelled by cancel scope 723888656570
      | 
      | During handling of the above exception, another exception occurred:
      | 
      | Traceback (most recent call last):
      |   File "/app/.venv/lib/python3.12/site-packages/pydantic_ai/mcp.py", line 255, in __aenter__
      |     with anyio.fail_after(self.timeout):
      |          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
      |   File "/usr/local/lib/python3.12/contextlib.py", line 158, in __exit__
      |     self.gen.throw(value)
      |   File "/app/.venv/lib/python3.12/site-packages/anyio/_core/_tasks.py", line 118, in fail_after
      |     raise TimeoutError
      | TimeoutError
      +------------------------------------

Additional Context

Might be related #2655

[DouweM]

@Eddydpyl Can you try explicitly entering the MCP server around everywhere you know it'll be used? Using async with mcp_server:?

[Eddydpyl]

@DouweM as in around the agent run, or around the agent's instantiation? I'll do both for now and get back to you. Just to clarify, I am already doing as indicated in the documentation: async with agent:

[Eddydpyl]

@DouweM as in around the agent run, or around the agent's instantiation? I'll do both for now and get back to you. Just to clarify, I am already doing as indicated in the documentation: async with agent:

But I wasn't doing it from both of the agents! I am now, so hopefully I won't see this error again.

This could explain the stacktrace I attached, though not all cases where I've found the issue.

[Eddydpyl]

Well this is new @DouweM, here's the code that caused the error:

class ValidationOutput(BaseModel):
    valid: bool = Field(description="Whether the output file(s) are valid.")
    reason: str = Field(description="The reason for the validation result.")


async def validate_attachments(
    abridged_history: str,
    attachments: list[str],
    base_dir: str,
) -> AgentRunResult:
    from ocean.toolkits import (
        FileSystemToolkit,
        ShellToolkit,
        ExcelToolkit,
    )

    prompt = shorten_text(f"""
        <chat_history>
        {abridged_history}
        </chat_history>

        <output_files>
        {attachments}
        </output_files>
    """)

    toolkits: List[BaseToolkit] = [
        FileSystemToolkit(base_dir=base_dir),
        ShellToolkit(base_dir=base_dir),
        ExcelToolkit(base_dir=base_dir),
    ]

    agent = Agent(
        DEFAULT_MODEL,
        tools=[tool for toolkit in toolkits for tool in toolkit.tools],
        toolsets=[toolset for toolkit in toolkits for toolset in toolkit.toolsets],
        output_type=build_output_type(output=ValidationOutput),
        model_settings=get_model_settings(temperature=0),
        instructions=shorten_text("""
            Your task is to validate the output file(s) generated by an agent.
            You must check that it is exactly what the user asked for, and for any glaring issues.
            Even if the agent achieved what it set out to do, if it's not EXACTLY what the user asked for, it's not valid.
            If the user did not ask for any files, then whatever the output file(s) are, it's directly valid.
            However if the user DOES expect files, and no output file(s) are present, then it's not valid.

            You will be provided with a summary of the conversation history, and the output file(s) to be evaluated.
            You should be particularly mindful of missing data, such as missing cells in Excel files.

            <environment>
            - You operate autonomously in a Linux sandboxed environment.
            - Files are stored relative to the sandbox directory, your root access point.
            </environment>

            <tool_usage>
            - You may use the `shell_exec` tool to run Python code to validate the output file(s).
            - You may use the `get_workbook_metadata` and `read_data_from_excel` tools to read Excel files.
            - You may use the `read_file` tool to directly access the output other file(s) contents.
            </tool_usage>
        """),
    )

    async with agent:
        result = await agent.run(prompt)
    return result

The toolset is created like so:

MCPServerStdio(
    command='uvx',
    args=['excel-mcp-server', 'stdio'],
    process_tool_call=self._process_tool_call,
    tool_prefix='excel',
    cwd=self.base_dir,
)

And here's the error in question:

+ Exception Group Traceback (most recent call last):
  |   File "/app/.venv/lib/python3.12/site-packages/opentelemetry/trace/__init__.py", line 589, in use_span
  |     yield span
  |   File "/app/.venv/lib/python3.12/site-packages/pydantic_graph/graph.py", line 259, in iter
  |     yield GraphRun[StateT, DepsT, RunEndT](
  |   File "/app/.venv/lib/python3.12/site-packages/pydantic_ai/agent/__init__.py", line 659, in iter
  |     yield agent_run
  |   File "/app/ocean/hooked/agent.py", line 258, in iter
  |     yield self._last_run
  |   File "/app/.venv/lib/python3.12/site-packages/pydantic_ai/agent/abstract.py", line 222, in run
  |     async for node in agent_run:
  |   File "/app/ocean/hooked/agent.py", line 366, in __anext__
  |     node = await self._apply_end_hooks(node)
  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/app/ocean/hooked/agent.py", line 429, in _apply_end_hooks
  |     node = await hook.end_hook(ctx, node)
  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/app/ocean/hooked/hooks/final_check.py", line 79, in end_hook
  |     return await self._validate_attachments(ctx, node)
  |            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/app/ocean/hooked/hooks/final_check.py", line 120, in _validate_attachments
  |     validate_result = await validate_attachments(abridge_result.output, last_call.tool_args.get("attachments"), self.base_dir)
  |                       ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/app/ocean/agents/validator.py", line 72, in validate_attachments
  |     async with agent:
  |                ^^^^^
  |   File "/app/.venv/lib/python3.12/site-packages/pydantic_ai/agent/__init__.py", line 1315, in __aenter__
  |     await exit_stack.enter_async_context(toolset)
  |   File "/usr/local/lib/python3.12/contextlib.py", line 659, in enter_async_context
  |     result = await _enter(cm)
  |              ^^^^^^^^^^^^^^^^
  |   File "/app/.venv/lib/python3.12/site-packages/pydantic_ai/toolsets/combined.py", line 56, in __aenter__
  |     await exit_stack.enter_async_context(toolset)
  |   File "/usr/local/lib/python3.12/contextlib.py", line 659, in enter_async_context
  |     result = await _enter(cm)
  |              ^^^^^^^^^^^^^^^^
  |   File "/app/.venv/lib/python3.12/site-packages/pydantic_ai/toolsets/combined.py", line 56, in __aenter__
  |     await exit_stack.enter_async_context(toolset)
  |   File "/usr/local/lib/python3.12/contextlib.py", line 659, in enter_async_context
  |     result = await _enter(cm)
  |              ^^^^^^^^^^^^^^^^
  |   File "/app/.venv/lib/python3.12/site-packages/pydantic_ai/mcp.py", line 244, in __aenter__
  |     async with AsyncExitStack() as exit_stack:
  |                ^^^^^^^^^^^^^^^^
  |   File "/usr/local/lib/python3.12/contextlib.py", line 754, in __aexit__
  |     raise exc_details[1]
  |   File "/usr/local/lib/python3.12/contextlib.py", line 737, in __aexit__
  |     cb_suppress = await cb(*exc_details)
  |                   ^^^^^^^^^^^^^^^^^^^^^^
  |   File "/usr/local/lib/python3.12/contextlib.py", line 231, in __aexit__
  |     await self.gen.athrow(value)
  |   File "/app/.venv/lib/python3.12/site-packages/pydantic_ai/mcp.py", line 512, in client_streams
  |     async with stdio_client(server=server) as (read_stream, write_stream):
  |                ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/usr/local/lib/python3.12/contextlib.py", line 231, in __aexit__
  |     await self.gen.athrow(value)
  |   File "/app/.venv/lib/python3.12/site-packages/mcp/client/stdio/__init__.py", line 181, in stdio_client
  |     anyio.create_task_group() as tg,
  |     ^^^^^^^^^^^^^^^^^^^^^^^^^
  |   File "/app/.venv/lib/python3.12/site-packages/anyio/_backends/_asyncio.py", line 772, in __aexit__
  |     raise BaseExceptionGroup(
  | ExceptionGroup: unhandled errors in a TaskGroup (1 sub-exception)
  +-+---------------- 1 ----------------
    | Traceback (most recent call last):
    |   File "/app/.venv/lib/python3.12/site-packages/mcp/client/stdio/__init__.py", line 160, in stdout_reader
    |     await read_stream_writer.send(session_message)
    |   File "/app/.venv/lib/python3.12/site-packages/anyio/streams/memory.py", line 243, in send
    |     self.send_nowait(item)
    |   File "/app/.venv/lib/python3.12/site-packages/anyio/streams/memory.py", line 214, in send_nowait
    |     raise BrokenResourceError
    | anyio.BrokenResourceError
    +------------------------------------

[DouweM]

@Eddydpyl Are you seeing this error consistently or intermittently? Does the error happen before the agent run actually starts (as indicated by e.g. logging or https://ai.pydantic.dev/logfire/), during a call to the MCP server, or at the end of the run?

[Eddydpyl]

@DouweM it happens intermittently, and before the agent run appears in logfire, so I imagine it's on instantiation. Another interesting bit about these errors is that I've never had them happen in my local machine. The two relevant differences I can think of between the two environments is that one is Linux (where it happens) and my local is a Mac, and that the Linux environment has a single virtual CPU, while on my Mac I have an M1 chip.

[DouweM]

@Eddydpyl It happening on instantiation (on this line in Pydantic AI, this line in mcp) suggests it's some kind of race-condition in mcp's use of anyio.create_memory_object_stream...

Specifically, the error comes from here inside anyio:

https://github.com/agronholm/anyio/blob/1b3bed72594114394d96e8709f46d1a45e7970f1/src/anyio/streams/memory.py#L214

open_receive_channels being 0 means that the MemoryObjectReceiveStream (write_stream_reader in the mcp code) was already closed:

https://github.com/agronholm/anyio/blob/1b3bed72594114394d96e8709f46d1a45e7970f1/src/anyio/streams/memory.py#L153

But I'm not sure how that could happen because the read_stream_writer is closed before the write_stream_reader:

https://github.com/modelcontextprotocol/python-sdk/blob/356dfa692bb7cdab71dcdd6d97f524fc77199291/src/mcp/client/stdio/__init__.py#L212-L215

Is it possible that the MCP server is failing?

I would suggest adding some print-debugging to stdio_client and the MemoryObjectReceiveStream/MemoryObjectSendStream objects to see the order in which things happen to get it into a broken state.

It'd also be interesting to see if you can reproduce this when using stdio_client directly, instead of going through Pydantic AI's MCPServer.

[DouweM]

@Eddydpyl By the way this may be related to #2607, if the issue is indeed that the MCP server process couldn't start successfully.

[github-actions]

This issue is stale, and will be closed in 3 days if no reply is received.

[github-actions]

Closing this issue as it has been inactive for 10 days.