Types

JobParams

from pipecat.pipeline.job_context import JobParams

Everything a job dispatch needs, in one object. Pass it as params= to job() or request_job().

Field	Type	Default	Description
`name`	`str \| None`	`None`	Routes the request to the matching `@job(name=...)` handler on the worker
`payload`	`dict \| None`	`None`	Structured data describing the work
`timeout`	`float \| None`	`None`	Seconds, covering both the wait for the worker to become ready and the job itself
`label`	`str \| None`	`None`	Human-readable description, e.g. `"Research: Radiohead"`. A `UIWorker` titles the client’s progress card with it
`cancellable`	`bool`	`True`	Whether an external requester, such as the client UI, may ask for the job to be cancelled

from pipecat.pipeline.job_context import JobParams

async with self.job("researcher", params=JobParams(payload={"topic": topic})) as j:
    pass
result = j.response

cancellable governs requests from outside the worker. Cancellation the worker initiates itself — on shutdown, on a timeout, or through cancel_on_error — proceeds either way.

JobGroupParams

from pipecat.pipeline.job_context import JobGroupParams

Carries every JobParams field, which apply to the group as a whole, plus one of its own. Pass it as params= to job_group(), request_job_group(), or create_job_group_and_request_job().

Field	Type	Default	Description
`cancel_on_error`	`bool`	`True`	Whether a worker responding with an error cancels the rest of the group

from pipecat.pipeline.job_context import JobGroupParams

async with self.job_group(
    "wikipedia", "news",
    params=JobGroupParams(payload={"query": query}, label=f"Research: {query}"),
) as jg:
    pass
results = jg.responses

JobStatus

from pipecat.pipeline.job_context import JobStatus

Status of a completed job. Inherits from 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.workers.base_worker import WorkerActivationArgs

Base activation arguments for any agent.

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

Activation arguments for 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 context manager and iterator for a single-agent job. Sends a job request on enter, waits for the response on exit. Supports async for to receive intermediate events. On normal completion, the result is available via response. On worker error or timeout, raises JobError. If the worker isn’t ready within the timeout, entering the block raises JobGroupError instead.

Properties

Property	Type	Description
`job_id`	`str`	The job identifier
`response`	`dict`	The worker’s response payload

Usage

async with self.job("worker", params=JobParams(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 context manager and iterator for structured job group execution. Sends job requests on enter, waits for all responses on exit. Supports 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", params=JobGroupParams(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

An event received from a worker during a single-agent job. Yielded by 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

Collected results from a completed job group. Passed to 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

Tracks a group of agents launched together by a job request. Exposed via 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`	`list[str]`		Names of the agents in the group, in the order they were dispatched
`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
`label`	`str \| None`	`None`	Human-readable description of the work, shown on the client’s card
`cancellable`	`bool`	`True`	Whether an external requester may ask for the group to be cancelled
`terminated`	`set[str]`	`set()`	Names of the workers that have finished

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

An event received from a worker during job group execution. Yielded by 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

Information about a registered agent. Passed to 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

Information about an agent error. Passed to 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

Information about an agent in a registry snapshot.

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

Tracks all known agents across local and remote runners. Owned by the 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

Look up a registered agent by name.

watch

async def watch(self, worker_name: str, handler: WatchHandler) -> None

Watch for a specific agent’s registration. If the agent is already registered, the handler fires immediately.

Parameter	Type	Description
`worker_name`	`str`	The agent name to watch for
`handler`	`Callable[[WorkerReadyData], Coroutine]`	Async callable invoked with the agent’s data

Get Started

Standards

Pipecat Server

Client SDKs

Pipecat Flows

Pipecat Cloud

CLI

Pipecat Context Hub

JobParams

JobGroupParams

JobStatus

WorkerActivationArgs

Methods

LLMWorkerActivationArgs

JobContext

Properties

Usage

JobGroupContext

Properties

Usage

JobEvent

Event Type Constants

JobGroupResponse

JobGroup

Properties

Methods

JobGroupEvent

Event Type Constants

WorkerReadyData

WorkerErrorData

WorkerRegistryEntry

WorkerRegistry

Properties

Methods

get

watch

​JobParams

​JobGroupParams

​JobStatus

​WorkerActivationArgs

​Methods

​LLMWorkerActivationArgs

​JobContext

​Properties

​Usage

​JobGroupContext

​Properties

​Usage

​JobEvent

​Event Type Constants

​JobGroupResponse

​JobGroup

​Properties

​Methods

​JobGroupEvent

​Event Type Constants

​WorkerReadyData

​WorkerErrorData

​WorkerRegistryEntry

​WorkerRegistry

​Properties

​Methods

​get

​watch

JobParams

JobGroupParams

JobStatus

WorkerActivationArgs

Methods

LLMWorkerActivationArgs

JobContext

Properties

Usage

JobGroupContext

Properties

Usage

JobEvent

Event Type Constants

JobGroupResponse

JobGroup

Properties

Methods

JobGroupEvent

Event Type Constants

WorkerReadyData

WorkerErrorData

WorkerRegistryEntry

WorkerRegistry

Properties

Methods

get

watch