Python: MCP connection not properly invalidated after async generator cleanup

[victordibia]

Summary

When an MCPStdioTool (or any MCP tool) is used e.g, within a FastAPI/Starlette StreamingResponse, the tool works on the first request but fails on all subsequent requests with anyio.ClosedResourceError.

The core issue: The is_connected property returns True even after the underlying stdio streams have been closed by async generator cleanup.

Affected: All MCP tool types (MCPStdioTool, MCPSseTool) when used in streaming HTTP contexts.

Root Cause

1. FastAPI StreamingResponse wraps an async generator
2. When the HTTP response completes, the generator is closed via GeneratorExit
3. This propagates through the async stack and triggers anyio task group cleanup (any async GeneratorExit event will also create similar events)
4. The MCP stdio reader/writer streams are closed as part of this cleanup
5. Bug: MCPTool._session is not cleared, so is_connected still returns True
6. Next request tries to use the dead session → ClosedResourceError

Reproduction

# bug_repro.py - Run with: python bug_repro.py
import os
from pathlib import Path
import uvicorn
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

from agent_framework import ChatAgent, MCPStdioTool
from agent_framework.azure import AzureOpenAIChatClient
from azure.identity import AzureCliCredential

TEST_DIR = Path(os.path.expanduser("~/mcp_test"))
TEST_DIR.mkdir(exist_ok=True)
(TEST_DIR / "test.txt").write_text("hello")

# Global agent - reused across HTTP requests (common pattern)
_agent, _mcp_tool = None, None

def get_agent():
    global _agent, _mcp_tool
    if _agent is None:
        _mcp_tool = MCPStdioTool(
            name="filesystem",
            command="npx",
            args=["-y", "@modelcontextprotocol/server-filesystem", str(TEST_DIR)],
        )
        _agent = ChatAgent(
            chat_client=AzureOpenAIChatClient(credential=AzureCliCredential()),
            name="Test Agent",
            instructions=f"Use filesystem tools to list files in {TEST_DIR}.",
            tools=_mcp_tool,
        )
    return _agent, _mcp_tool

app = FastAPI()

class ChatRequest(BaseModel):
    message: str

@app.post("/chat")
async def chat(req: ChatRequest):
    """Streaming endpoint - the StreamingResponse triggers the bug."""
    agent, mcp = get_agent()
    print(f"is_connected BEFORE: {mcp.is_connected}")  # True on 2nd request (BUG!)

    async def stream():
        async for chunk in agent.run_stream(req.message):
            if hasattr(chunk, "text") and chunk.text:
                yield chunk.text
        print(f"is_connected AFTER: {mcp.is_connected}")  # Still True (BUG!)

    return StreamingResponse(stream(), media_type="text/plain")

if __name__ == "__main__":
    uvicorn.run(app, port=8766)

Test commands:

# Terminal 1: Start server
python bug_repro.py

# Terminal 2: Make requests
curl -X POST http://localhost:8766/chat -H "Content-Type: application/json" \
     -d '{"message": "list files"}'
# ✅ Works - returns file list

curl -X POST http://localhost:8766/chat -H "Content-Type: application/json" \
     -d '{"message": "list files"}'
# ❌ Fails - model says "unable to access directory"

Server logs show:

Request 1: "Function list_directory succeeded."
Request 2: [no succeeded log - tool call fails with ClosedResourceError]

Error Traceback

anyio.ClosedResourceError
  File "mcp/shared/session.py", line 281, in send_request
      await self._write_stream.send(SessionMessage(...))
  File "anyio/streams/memory.py", line 218, in send_nowait
      raise ClosedResourceError

Suggested Fixes

Option 1: Catch ClosedResourceError and invalidate session (Recommended)

In _mcp.py lines 758-763:

from anyio import ClosedResourceError

async def call_tool(self, tool_name: str, **kwargs: Any) -> list[AIContent]:
    if not self.session:
        raise ToolExecutionException("MCP server not connected")
    try:
        return _mcp_call_tool_result_to_ai_contents(
            await self.session.call_tool(tool_name, arguments=kwargs)
        )
    except ClosedResourceError as ex:
        # Underlying streams closed - invalidate session
        self._session = None
        self._exit_stack = None
        raise ToolExecutionException(
            f"MCP connection closed unexpectedly. Call connect() to reconnect.",
            inner_exception=ex
        ) from ex
    except McpError as mcp_exc:
        raise ToolExecutionException(mcp_exc.error.message) from mcp_exc

Option 2: Validate stream state in is_connected

In _mcp.py lines 540-542:

@property
def is_connected(self) -> bool:
    if self._session is None:
        return False
    # Check if write stream is still open
    try:
        return not getattr(self._session._write_stream, '_closed', False)
    except AttributeError:
        return True

Option 3: Auto-reconnect on ClosedResourceError

Wrap call_tool() to catch ClosedResourceError, call connect(), and retry once.

Environment

• agent-framework: main branch
• Python: 3.10+
• OS: macOS/Linux
• FastAPI + uvicorn (or any ASGI server with streaming)

Related to #1476 #1515 #2865

[victordibia]

@eavanvalkenburg , would be happy to get your thoughts/discuss this at some point.
I can open a PR if that makes sense.