Function serve_loop

src/mcp/server/runner.py:440–470

Drive ``server`` in loop mode over a stream pair until the channel closes. Builds the loop-mode `JSONRPCDispatcher` + `Connection` and hands them to `serve_connection`, so loop-mode callers share one dispatcher-construction recipe (notably the `inline_methods={"initialize"}` rule). Call

(
    server: Server[LifespanT],
    read_stream: ReadStream[SessionMessage | Exception],
    write_stream: WriteStream[SessionMessage],
    *,
    lifespan_state: LifespanT,
    session_id: str | None = None,
    init_options: InitializationOptions | None = None,
    raise_exceptions: bool = False,
)

async def serve_loop(
    server: Server[LifespanT],
    read_stream: ReadStream[SessionMessage | Exception],
    write_stream: WriteStream[SessionMessage],
    *,
    lifespan_state: LifespanT,
    session_id: str | None = None,
    init_options: InitializationOptions | None = None,
    raise_exceptions: bool = False,
) -> None:
    """Drive ``server`` in loop mode over a stream pair until the channel closes.

    Builds the loop-mode `JSONRPCDispatcher` + `Connection` and hands them to
    `serve_connection`, so loop-mode callers share one dispatcher-construction
    recipe (notably the `inline_methods={"initialize"}` rule). Callers that own
    a lifespan (the streamable-HTTP manager) pass it in; callers that don't
    (`Server.run` for stdio/memory) enter the lifespan and then call this.
    """
    dispatcher: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(
        read_stream,
        write_stream,
        raise_handler_exceptions=raise_exceptions,
        # Handle `initialize` inline so a client that pipelines it with the
        # next request (spec: SHOULD NOT, not MUST NOT) sees the initialized
        # state instead of failing the init-gate.
        inline_methods=frozenset({"initialize"}),
    )
    connection = Connection.for_loop(dispatcher, session_id=session_id)
    await serve_connection(
        server, dispatcher, connection=connection, lifespan_state=lifespan_state, init_options=init_options
    )


async def serve_one(