Types

TaskStatus

from pipecat_subagents.agents.task_context import TaskStatus

Status of a completed task. Inherits from str so values compare naturally with plain strings.

Value	Description
`TaskStatus.COMPLETED`	The task finished successfully
`TaskStatus.CANCELLED`	The task was cancelled by the requester
`TaskStatus.FAILED`	The task failed due to a logical or business error
`TaskStatus.ERROR`	The task encountered an unexpected runtime error

AgentActivationArgs

from pipecat_subagents.agents.base_agent import AgentActivationArgs

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 = AgentActivationArgs.from_dict({"metadata": {...}})

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

LLMAgentActivationArgs

from pipecat_subagents.agents.llm_agent import LLMAgentActivationArgs

Activation arguments for LLMAgent. Extends AgentActivationArgs.

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

TaskContext

from pipecat_subagents.agents.task_context import TaskContext

Async context manager and iterator for a single-agent task. Sends a task 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 TaskError.

Properties

Property	Type	Description
`task_id`	`str`	The task identifier
`response`	`dict`	The worker’s response payload

Usage

async with self.task("worker", payload=data) as t:
    async for event in t:
        print(f"[{event.type}]: {event.data}")

print(t.response)

TaskGroupContext

from pipecat_subagents.agents.task_context import TaskGroupContext

Async context manager and iterator for structured task group execution. Sends task 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 TaskGroupError.

Properties

Property	Type	Description
`task_id`	`str`	The shared task identifier for this group
`responses`	`dict[str, dict]`	Collected responses keyed by agent name

Usage

async with self.task_group("w1", "w2", payload=data) as tg:
    async for event in tg:
        print(f"{event.agent_name} [{event.type}]: {event.data}")

for name, result in tg.responses.items():
    print(name, result)

TaskGroupResponse

from pipecat_subagents.agents.task_context import TaskGroupResponse

Collected results from a completed task group. Passed to on_task_completed.

Field	Type	Description
`task_id`	`str`	The shared task identifier
`responses`	`dict[str, dict]`	Collected responses keyed by agent name

TaskGroupEvent

from pipecat_subagents.agents.task_context import TaskGroupEvent

An event received from a worker during task group execution.

Field	Type	Description
`type`	`str`	The event type (see constants below)
`agent_name`	`str`	The name of the agent that sent the event
`data`	`dict	None`	Optional event payload

Event Type Constants

Constant	Value	Description
`TaskGroupEvent.UPDATE`	`"update"`	Progress update from a worker
`TaskGroupEvent.STREAM_START`	`"stream_start"`	Worker started streaming
`TaskGroupEvent.STREAM_DATA`	`"stream_data"`	Streaming data chunk
`TaskGroupEvent.STREAM_END`	`"stream_end"`	Worker finished streaming

AgentReadyData

from pipecat_subagents.types import AgentReadyData

Information about a registered agent. Passed to on_agent_ready and @agent_ready handlers.

Field	Type	Description
`agent_name`	`str`	The name of the agent
`runner`	`str`	The name of the runner managing this agent

AgentErrorData

from pipecat_subagents.types import AgentErrorData

Information about an agent error. Passed to on_agent_error.

Field	Type	Description
`agent_name`	`str`	The name of the agent that errored
`error`	`str`	Description of the error

AgentRegistryEntry

from pipecat_subagents.types import AgentRegistryEntry

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

AgentRegistry

from pipecat_subagents.registry import AgentRegistry

Tracks all known agents across local and remote runners. Owned by the AgentRunner and shared with its agents.

Properties

Property	Type	Description
`runner_name`	`str`	The name of the runner that owns this registry
`local_agents`	`list[str]`	Names of agents registered under this runner
`remote_agents`	`list[str]`	Names of agents registered under remote runners

Methods

get

def get(self, agent_name: str) -> AgentReadyData | None

Look up a registered agent by name.

watch

async def watch(self, agent_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
`agent_name`	`str`	The agent name to watch for
`handler`	`Callable[[AgentReadyData], Coroutine]`	Async callable invoked with the agent’s data

register

async def register(self, agent_data: AgentReadyData) -> bool

Pipecat Server

Pipecat Subagents

Client SDKs

Pipecat Flows

Pipecat Cloud

CLI

TaskStatus

AgentActivationArgs

Methods

LLMAgentActivationArgs

TaskContext

Properties

Usage

TaskGroupContext

Properties

Usage

TaskGroupResponse

TaskGroupEvent

Event Type Constants

AgentReadyData

AgentErrorData

AgentRegistryEntry

AgentRegistry

Properties

Methods

get

watch

register

​TaskStatus

​AgentActivationArgs

​Methods

​LLMAgentActivationArgs

​TaskContext

​Properties

​Usage

​TaskGroupContext

​Properties

​Usage

​TaskGroupResponse

​TaskGroupEvent

​Event Type Constants

​AgentReadyData

​AgentErrorData

​AgentRegistryEntry

​AgentRegistry

​Properties

​Methods

​get

​watch

​register

TaskStatus

AgentActivationArgs

Methods

LLMAgentActivationArgs

TaskContext

Properties

Usage

TaskGroupContext

Properties

Usage

TaskGroupResponse

TaskGroupEvent

Event Type Constants

AgentReadyData

AgentErrorData

AgentRegistryEntry

AgentRegistry

Properties

Methods

get

watch

register