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

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

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

AfterToolCallResult

class

source

AgentContext

class

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

source

AgentEndEvent

class

source

AgentEvent

attribute

source

AgentListener

attribute

source

AgentStartEvent

class

source

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

AgentToolResult

class

source

BeforeToolCallContext

class

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

source

BeforeToolCallResult

class

source

MessageEndEvent

class

source

MessageStartEvent

class

source

MessageUpdateEvent

class

source

ShouldStopAfterTurnContext

class

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

source

ToolExecutionEndEvent

class

source

ToolExecutionStartEvent

class

source

ToolExecutionUpdateEvent

class

source

TurnEndEvent

class

source

TurnStartEvent

class

source

emit_event

function

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

source