Documentation Index
Fetch the complete documentation index at: https://docs.pipecat.ai/llms.txt
Use this file to discover all available pages before exploring further.
JobStatus
from pipecat.pipeline.job_context import JobStatus
str so values compare naturally with plain strings.
| Value | Description |
|---|
JobStatus.COMPLETED | The job finished successfully |
JobStatus.CANCELLED | The job was cancelled by the requester |
JobStatus.FAILED | The job failed due to a logical or business error |
JobStatus.ERROR | The job encountered an unexpected runtime error |
WorkerActivationArgs
from pipecat.pipeline.base_worker import WorkerActivationArgs
| Field | Type | Default | Description |
|---|
metadata | dict | None | None | Optional structured data passed during activation |
Methods
# Create from a dict (ignores unknown keys)
args = WorkerActivationArgs.from_dict({"metadata": {...}})
# Convert to a dict (excludes None values)
data = args.to_dict()
LLMWorkerActivationArgs
from pipecat.workers.llm import LLMWorkerActivationArgs
LLMWorker. Extends WorkerActivationArgs.
| Field | Type | Default | Description |
|---|
metadata | dict | None | None | Optional structured data passed during activation |
messages | list | None | None | LLM context messages to inject on activation |
run_llm | bool | None | None | Whether to run the LLM after appending messages. Defaults to True when messages is set |
JobContext
from pipecat.pipeline.job_context import JobContext
async for to receive intermediate events.
On normal completion, the result is available via response. On worker error or timeout, raises JobError.
Properties
| Property | Type | Description |
|---|
job_id | str | The job identifier |
response | dict | The worker’s response payload |
Usage
async with self.job("worker", payload=data) as j:
async for event in j:
print(f"[{event.type}]: {event.data}")
print(j.response)
JobGroupContext
from pipecat.pipeline.job_context import JobGroupContext
async for to receive intermediate events.
On normal completion, results are available via responses. On worker error (with cancel_on_error=True) or timeout, raises JobGroupError.
Properties
| Property | Type | Description |
|---|
job_id | str | The shared job identifier for this group |
responses | dict[str, dict] | Collected responses keyed by worker name |
Usage
async with self.job_group("w1", "w2", payload=data) as jg:
async for event in jg:
print(f"{event.worker_name} [{event.type}]: {event.data}")
for name, result in jg.responses.items():
print(name, result)
JobEvent
from pipecat.pipeline.job_context import JobEvent
async for over a JobContext.
| Field | Type | Description |
|---|
type | str | The event type (see constants below) |
data | dict | None | Optional event payload |
Event Type Constants
| Constant | Value | Description |
|---|
JobEvent.UPDATE | "update" | Progress update from a worker |
JobEvent.STREAM_START | "stream_start" | Worker started streaming |
JobEvent.STREAM_DATA | "stream_data" | Streaming data chunk |
JobEvent.STREAM_END | "stream_end" | Worker finished streaming |
JobGroupResponse
from pipecat.pipeline.job_context import JobGroupResponse
on_job_completed.
| Field | Type | Description |
|---|
job_id | str | The shared job identifier |
responses | dict[str, dict] | Collected responses keyed by worker name |
JobGroup
from pipecat.pipeline.job_context import JobGroup
BaseWorker.job_groups and returned by create_job_group_and_request_job.
| Field | Type | Default | Description |
|---|
job_id | str | | Shared identifier for all agents in this group |
worker_names | set[str] | | Names of the agents in the group |
responses | dict[str, dict] | {} | Collected responses keyed by worker name |
timeout_task | asyncio.Task | None | None | Optional task that cancels the group on timeout |
cancel_on_error | bool | True | Whether to cancel the group if a worker errors |
Properties
| Property | Type | Description |
|---|
is_done | bool | Whether the group has completed or failed |
Methods
# Wait for all agents in the group to respond (raises JobGroupError on failure)
await group.wait()
JobGroupEvent
from pipecat.pipeline.job_context import JobGroupEvent
async for over a JobGroupContext.
| Field | Type | Description |
|---|
type | str | The event type (see constants below) |
worker_name | str | The name of the worker that sent the event |
data | dict | None | Optional event payload |
Event Type Constants
| Constant | Value | Description |
|---|
JobGroupEvent.UPDATE | "update" | Progress update from a worker |
JobGroupEvent.STREAM_START | "stream_start" | Worker started streaming |
JobGroupEvent.STREAM_DATA | "stream_data" | Streaming data chunk |
JobGroupEvent.STREAM_END | "stream_end" | Worker finished streaming |
WorkerReadyData
from pipecat.registry.types import WorkerReadyData
on_worker_ready and @worker_ready handlers.
| Field | Type | Description |
|---|
worker_name | str | The name of the agent |
runner | str | The name of the runner managing this agent |
WorkerErrorData
from pipecat.registry.types import WorkerErrorData
on_worker_failed.
| Field | Type | Description |
|---|
worker_name | str | The name of the agent that errored |
error | str | Description of the error |
WorkerRegistryEntry
from pipecat.registry.types import WorkerRegistryEntry
| Field | Type | Default | Description |
|---|
name | str | | The agent’s name |
parent | str | None | None | Name of the parent agent, or None for root agents |
active | bool | False | Whether the agent is currently active |
bridged | bool | False | Whether the agent is bridged |
started_at | float | None | None | Unix timestamp when the agent became ready |
WorkerRegistry
from pipecat.registry import WorkerRegistry
WorkerRunner and shared with its agents.
Properties
| Property | Type | Description |
|---|
runner_name | str | The name of the runner that owns this registry |
local_workers | list[str] | Names of agents registered under this runner |
remote_workers | list[str] | Names of agents registered under remote runners |
Methods
get
def get(self, worker_name: str) -> WorkerReadyData | None
watch
async def watch(self, worker_name: str, handler: WatchHandler) -> None
| Parameter | Type | Description |
|---|
worker_name | str | The agent name to watch for |
handler | Callable[[WorkerReadyData], Coroutine] | Async callable invoked with the agent’s data |
register
async def register(self, worker_data: WorkerReadyData) -> bool
True if the agent was newly registered, False if already known.
