> ## Documentation Index
> Fetch the complete documentation index at: https://daily-ms-ws-body-url-encode.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Types

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

## JobParams

```python theme={null}
from pipecat.pipeline.job_context import JobParams
```

Everything a job dispatch needs, in one object. Pass it as `params=` to [`job()`](/api-reference/server/workers/base-worker#job) or [`request_job()`](/api-reference/server/workers/base-worker#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`](/api-reference/server/workers/ui-worker) 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 |

```python theme={null}
from pipecat.pipeline.job_context import JobParams

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

<Note>
  `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.
</Note>

## JobGroupParams

```python theme={null}
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()`](/api-reference/server/workers/base-worker#job_group), [`request_job_group()`](/api-reference/server/workers/base-worker#request_job_group), or [`create_job_group_and_request_job()`](/api-reference/server/workers/base-worker#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 |

```python theme={null}
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:
    results = await jg.wait()
```

## JobStatus

```python theme={null}
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

```python theme={null}
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

```python theme={null}
# Create from a dict (ignores unknown keys)
args = WorkerActivationArgs.from_dict({"metadata": {...}})

# Convert to a dict (excludes None values)
data = args.to_dict()
```

## LLMWorkerActivationArgs

```python theme={null}
from pipecat.workers.llm import LLMWorkerActivationArgs
```

Activation arguments for [`LLMWorker`](/api-reference/server/workers/llm-worker). Extends [`WorkerActivationArgs`](#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

```python theme={null}
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`](/api-reference/server/workers/exceptions#joberror).

### Properties

| Property | Type | Description |
| - | - | - |
| `job_id` | `str` | The job identifier |
| `response` | `dict` | The worker's response payload |

### Usage

```python theme={null}
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

```python theme={null}
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`](/api-reference/server/workers/exceptions#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

```python theme={null}
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

```python theme={null}
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`](#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

```python theme={null}
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

```python theme={null}
from pipecat.pipeline.job_context import JobGroup
```

Tracks a group of agents launched together by a job request. Exposed via [`BaseWorker.job_groups`](/api-reference/server/workers/base-worker#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

```python theme={null}
# Wait for all agents in the group to respond (raises JobGroupError on failure)
await group.wait()
```

## JobGroupEvent

```python theme={null}
from pipecat.pipeline.job_context import JobGroupEvent
```

An event received from a worker during job group execution. Yielded by `async for` over a [`JobGroupContext`](#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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
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

```python theme={null}
from pipecat.registry import WorkerRegistry
```

Tracks all known agents across local and remote runners. Owned by the [`WorkerRunner`](/api-reference/server/workers/runner) 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

```python theme={null}
def get(self, worker_name: str) -> WorkerReadyData | None
```

Look up a registered agent by name.

#### watch

```python theme={null}
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 |

#### register

```python theme={null}
async def register(self, worker_data: WorkerReadyData) -> bool
```

Register an agent. Returns `True` if the agent was newly registered, `False` if already known.