WorkerRunner

Overview

WorkerRunner runs agents to completion and acts as the host for a multi-agent session. It owns the shared WorkerBus and WorkerRegistry that back the whole session, and responds to bus messages for lifecycle coordination.

from pipecat.workers.runner import WorkerRunner

PipelineRunner is a deprecated alias for WorkerRunner; existing code that uses it keeps working but should migrate to WorkerRunner.

runner = WorkerRunner()

agent = MyAgent("my_agent", ...)
await runner.add_workers(agent)

await runner.run()

Adding an agent with add_workers() attaches it to the runner’s bus and registry, so agents do not take a bus in their constructor.

Configuration

str | None

default:"None"

Unique name for this runner. Defaults to a UUID-based name. Must be unique across all runners in a distributed setup.

WorkerBus | None

default:"None"

The WorkerBus instance. Creates an AsyncQueueBus if not provided.

bool

default:"True"

Whether to automatically handle SIGINT signals for graceful shutdown.

bool

default:"False"

Whether to automatically handle SIGTERM signals for graceful shutdown.

bool

default:"False"

Whether to force garbage collection after the main worker completes.

bool

default:"True"

Whether to warn about tasks left running on the shared task manager once every worker has finished.

BaseTaskManager | None

default:"None"

Optional task manager for handling asyncio tasks. If None, a default TaskManager is created with the current running loop.

asyncio.AbstractEventLoop | None

default:"None"

Event loop to use. If None, uses the current running loop.

Deprecated since 1.5.0. Use task_manager (which owns its own loop) instead. This parameter will be removed in 2.0.0.

Properties

bus

runner.bus -> WorkerBus

The bus this runner hosts, shared across all added agents.

registry

runner.registry -> WorkerRegistry

The shared agent registry this runner owns.

Methods

add_workers

async def add_workers(self, *workers: BaseWorker) -> None

Add one or more agents to this runner. Each agent is attached to the runner’s bus and registry and started in the background. Can be called before or after run(). When called after run() has started, each agent starts immediately; otherwise agents are queued and started during run setup.

Parameter	Type	Description
`*workers`	`BaseWorker`	One or more agents to add

get_worker

def get_worker(self, name: str) -> BaseWorker | None

Return an agent added to this runner, by name.

Parameter	Type	Description
`name`	`str`	Name of the agent to find

Returns: The agent, or None if this runner has none by that name. Only agents on the same runner have a local instance to return. One on another runner is addressable over the bus but has no object to hand back, so this returns None for it.

async def research(params: FunctionCallParams, query: str):
    ui_jobs = params.worker_runner.get_worker("ui-jobs")

run

async def run(self, worker: BaseWorker | None = None, *, auto_end: bool = True) -> None

Run all added agents and block until the runner is stopped. By default (auto_end=True), the runner ends once every root agent has finished. For long-lived hosts that add and remove agents over many sessions, pass auto_end=False so the runner does not exit when no agents are left.

Parameter	Type	Default	Description
`worker`	`BaseWorker \| None`	`None`	Optional agent to run. Deprecated — register with `add_workers()` before calling `run()` instead.
`auto_end`	`bool`	`True`	When `True`, the runner ends once every root agent has finished. When `False`, it blocks until `end()` or `cancel()` is called.

end

async def end(self, reason: str | None = None) -> None

Gracefully end all agents and shut down. Sends end messages to root agents; parent agents propagate shutdown to their children automatically. The messages go out as the runner exits, carrying the reason given here. Idempotent.

Parameter	Type	Default	Description
`reason`	`str \| None`	`None`	Human-readable reason for ending. Reaches each worker on the end message sent as the runner exits.

cancel

async def cancel(self, reason: str | None = None) -> None

Signals shutdown, which run() answers by cancelling every worker still going and waiting for it. The messages go out as the runner exits, carrying the reason given here. Idempotent; subsequent calls are ignored.

Parameter	Type	Default	Description
`reason`	`str \| None`	`None`	Human-readable reason for cancelling. Reaches each worker on the cancel message sent as the runner exits.

Event Handlers

@runner.event_handler("on_ready")
async def on_ready(runner):
    print("All agents started")

@runner.event_handler("on_error")
async def on_error(runner, error):
    print(f"Runner error: {error}")

Event	Description
`on_ready`	Fired after the runner has finished setup and any added agents have started
`on_error`	Fired when starting an added agent fails

Get Started

Pipecat Server

Client SDKs

Pipecat Flows

Pipecat Cloud

CLI

Pipecat Context Hub

Overview

Configuration

Properties

bus

registry

Methods

add_workers

get_worker

run

end

cancel

Event Handlers

​Overview

​Configuration

​Properties

​bus

​registry

​Methods

​add_workers

​get_worker

​run

​end

​cancel

​Event Handlers

Overview

Configuration

Properties

bus

registry

Methods

add_workers

get_worker

run

end

cancel

Event Handlers