Types

Shared types for Pipecat workers and job coordination: JobStatus, job context enums, and data classes.

​

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