---
id: cubepi-agent
title: cubepi.agent
sidebar_position: 1
hide_table_of_contents: false
---

# `cubepi.agent`

### Agent {#agent}

_class_

```python
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](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/agent.py#L107)


**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` {#agent-subscribe}

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

[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/agent.py#L175)


#### `steer` {#agent-steer}

```python
steer(message: Message) -> None
```

[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/agent.py#L181)


#### `follow_up` {#agent-follow-up}

```python
follow_up(message: Message) -> None
```

[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/agent.py#L184)


#### `abort` {#agent-abort}

```python
abort() -> None
```

[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/agent.py#L187)


#### `wait_for_idle` {#agent-wait-for-idle}

```python
wait_for_idle() -> None
```

[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/agent.py#L191)


#### `reset` {#agent-reset}

```python
reset() -> None
```

[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/agent.py#L195)


#### `prompt` {#agent-prompt}

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

[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/agent.py#L204)


#### `resume` {#agent-resume}

```python
resume() -> None
```

[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/agent.py#L230)



### AgentState {#agentstate}

_class_

```python
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](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/agent.py#L68)


**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 {#run-agent-loop}

_function_

```python
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](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/loop.py#L30)


### run_agent_loop_continue {#run-agent-loop-continue}

_function_

```python
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](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/loop.py#L87)


### execute_tool_calls {#execute-tool-calls}

_function_

```python
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](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/tools.py#L206)


### AfterToolCallContext {#aftertoolcallcontext}

_class_

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


[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/types.py#L79)


**Attributes**

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


### AfterToolCallResult {#aftertoolcallresult}

_class_


[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/types.py#L64)


**Attributes**

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


### AgentContext {#agentcontext}

_class_

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


[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/types.py#L48)


**Attributes**

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


### AgentEndEvent {#agentendevent}

_class_


[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/types.py#L104)


**Attributes**

- `type`: `Literal['agent_end']`
- `messages`: `list[Message]`


### AgentEvent {#agentevent}

_attribute_

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


[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/types.py#L170)


### AgentListener {#agentlistener}

_attribute_

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


[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/types.py#L185)


### AgentStartEvent {#agentstartevent}

_class_


[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/types.py#L100)


**Attributes**

- `type`: `Literal['agent_start']`


### AgentTool {#agenttool}

_class_

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


[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/types.py#L30)


**Attributes**

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


**Methods**

#### `to_definition` {#agenttool-to-definition}

```python
to_definition() -> ToolDefinition
```

[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/types.py#L39)



### AgentToolResult {#agenttoolresult}

_class_


[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/types.py#L23)


**Attributes**

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


### BeforeToolCallContext {#beforetoolcallcontext}

_class_

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


[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/types.py#L71)


**Attributes**

- `assistant_message`: `AssistantMessage`
- `tool_call`: `ToolCall`
- `args`: `Any`
- `context`: `AgentContext`


### BeforeToolCallResult {#beforetoolcallresult}

_class_


[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/types.py#L59)


**Attributes**

- `block`: `bool`
- `reason`: `str | None`


### MessageEndEvent {#messageendevent}

_class_


[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/types.py#L130)


**Attributes**

- `type`: `Literal['message_end']`
- `message`: `Message`


### MessageStartEvent {#messagestartevent}

_class_


[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/types.py#L119)


**Attributes**

- `type`: `Literal['message_start']`
- `message`: `Message`


### MessageUpdateEvent {#messageupdateevent}

_class_


[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/types.py#L124)


**Attributes**

- `type`: `Literal['message_update']`
- `message`: `AssistantMessage`
- `stream_event`: `StreamEvent`


### ShouldStopAfterTurnContext {#shouldstopafterturncontext}

_class_

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


[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/types.py#L89)


**Attributes**

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


### ToolExecutionEndEvent {#toolexecutionendevent}

_class_


[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/types.py#L150)


**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 {#toolexecutionstartevent}

_class_


[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/types.py#L135)


**Attributes**

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


### ToolExecutionUpdateEvent {#toolexecutionupdateevent}

_class_


[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/types.py#L142)


**Attributes**

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


### TurnEndEvent {#turnendevent}

_class_


[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/types.py#L113)


**Attributes**

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


### TurnStartEvent {#turnstartevent}

_class_


[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/agent/types.py#L109)


**Attributes**

- `type`: `Literal['turn_start']`


### emit_event {#emit-event}

_function_

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


[source](https://github.com/cubeplexai/cubepi/blob/main/cubepi/cubepi/utils/emit.py#L7)



<!-- GENERATED by build_api_reference.py — DO NOT EDIT -->