LLMWorker

Overview

LLMWorker extends PipelineWorker with an LLM pipeline and automatic tool registration. Pass an LLMService to the constructor and define tools with the @tool decorator. Decorated methods are registered as direct functions on the LLM and tracked so frames queued during tool execution can be deferred until all tools complete.

from pipecat.workers.llm import LLMWorker, tool

class MyAgent(LLMWorker):
    @tool
    async def my_function(self, params, arg: str):
        ...

agent = MyAgent("my_agent", llm=OpenAILLMService(api_key="..."))

Configuration

str

required

Unique name for this agent.

LLMService

required

The LLM service. @tool decorated methods are automatically registered on it.

Pipeline | None

default:"None"

Optional pipeline override. When None, defaults to Pipeline([llm]). Subclasses can pass a custom pipeline that wraps the LLM with additional processors.

bool

default:"False"

Whether the agent starts active. Defaults to False, since LLM agents typically wait to be activated.

tuple[str, ...] | None

default:"None"

Bridge configuration forwarded to PipelineWorker. Pass () to wrap the LLM pipeline with bus edge processors so it can exchange frames with another bridged agent. See PipelineWorker for details.

bool

default:"True"

Whether to defer frames queued from inside a tool handler until all tools complete. When True, frames queued via queue_frame() from inside a @tool method, or from anything that method awaits, are held in an internal queue and delivered automatically once the last tool finishes. Frames queued from elsewhere are not deferred.

Properties

Inherits all properties from PipelineWorker.

llm

agent.llm -> LLMService

The LLM service this agent wraps.

tool_call_active

agent.tool_call_active -> bool

True when one or more @tool methods are executing.

Methods

build_tools

def build_tools(self) -> list

Return the tools for this agent’s LLM. By default, returns all methods decorated with @tool. Override to provide additional or different tools. Returns: List of tool functions.

on_activated

async def on_activated(self, args: dict | None) -> None

Configure the LLM with tools and activation messages. The decorated @tool methods are set on the LLM. When args contains messages, they are appended to the LLM context. When args contains run_llm (defaults to True when messages are set), the LLM is triggered after appending.

Parameter	Type	Description
`args`	`dict \| None`	Activation arguments (see `LLMWorkerActivationArgs`)

activate_worker

async def activate_worker(
    self,
    worker_name: str,
    *,
    args: WorkerActivationArgs | None = None,
    deactivate_self: bool = False,
    messages: list | None = None,
    result_callback: FunctionCallResultCallback | None = None,
) -> None

Activate another worker, draining this worker’s pipeline to hand over. When called from a @tool handler, deliver the function call result first with await params.result_callback(result): the output it triggers is delivered before the target is activated. The handover itself waits until the tool call asking for it has finished.

Parameter	Type	Default	Description
`worker_name`	`str`		Name of the worker to activate
`args`	`WorkerActivationArgs \| None`	`None`	Arguments forwarded to `on_activated`
`deactivate_self`	`bool`	`False`	Whether to deactivate this worker before activating the target. Deactivating this worker drains its pipeline first; staying active does not. A worker that stays active and wants to drain anyway can call `flush_pipeline()` before this.
`messages`	`list \| None`	`None`	Deprecated. LLM messages to inject and deliver before activating the target. Call `params.result_callback(result)` before `activate_worker()` instead. Will be removed in 2.0.0.
`result_callback`	`FunctionCallResultCallback \| None`	`None`	Deprecated. The `result_callback` from `FunctionCallParams`. Call `params.result_callback(result)` before `activate_worker()` instead. Will be removed in 2.0.0.

end

async def end(
    self,
    *,
    reason: str | None = None,
    messages: list | None = None,
    result_callback: FunctionCallResultCallback | None = None,
) -> None

Request a graceful end of the session. When called from a @tool handler, deliver the function call result first with await params.result_callback(result): the LLM output it triggers is delivered before the session ends.

Parameter	Type	Default	Description
`reason`	`str \| None`	`None`	Human-readable reason for ending
`messages`	`list \| None`	`None`	Deprecated. LLM messages to inject and deliver before ending. Call `params.result_callback(result)` before `end()` instead. Will be removed in 2.0.0.
`result_callback`	`FunctionCallResultCallback \| None`	`None`	Deprecated. The `result_callback` from `FunctionCallParams`. Call `params.result_callback(result)` before `end()` instead. Will be removed in 2.0.0.

queue_frame

async def queue_frame(
    self,
    frame: Frame,
    direction: FrameDirection = FrameDirection.DOWNSTREAM,
) -> None

Queue a frame, holding it if a tool handler queued it. A frame queued from inside one of this worker’s @tool handlers, or from anything that handler awaits, is held and delivered once the last tool finishes. Frames from anywhere else are queued immediately: the worker’s own traffic and frames arriving over the bus, which run outside any handler’s context, and frames a handler on a different worker queues here, which this worker would never release.

Parameter	Type	Default	Description
`frame`	`Frame`		The frame to queue
`direction`	`FrameDirection`	`DOWNSTREAM`	Direction the frame should travel

process_deferred_tool_frames

async def process_deferred_tool_frames(
    self,
    frames: list[tuple[Frame, FrameDirection]],
) -> list[tuple[Frame, FrameDirection]]

Process deferred frames before they are flushed. Called after all in-flight tools complete, before the deferred frames are queued into the pipeline. Override to inspect, modify, reorder, or filter the frames.

Parameter	Type	Description
`frames`	`list[tuple[Frame, FrameDirection]]`	The deferred frames collected during tool execution

Returns: The frames to queue. Return the list as-is for default behavior.

Decorators

@tool

Mark an LLMWorker method as an LLM tool. Decorated methods are automatically registered with the LLM and included in build_tools().

from pipecat.workers.llm 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.

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.

Get Started

Pipecat Server

Client SDKs

Pipecat Flows

Pipecat Cloud

CLI

Pipecat Context Hub

Overview

Configuration

Properties

llm

tool_call_active

Methods

build_tools

on_activated

activate_worker

end

queue_frame

process_deferred_tool_frames

Decorators

@tool

Parameters

Method signature

​Overview

​Configuration

​Properties

​llm

​tool_call_active

​Methods

​build_tools

​on_activated

​activate_worker

​end

​queue_frame

​process_deferred_tool_frames

​Decorators

​@tool

​Parameters

​Method signature

Overview

Configuration

Properties

llm

tool_call_active

Methods

build_tools

on_activated

activate_worker

end

queue_frame

process_deferred_tool_frames

Decorators

@tool

Parameters

Method signature