René's URL Explorer Experiment

{"@context":"https://schema.org","@type":"DiscussionForumPosting","headline":"Concurrent server-side requests are serialized end-to-end by BaseSession","articleBody":"### Description\n\nWhen a server tool issues multiple concurrent server to client requests over a single MCP session — for example, fanning out several `ServerSession.create_message(...)` (sampling) calls via `asyncio.gather` — only one is ever in flight on the receiving side: `BaseSession._receive_loop` awaits each incoming request handler inline before reading the next message off the stream, so subsequent responses can't even be dequeued until the current handler returns. A minimal reproducer using `mcp.server.fastmcp.FastMCP` + `ClientSession` over the SDK's own in-memory transport (5 concurrent `create_message` calls, sampling callback sleeps 0.5s) shows peak in-flight = 1 and elapsed ≈ 2.5s instead of ≈ 0.5s.\n\nConcrete use case: a tool that fans out dozens of independent sampling requests per invocation against an LLM deployment with provisioned capacity sitting idle, the work is embarrassingly parallel and there's token budget to burn, but wall-clock scales linearly with the fan-out because every request blocks the next one. The same shape applies to elicitation, list-roots, and any other server to client request type, since they all flow through the same loop. \n\nI'd like the SDK to support concurrent handling of incoming requests on a single session, likely as an opt-in knob to preserve today's strict-ordering semantics for handlers that depend on them, with whatever bounding/back-pressure mechanism the maintainers prefer.\n\n### References\n\n```python\n\"\"\"Reproducer: concurrent server-\u003eclient requests are serialized end-to-end.\n\nA FastMCP tool fans out N concurrent ``ServerSession.create_message`` calls\nvia ``asyncio.gather``. The client's sampling callback records peak in-flight.\n\nExpected with concurrent dispatch:  peak == N,  elapsed ~= SLEEP\nObserved today:                     peak == 1,  elapsed ~= N * SLEEP\n\nRoot cause: ``BaseSession._receive_loop`` awaits each incoming request\nhandler inline before reading the next message off the stream\n(mcp/shared/session.py).\n\nRun:  python repro_concurrent_sampling.py\n\"\"\"\n\nimport asyncio\n\nfrom mcp.server.fastmcp import Context, FastMCP\nfrom mcp.shared.context import RequestContext\nfrom mcp.shared.memory import create_connected_server_and_client_session\nfrom mcp.types import (\n    CreateMessageRequestParams,\n    CreateMessageResult,\n    SamplingMessage,\n    TextContent,\n)\n\nN = 5\nSLEEP = 0.5\n\n\nasync def main() -\u003e None:\n    inflight = 0\n    peak = 0\n\n    async def sampling_callback(\n        _ctx: RequestContext,\n        params: CreateMessageRequestParams,\n    ) -\u003e CreateMessageResult:\n        nonlocal inflight, peak\n        inflight += 1\n        peak = max(peak, inflight)\n        try:\n            await asyncio.sleep(SLEEP)\n        finally:\n            inflight -= 1\n        msg = params.messages[0].content\n        echo = msg.text if isinstance(msg, TextContent) else \"\"\n        return CreateMessageResult(\n            role=\"assistant\",\n            content=TextContent(type=\"text\", text=f\"echo:{echo}\"),\n            model=\"test-model\",\n        )\n\n    server = FastMCP(\"repro\")\n\n    @server.tool()\n    async def fanout(ctx: Context) -\u003e str:\n        async def one(i: int) -\u003e str:\n            r = await ctx.session.create_message(\n                messages=[\n                    SamplingMessage(\n                        role=\"user\",\n                        content=TextContent(type=\"text\", text=str(i)),\n                    )\n                ],\n                max_tokens=8,\n            )\n            return r.content.text if isinstance(r.content, TextContent) else \"\"\n\n        return \",\".join(await asyncio.gather(*(one(i) for i in range(N))))\n\n    loop = asyncio.get_running_loop()\n    t0 = loop.time()\n    async with create_connected_server_and_client_session(\n        server, sampling_callback=sampling_callback\n    ) as session:\n        await session.call_tool(\"fanout\", {})\n    elapsed = loop.time() - t0\n\n    print(f\"N={N}, per-call sleep={SLEEP}s\")\n    print(f\"peak in-flight: {peak}    (concurrent: {N}, serialized: 1)\")\n    print(f\"elapsed:        {elapsed:.2f}s  (concurrent: ~{SLEEP}s, serialized: ~{N * SLEEP}s)\")\n\n    assert peak == N, (\n        f\"server-\u003eclient requests were serialized: peak in-flight={peak}, expected {N}\"\n    )\n\n\nif __name__ == \"__main__\":\n    asyncio.run(main())\n```","author":{"url":"https://github.com/demoray","@type":"Person","name":"demoray"},"datePublished":"2026-04-21T21:00:57.000Z","interactionStatistic":{"@type":"InteractionCounter","interactionType":"https://schema.org/CommentAction","userInteractionCount":0},"url":"https://github.com/2489/python-sdk/issues/2489"}