cubepi.agent

Agent

class

Agent(self, *, provider: Provider, model: Model, system_prompt: str = '', tools: list[AgentTool] | None = None, thinking: ThinkingLevel = 'off', convert_to_llm: Callable[[list[Message]], list[Message]] | None = None, transform_context: Callable | None = None, transform_system_prompt: Callable | None = None, after_model_response: Callable | None = None, before_tool_call: Callable | None = None, after_tool_call: Callable | None = None, should_stop_after_turn: Callable | None = None, on_payload: OnPayloadCallback | None = None, on_response: OnResponseCallback | None = None, steering_mode: str = 'one-at-a-time', follow_up_mode: str = 'one-at-a-time', tool_execution: str = 'parallel', checkpointer: Any = None, thread_id: str | None = None, middleware: list[Middleware] | None = None)

source

Attributes

• convert_to_llm
• transform_context
• transform_system_prompt
• after_model_response
• before_tool_call
• after_tool_call
• should_stop_after_turn
• on_payload
• on_response
• tool_execution
• checkpointer
• thread_id
• state: AgentState

Methods

subscribe

subscribe(listener: Callable) -> Callable[[], None]

source

steer

steer(message: Message) -> None

source

follow_up

follow_up(message: Message) -> None

source

abort

abort() -> None

source

wait_for_idle

wait_for_idle() -> None

source

reset

reset() -> None

source

prompt

prompt(message: str | Message | list[Message]) -> None

source

resume

resume() -> None

source

AgentState

class

AgentState(self, system_prompt: str = '', model: Model = (lambda: Model(id='unknown', provider='unknown'))(), thinking: ThinkingLevel = 'off', is_streaming: bool = False, streaming_message: Message | None = None, error_message: str | None = None, _tools: list[AgentTool] = list(), _messages: list[Message] = list(), _pending_tool_calls: set[str] = set())

source

Attributes

• system_prompt: str
• model: Model
• thinking: ThinkingLevel
• is_streaming: bool
• streaming_message: Message | None
• error_message: str | None
• tools: list[AgentTool]
• messages: list[Message]
• pending_tool_calls: set[str]

run_agent_loop

function

run_agent_loop(*, prompts: list[Message], context: AgentContext, provider: Provider, model: Model, convert_to_llm: Callable, emit: Callable, transform_context: Callable | None = None, transform_system_prompt: Callable | None = None, after_model_response: Callable | None = None, before_tool_call: Callable | None = None, after_tool_call: Callable | None = None, should_stop_after_turn: Callable | None = None, get_steering_messages: Callable | None = None, get_follow_up_messages: Callable | None = None, stream_options: StreamOptions | None = None, tool_execution: str = 'parallel', system_prompt: str | None = None) -> list[Message]

source

run_agent_loop_continue

function

run_agent_loop_continue(*, context: AgentContext, provider: Provider, model: Model, convert_to_llm: Callable, emit: Callable, transform_context: Callable | None = None, transform_system_prompt: Callable | None = None, after_model_response: Callable | None = None, before_tool_call: Callable | None = None, after_tool_call: Callable | None = None, should_stop_after_turn: Callable | None = None, get_steering_messages: Callable | None = None, get_follow_up_messages: Callable | None = None, stream_options: StreamOptions | None = None, tool_execution: str = 'parallel', system_prompt: str | None = None) -> list[Message]

source

execute_tool_calls

function

execute_tool_calls(context: AgentContext, assistant_message: AssistantMessage, *, tool_execution: str = 'parallel', before_tool_call: Callable | None = None, after_tool_call: Callable | None = None, signal: asyncio.Event | None = None, emit: Callable) -> ToolCallBatch

source

AfterToolCallContext

class

AfterToolCallContext(self, assistant_message: AssistantMessage, tool_call: ToolCall, args: Any, result: AgentToolResult, is_error: bool, context: AgentContext)

source

Attributes

• assistant_message: AssistantMessage
• tool_call: ToolCall
• args: Any
• result: AgentToolResult
• is_error: bool
• context: AgentContext

AfterToolCallResult

class

source

Attributes

• content: list[Content] | None
• details: Any
• is_error: bool | None
• terminate: bool | None

AgentContext

class

AgentContext(self, system_prompt: str, messages: list[Message], tools: list[AgentTool] | None = None, extra: dict[str, Any] = dict())

source

Attributes

• system_prompt: str
• messages: list[Message]
• tools: list[AgentTool] | None
• extra: dict[str, Any]

AgentEndEvent

class

source

Attributes

• type: Literal['agent_end']
• messages: list[Message]

AgentEvent

attribute

AgentEvent = AgentStartEvent | AgentEndEvent | TurnStartEvent | TurnEndEvent | MessageStartEvent | MessageUpdateEvent | MessageEndEvent | ToolExecutionStartEvent | ToolExecutionUpdateEvent | ToolExecutionEndEvent

source

AgentListener

attribute

AgentListener = Callable[[AgentEvent, asyncio.Event | None], Awaitable[None] | None]

source

AgentStartEvent

class

source

Attributes

• type: Literal['agent_start']

AgentTool

class

AgentTool(self, name: str, description: str, parameters: type[TParams], execute: Callable[..., Awaitable[AgentToolResult]], label: str = '', execution_mode: Literal['sequential', 'parallel'] | None = None)

source

Attributes

• name: str
• description: str
• parameters: type[TParams]
• execute: Callable[..., Awaitable[AgentToolResult]]
• label: str
• execution_mode: Literal['sequential', 'parallel'] | None

Methods

to_definition

to_definition() -> ToolDefinition

source

AgentToolResult

class

source

Attributes

• content: list[Content]
• details: Any
• is_error: bool | None
• terminate: bool | None

BeforeToolCallContext

class

BeforeToolCallContext(self, assistant_message: AssistantMessage, tool_call: ToolCall, args: Any, context: AgentContext)

source

Attributes

• assistant_message: AssistantMessage
• tool_call: ToolCall
• args: Any
• context: AgentContext

BeforeToolCallResult

class

source

Attributes

• block: bool
• reason: str | None

MessageEndEvent

class

source

Attributes

• type: Literal['message_end']
• message: Message

MessageStartEvent

class

source

Attributes

• type: Literal['message_start']
• message: Message

MessageUpdateEvent

class

source

Attributes

• type: Literal['message_update']
• message: AssistantMessage
• stream_event: StreamEvent

ShouldStopAfterTurnContext

class

ShouldStopAfterTurnContext(self, message: AssistantMessage, tool_results: list[ToolResultMessage], context: AgentContext, new_messages: list[Message])

source

Attributes

• message: AssistantMessage
• tool_results: list[ToolResultMessage]
• context: AgentContext
• new_messages: list[Message]

ToolExecutionEndEvent

class

source

Attributes

• type: Literal['tool_execution_end']
• tool_call_id: str
• tool_name: str
• result: Any
• is_error: bool
• terminate: bool — True iff the tool's AgentToolResult.terminate was True (or the after_tool_call hook set terminate=True). Recorders use this to mark the turn as terminated-by-tool without unwrapping result.
• blocked_by_hook: bool — True iff the tool call was blocked by a before_tool_call hook returning block=True. Distinguishes hook-blocks from other immediate errors (tool-not-found, arg-validation failure).
• block_reason: str | None — When blocked_by_hook is True, the reason string from BeforeToolCallResult.reason (or None if the hook supplied no reason).

ToolExecutionStartEvent

class

source

Attributes

• type: Literal['tool_execution_start']
• tool_call_id: str
• tool_name: str
• args: Any

ToolExecutionUpdateEvent

class

source

Attributes

• type: Literal['tool_execution_update']
• tool_call_id: str
• tool_name: str
• args: Any
• partial_result: Any

TurnEndEvent

class

source

Attributes

• type: Literal['turn_end']
• message: AssistantMessage
• tool_results: list[ToolResultMessage]

TurnStartEvent

class

source

Attributes

• type: Literal['turn_start']

emit_event

function

emit_event(emit_fn: Callable, event: object) -> None

source