pi_llm_agent.agent

agent

Stateful Agent wrapper — the primary public API.

The Agent class manages conversation state, tool execution, event dispatch, and message queueing. It wraps the low-level agent loop.

PendingMessageQueue(mode)

Message queue with configurable drain mode.

In "all" mode, drain() returns all queued messages at once. In "one-at-a-time" mode, drain() returns only the first message.

InitialAgentState(system_prompt='', model=None, thinking_level='off', tools=None, messages=None) dataclass

Initial state for constructing an Agent.

Attributes:

Name	Type	Description
system_prompt	str	System-level instructions for the LLM.
model	Model | None	LLM model to use.
thinking_level	AgentThinkingLevel	Extended thinking level (default "off").
tools	list[AgentTool] | None	Available tools.
messages	list[Any] | None	Pre-existing conversation history.

AgentOptions(initial_state=None, convert_to_llm=None, transform_context=None, stream_fn=None, get_api_key=None, on_payload=None, before_tool_call=None, after_tool_call=None, steering_mode='one-at-a-time', follow_up_mode='one-at-a-time', session_id=None, thinking_budgets=None, max_retry_delay_ms=None, tool_execution='parallel') dataclass

Options for constructing an Agent.

Attributes:

Name	Type	Description
initial_state	InitialAgentState | None	Initial agent state (model, tools, system prompt).
convert_to_llm	Callable[[list[Any]], list[Any] | Awaitable[list[Any]]] | None	Custom message filter for LLM calls.
transform_context	Callable[[list[Any], CancellationToken | None], Awaitable[list[Any]]] | None	Pre-process messages before each LLM call.
stream_fn	StreamFn | None	Custom streaming function (default: stream_simple).
get_api_key	Callable[[str], str | None | Awaitable[str | None]] | None	Dynamic API key resolver.
on_payload	Callable[..., Any] | None	Hook called with raw API params before each request.
before_tool_call	BeforeToolCallHook | None	Hook called before each tool execution.
after_tool_call	AfterToolCallHook | None	Hook called after each tool execution.
steering_mode	QueueMode	How steering queue drains (default "one-at-a-time").
follow_up_mode	QueueMode	How follow-up queue drains (default "one-at-a-time").
session_id	str | None	Session ID for prompt caching.
thinking_budgets	dict[str, int] | None	Custom token budgets per thinking level.
max_retry_delay_ms	int | None	Cap on retry backoff.
tool_execution	ToolExecutionMode	"parallel" (default) or "sequential".

AgentState(system_prompt='', model=None, thinking_level='off', tools=None, messages=None)

Mutable agent state with copy-on-assign for tools and messages.

Access via agent.state. Assigning to tools or messages creates a shallow copy, preventing external mutation.

Attributes:

Name	Type	Description
system_prompt		System-level instructions.
model	Model	Current LLM model.
thinking_level	AgentThinkingLevel	Current extended thinking level.
tools	list[AgentTool]	Available tools (copy-on-assign).
messages	list[Any]	Conversation transcript (copy-on-assign).
is_streaming	bool	Whether the agent is currently processing.
streaming_message	Any | None	Partial message being streamed, or None.
pending_tool_calls	set[str]	Set of tool call IDs awaiting completion.
error_message	str | None	Last error message, or None.

Agent(options=None)

Stateful agent with tool execution and event streaming.

The Agent wraps the low-level agent loop, managing the conversation transcript, tool execution lifecycle, event dispatch, and message queueing for steering and follow-up.

Example

agent = Agent(AgentOptions( ... initial_state=InitialAgentState( ... model=get_model("openai", "gpt-4o"), ... system_prompt="You are helpful.", ... tools=[my_tool], ... ), ... stream_fn=stream_simple, ... get_api_key=lambda _: os.environ["OPENAI_API_KEY"], ... )) agent.subscribe(lambda event, cancel: print(event.type)) await agent.prompt("Hello!")

cancellation property

The current cancellation token, or None if idle.

steer(message)

Queue a steering message to interrupt between turns.

follow_up(message)

Queue a follow-up message for after the current run.

clear_steering_queue()

Clear all pending steering messages.

clear_follow_up_queue()

Clear all pending follow-up messages.

clear_all_queues()

Clear both steering and follow-up queues.

has_queued_messages()

Check if either queue has pending messages.

abort()

Cancel the current processing run.

wait_for_idle() async

Wait until the agent finishes processing.

reset()

Clear all messages, state, and queues.

subscribe(listener)

Subscribe to agent events.

Parameters:

Name	Type	Description	Default
listener	Callable[[AgentEvent, CancellationToken], Awaitable[None] | None]	Callback receiving (event, cancellation_token). Can be sync or async.	required

Returns:

Type	Description
Callable[[], None]	An unsubscribe function. Call it to stop receiving events.

prompt(input, images=None) async

Send a prompt to the agent.

Parameters:

Name	Type	Description	Default
input	str | Any | list[Any]	A text string, a message object, or a list of messages.	required
images	list[ImageContent] | None	Optional images to include with a text prompt.	None

Raises:

Type	Description
RuntimeError	If the agent is already processing.

continue_() async

Continue from the current conversation state.

Resumes processing without adding a new message. The last message must be user or toolResult (not assistant).

Raises:

Type	Description
RuntimeError	If the agent is already processing or has no messages.

default_convert_to_llm(messages)

Default message filter: passes through user, assistant, and toolResult messages.