`StdioTransport(keep_alive=True)` can't be reused across concurrently-alive event loops — `initialize` handshake hangs for the full `init_timeout`

[dsfaccini]

Summary

A Client(StdioTransport(...)) with the default keep_alive=True cannot be reused on a second event loop while the first loop is still alive. The persisted subprocess connection (reader task + anyio streams) is bound to the loop of first connect; on a second concurrently-alive loop the MCP initialize handshake hangs for the full init_timeout, then raises RuntimeError: Failed to initialize server session. With init_timeout=None it deadlocks indefinitely.

Why it matters

Any app that shares one Client/transport across event loops hits this: FastAPI/ASGI apps with a background loop, worker pools, and durable-execution frameworks (DBOS/Temporal) that recover or fork a run onto a different loop. keep_alive=False avoids it but forfeits connection reuse.

Mechanism

StdioTransport.__init__ defaults keep_alive=True, persisting _connect_task + streams on the first loop. Client.__aenter__ reuses that connection; _is_session_dead only checks stream closure, not loop identity, so a foreign-loop-bound session is reused. mcp/shared/session.py:send_request → response_stream_reader.receive() never wakes on the second loop, and anyio.fail_after(init_timeout) trips → RuntimeError.

Expected

On reuse from a different running loop, StdioTransport should detect the loop change and reconnect (as keep_alive=False effectively does) rather than reuse the foreign-loop-bound session.

Environment

fastmcp 3.4.2, mcp 1.25.0, Python 3.13.

Minimal reproduction (self-contained)

"""MRE-1 — pure fastmcp, NO pydantic-ai.

Shows that a fastmcp `Client(StdioTransport(...))` with the default `keep_alive=True`
cannot be reused on a second, concurrently-alive event loop: the MCP `initialize`
handshake hangs until `init_timeout` and then raises. `keep_alive=False` fixes it.

Self-contained: writes a tiny stdio MCP server to a temp file and runs it as the subprocess.
"""

import asyncio
import textwrap
import threading
import time
from pathlib import Path
from tempfile import mkdtemp

from fastmcp import Client
from fastmcp.client.transports import StdioTransport

# --- tiny inline stdio MCP server ---
SERVER = textwrap.dedent('''
    from fastmcp import FastMCP
    mcp = FastMCP('mre-server')

    @mcp.tool
    def ping() -> str:
        return 'pong'

    if __name__ == '__main__':
        mcp.run()  # stdio transport by default
''')
_srv = Path(mkdtemp()) / 'mre_server.py'
_srv.write_text(SERVER)


def make_transport(keep_alive: bool) -> StdioTransport:
    return StdioTransport(command='python', args=[str(_srv)], keep_alive=keep_alive)


async def use(client: Client) -> int:
    async with client:
        tools = await client.list_tools()
    return len(tools)


def run_on_fresh_loop(coro_factory, timeout: float) -> tuple[str, float]:
    """Run coro on a brand-new event loop in the CURRENT thread; return (result, seconds)."""
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    t0 = time.monotonic()
    try:
        loop.run_until_complete(asyncio.wait_for(coro_factory(), timeout=timeout))
        return 'OK', time.monotonic() - t0
    except asyncio.TimeoutError:
        return 'OUTER-TIMEOUT', time.monotonic() - t0
    except Exception as e:  # noqa: BLE001
        return f'{type(e).__name__}: {str(e)[:80]}', time.monotonic() - t0
    finally:
        loop.close()


def scenario(label: str, keep_alive: bool, init_timeout: float) -> None:
    print(f'\n=== {label}  (keep_alive={keep_alive}, init_timeout={init_timeout}s) ===', flush=True)
    client = Client(make_transport(keep_alive), init_timeout=init_timeout)

    # Loop A: first use, kept ALIVE in its own thread for the whole scenario.
    loopA = asyncio.new_event_loop()
    threading.Thread(target=lambda: (asyncio.set_event_loop(loopA), loopA.run_forever()), daemon=True).start()
    time.sleep(0.2)
    n = asyncio.run_coroutine_threadsafe(use(client), loopA).result(timeout=30)
    print(f'  loop A (first use): OK, {n} tools', flush=True)

    # Loop B: reuse the SAME client on a different, concurrently-alive loop (loop A still running).
    result: dict[str, tuple[str, float]] = {}
    tB = threading.Thread(target=lambda: result.__setitem__('r', run_on_fresh_loop(lambda: use(client), timeout=init_timeout + 30)))
    tB.start()
    tB.join(timeout=init_timeout + 45)
    if tB.is_alive():
        print(f'  loop B (reuse): HUNG (never returned)', flush=True)
    else:
        res, secs = result['r']
        print(f'  loop B (reuse): {res}   [{secs:.1f}s]', flush=True)
    loopA.call_soon_threadsafe(loopA.stop)


def main() -> None:
    import fastmcp, mcp  # noqa
    print(f'fastmcp={fastmcp.__version__}  mcp={__import__("importlib.metadata", fromlist=["version"]).version("mcp")}')
    # BROKEN: default keep_alive=True, two init_timeout values to show the hang is bounded by init_timeout.
    scenario('BROKEN #1', keep_alive=True, init_timeout=5)
    scenario('BROKEN #2 (longer timeout => longer hang)', keep_alive=True, init_timeout=15)
    # FIXED: keep_alive=False tears the subprocess down each exit, so loop-B reuse reconnects cleanly.
    scenario('FIXED', keep_alive=False, init_timeout=5)


if __name__ == '__main__':
    main()

Output: BROKEN hangs 5.0s / 15.0s (== init_timeout) then raises; FIXED (keep_alive=False) succeeds in 0.6s. (Full output identical to the pydantic-ai tracker issue.)