Decorators

@tool, @task, @agent_ready, and @on_ui_event 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

cancel_on_interruption · bool · default: "True"

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

timeout · 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

Can be used with or without arguments:

# Default handler (receives unnamed requests)
@task
async def on_task_request(self, message):
    result = await do_work(message.payload)
    await self.send_task_response(result)

# Named handler (receives only "research" requests)
@task(name="research")
async def on_research(self, message):
    result = await do_research(message.payload)
    await self.send_task_response(result)

Parameters

name · str | None · default: "None"

Task name to match. When set, this handler only receives requests with a matching name. When None, handles all unnamed requests (or requests with no matching named handler).

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

name · 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}")

@on_ui_event

Mark an agent method as a handler for a named UI event.

On UIAgent subclasses, decorated methods are automatically dispatched when a BusUIEventMessage with a matching name arrives. Each handler runs in its own asyncio task so the bus dispatcher is never held open while the handler awaits downstream work.

from pipecat_subagents.agents import on_ui_event

class MyUIAgent(UIAgent):
    @on_ui_event("nav_click")
    async def on_nav(self, message):
        view = message.payload.get("view")
        ...

Parameters

name · str · required

The UI event name to match. Must match the name argument the client passes to PipecatClient.sendUIEvent(event, payload).

Method Signature

@on_ui_event handler methods receive a BusUIEventMessage:

@on_ui_event("track_played")
async def on_track_played(self, message: BusUIEventMessage):
    track_id = message.payload.get("track_id")
    ...

Two handlers cannot share the same event name on the same class. Defining a duplicate raises ValueError at agent instantiation time. Subclasses can override a base-class handler by redeclaring the method with the same name.