Decorators

@tool

Mark an agent method as an LLM tool. On LLMAgent subclasses, decorated methods are automatically registered with the LLM and included in build_tools(). On FlowsAgent subclasses, decorated methods are automatically collected as global functions available at every flow node.

from pipecat_subagents.agents import tool

Can be used with or without arguments:

@tool
async def my_tool(self, params, arg: str):
    ...

@tool(cancel_on_interruption=False, timeout=60)
async def my_tool(self, params, arg: str):
    ...

Parameters

bool

default:"True"

Whether to cancel this tool call when an interruption occurs. Only applies to LLMAgent tools.

float | None

default:"None"

Timeout in seconds for this tool call. Defaults to None (uses the LLM service default).

Method Signature

Tool methods receive the LLM function call parameters object as their first argument (after self), followed by the tool’s declared parameters:

@tool
async def get_weather(self, params, location: str, unit: str = "celsius"):
    """Get the current weather for a location."""
    result = await fetch_weather(location, unit)
    await params.result_callback(result)

Tool methods must call params.result_callback() to return a result to the LLM. The method signature (parameter names, types, and docstring) is automatically used to generate the tool schema.

@task

Mark an agent method as a task handler. Decorated methods are automatically collected by BaseAgent at initialization and dispatched when matching task requests arrive. Each request runs in its own asyncio task so the bus message loop is never blocked.

from pipecat_subagents.agents import task

The decorator requires a name argument:

@task(name="research")
async def on_research(self, message):
    result = await do_research(message.payload)
    await self.send_task_response(message.task_id, result)

@task(name="write", sequential=True)
async def on_write(self, message):
    result = await do_write(message.payload)
    await self.send_task_response(message.task_id, result)

Parameters

str

required

Task name to match. The handler only receives requests with a matching name.

bool

default:"False"

When True, requests with this name run one at a time in FIFO order. Concurrent requests wait for the previous one to finish before running. When False (the default), multiple requests run concurrently. The wait time counts against the requester’s timeout, so a slow predecessor can cause queued requests to time out before they start.

Method Signature

Task handler methods receive a BusTaskRequestMessage:

@task(name="analyze")
async def on_analyze(self, message: BusTaskRequestMessage):
    data = message.payload
    # Send progress updates
    await self.send_task_update({"progress": 50})
    # Send final response
    await self.send_task_response({"result": "done"})

@agent_ready

Mark a method as a handler for a specific agent becoming ready. Decorated methods are automatically collected by BaseAgent at initialization. When on_ready fires, the agent calls watch_agent for each decorated handler. When the watched agent registers, the decorated method is called.

from pipecat_subagents.agents import agent_ready

@agent_ready(name="greeter")
async def on_greeter_ready(self, data: AgentReadyData) -> None:
    await self.activate_agent("greeter", args=...)

Parameters

str

required

The name of the agent to watch.

Method Signature

Agent-ready handler methods receive an AgentReadyData instance:

@agent_ready(name="worker")
async def on_worker_ready(self, data: AgentReadyData) -> None:
    print(f"Agent {data.agent_name} is ready on runner {data.runner}")

Pipecat Server

Pipecat Subagents

Client SDKs

Pipecat Flows

Pipecat Cloud

CLI

@tool

Parameters

Method Signature

@task

Parameters

Method Signature

@agent_ready

Parameters

Method Signature

​@tool

​Parameters

​Method Signature

​@task

​Parameters

​Method Signature

​@agent_ready

​Parameters

​Method Signature

@tool

Parameters

Method Signature

@task

Parameters

Method Signature

@agent_ready

Parameters

Method Signature