nemo_voice_agent.pipecat.bot_server | NeMo Labs Voice Agent

Shared bot-server runner.

Bots build their own services, RTVI processor, pipeline, and PipelineWorker; this module handles the boilerplate that sits around the task — transport event handlers, the RTVI on_client_ready kick-off, audio-logger finalization, shutdown, and optional FastAPI /connect endpoint.

The runner makes no assumptions about pipeline contents: it only needs the task, ws_transport, and rtvi the bot already constructed. Per-bot customization (what to reset on disconnect, which initial frame to queue on client-ready) is passed via keyword arguments.

Module Contents

Functions

Name	Description
`_reset_services`	-
`create_fastapi_app`	FastAPI app with CORS + `/connect` + a stub `/ws` endpoint.
`run_bot_websocket_server`	Wire event handlers onto `ws_transport`/`rtvi` and run the pipeline.
`run_bot_with_fastapi`	Run the websocket server coroutine and uvicorn concurrently.

Data

__all__

API

nemo_voice_agent.pipecat.bot_server._reset_services(
    services: typing.Optional[typing.List[typing.Any]]
) -> None

nemo_voice_agent.pipecat.bot_server.create_fastapi_app(
    websocket_port: int,
    public_host: str = '127.0.0.1',
    ws_scheme: str = 'ws'
) -> fastapi.FastAPI

FastAPI app with CORS + /connect + a stub /ws endpoint.

/connect returns the ws URL the pipecat client should dial. /ws is kept as a placeholder mirroring the original bot scripts’ unimplemented FastAPI websocket path.

nemo_voice_agent.pipecat.bot_server.run_bot_websocket_server(
    task: pipecat.pipeline.worker.PipelineWorker,
    ws_transport: pipecat.transports.websocket.server.SingleClientWebsocketServerTransport,
    rtvi: pipecat.processors.frameworks.rtvi.RTVIProcessor,
    task_ref: typing.Optional[nemo_voice_agent.pipecat.processors.frameworks.rtvi_actions.TaskRef] = None,
    audio_logger: typing.Optional[nemo_voice_agent.pipecat.services.nemo.audio_logger.AudioLogger] = None,
    talk_first: bool = True,
    initial_frame_factory: typing.Optional[typing.Callable[[], pipecat.frames.frames.Frame]] = None,
    on_disconnect_reset_services: typing.Optional[typing.List[typing.Any]] = None,
    on_disconnect_hook: typing.Optional[typing.Callable[[], typing.Awaitable[None]]] = None
) -> None

async

Wire event handlers onto ws_transport/rtvi and run the pipeline.

Parameters:

task

PipelineWorker

Fully constructed PipelineWorker with observers already attached.

ws_transport

SingleClientWebsocketServerTransport

Transport the task’s pipeline uses for I/O.

rtvi

RTVIProcessor

RTVI processor embedded in the pipeline. Must already have any actions the bot wants registered.

task_ref

Optional[TaskRef]" default="None

Optional TaskRef passed to the RTVI handler factories. Populated here so handlers can reach the live pipeline worker.

audio_logger

Optional[AudioLogger]" default="None

Audio logger to finalize on disconnect / shutdown.

talk_first

bool" default="True

If True, queue initial_frame_factory() on client-ready.

initial_frame_factory

Optional[Callable[[], Frame]]" default="None

Callable returning the Frame to kick off with (e.g. lambda: LLMRunFrame()). Invoked once per client-ready.

on_disconnect_reset_services

Optional[List[Any]]" default="None

Services whose .reset() is called on client disconnect. None entries are skipped.

on_disconnect_hook

Optional[Callable[[], Awaitable[None]]]" default="None

Optional extra async cleanup called on disconnect after the default reset logic.

nemo_voice_agent.pipecat.bot_server.run_bot_with_fastapi(
    ws_coro: typing.Awaitable[None],
    app: fastapi.FastAPI,
    host: str,
    fastapi_port: int
) -> None

async

Run the websocket server coroutine and uvicorn concurrently.

nemo_voice_agent.pipecat.bot_server.__all__ = ['run_bot_websocket_server', 'create_fastapi_app', 'run_bot_with_fastapi']