BaseUIWorker

Worker that surfaces its jobs and job groups on the client UI as progress cards, without involving an LLM.

BaseUIWorker is deprecated since 1.12.0 and will be removed in 2.0.0. Use UIWorker instead. UIWorker now reports its job groups to the client on its own, so instead of a separate BaseUIWorker dispatcher, dispatch job groups from a @job handler in your UIWorker.

BaseUIWorker extends BaseWorker with client visibility. Every job group it dispatches streams its lifecycle to the client as ui-job-group envelopes, so background work appears as a progress card the user can watch and cancel.

It involves no LLM. Instantiate one directly and register it on the runner as a dispatcher when a pipeline app wants client-visible background work driven by its own tools:

from pipecat.pipeline.job_context import JobGroupParams
from pipecat.workers.base_ui_worker import BaseUIWorker

ui_jobs = BaseUIWorker("ui-jobs")

async def research(params: FunctionCallParams, query: str):
    job_id = await params.worker_runner.get_worker("ui-jobs").request_job_group(
        "wikipedia", "news",
        params=JobGroupParams(payload={"query": query}, label=f"Research: {query}"),
    )
    await params.result_callback({"status": "started", "job_id": job_id})

Migrating to UIWorker

UIWorker no longer inherits from BaseUIWorker: it reports its own job groups to the client in the same way. Instead of a BaseUIWorker dispatcher that the main pipeline’s tools reach through the runner, give the work a @job handler on your UIWorker and have the tool send that job:

class ResearchWorker(UIWorker):
    @job(name="research")
    async def _research(self, message: BusJobRequestMessage) -> None:
        query = (message.payload or {}).get("query", "")
        async with self.job_group(
            "wikipedia", "news",
            params=JobGroupParams(payload={"query": query}, label=f"Research: {query}"),
        ) as group:
            pass
        await self.send_job_response(message.job_id, {"results": group.responses})


async def research(params: FunctionCallParams, query: str):
    async with params.pipeline_worker.job(
        "ui", params=JobParams(name="research", payload={"query": query}, timeout=60)
    ) as t:
        pass
    await params.result_callback(t.response)

The client sees the same cards. See Job groups on the UIWorker page.

What reaches the client

A group dispatched by a BaseUIWorker publishes:

Envelope	When
group_started	At dispatch, carrying the group’s workers, label, and cancellable
job_update	As each worker sends an intermediate update
job_completed	As each worker responds
group_completed	At group teardown — normal completion, cancellation, or timeout

The client’s reserved __cancel_job_group event is translated into a cancellation for any group dispatched with JobGroupParams(cancellable=True). A group dispatched without it is still reported to the client, but the client cannot stop it.

Dispatching

BaseUIWorker uses the same job API as BaseWorker — job_group(), request_job_group(), and create_job_group_and_request_job(). There is no separate UI-specific dispatch method: every group this worker dispatches is client-visible by virtue of the class.

Give the group a label. It titles the card the user sees, and without one the card has nothing to name the work:

async with self.job_group(
    "wikipedia", "news", "scholar",
    params=JobGroupParams(
        payload={"query": query},
        label=f"Research: {query}",
        cancellable=True,
    ),
) as jg:
    results = await jg.wait()

Job hooks

The same hooks as BaseWorker — on_job_update, on_job_response, on_job_stream_end, and on_job_completed — with the client forwarding layered on top. Always call super() when overriding, or the client stops receiving the group’s lifecycle.