serving

async def serve_stream(
    server: Server[LifespanT],
    read_stream: ReadStream[SessionMessage | Exception],
    write_stream: WriteStream[SessionMessage],
    *,
    initialization_options: InitializationOptions | None = None,
    lifespan_state: LifespanT | _Unset = _UNSET,
    raise_exceptions: bool = False,
    task_status: anyio.abc.TaskStatus[None] = anyio.TASK_STATUS_IGNORED,
) -> None:
    """Serve `server` over a duplex message stream until the read side closes.

    The driver for stream transports: the client's opening messages decide the
    connection's era. Enters `server.lifespan()` unless `lifespan_state` is
    given.

    Args:
        initialization_options: The legacy handshake's `InitializeResult`
            payload; defaults to `server.create_initialization_options()`.
        lifespan_state: An already-entered lifespan state, for several
            connections sharing one lifespan.
        raise_exceptions: Also re-raise handler exceptions out of this call
            after the peer has been answered (an in-process testing aid).
    """
    if isinstance(lifespan_state, _Unset):
        async with server.lifespan() as state:
            await serve_stream(
                server,
                read_stream,
                write_stream,
                initialization_options=initialization_options,
                lifespan_state=state,
                raise_exceptions=raise_exceptions,
                task_status=task_status,
            )
        return
    connection = _StreamConnection(
        server,
        read_stream,
        write_stream,
        lifespan_state=lifespan_state,
        init_options=initialization_options,
        session_id=None,
        raise_exceptions=raise_exceptions,
    )
    await connection.run(task_status=task_status)

async def serve_legacy_stream(
    server: Server[LifespanT],
    read_stream: ReadStream[SessionMessage | Exception],
    write_stream: WriteStream[SessionMessage],
    *,
    lifespan_state: LifespanT,
    session_id: str | None = None,
) -> None:
    """Serve a stream the transport has already routed to the legacy handshake era.

    Transport-internal (the streamable-HTTP manager's stateful sessions are
    born legacy); not part of the author-facing surface.
    """
    connection = _StreamConnection(
        server,
        read_stream,
        write_stream,
        lifespan_state=lifespan_state,
        init_options=None,
        session_id=session_id,
        raise_exceptions=False,
    )
    connection.open_legacy_era()
    await connection.run()

async def serve_listener(server: Server[LifespanT], listener: anyio.abc.Listener[anyio.abc.ByteStream]) -> None:
    """Serve every connection `listener` accepts, over the stdio wire, until cancelled.

    Enters the server's lifespan once (shared by every connection), frames each
    accepted byte stream with `newline_json_transport`, and drives it with
    `serve_stream`; takes ownership of `listener` and closes it on the way out.

        listener = await anyio.create_unix_listener("/tmp/mcp.sock")
        await serve_listener(server, listener)

    Caveat: `subscriptions/listen` streams belong to the shared `server`, so
    one connection's disconnect gracefully ends the open listen streams of
    every connection this server is serving.
    """
    async with listener, server.lifespan() as lifespan_state:

        async def handle(stream: anyio.abc.ByteStream) -> None:
            # Accepted here, so closed here.
            async with stream, newline_json_transport(stream) as (read_stream, write_stream):
                await serve_stream(server, read_stream, write_stream, lifespan_state=lifespan_state)

        await listener.serve(handle)