Skip to content
Roboto
Esc
↑↓navigate↵open⌘Jpreview
On this page

roboto.ai.agent_thread

Submodules

Package Contents

AGENT_CONTENT_MODEL_BY_TYPE

roboto.ai.agent_thread.AGENT_CONTENT_MODEL_BY_TYPE: dict[AgentContentType, type[AgentContent]]#View Source

The model class for each JSON-serialized content type, keyed by discriminator.

Excludes AgentContentType.TEXT, whose payload is persisted as raw text rather than a serialized model. Every other member of AgentContent carries a content_type discriminator and round-trips through model_dump_json / model_validate_json; driving both serialization directions off this one map keeps them symmetric, so a member added to the union without a home here fails loudly instead of being silently dropped on write or reconstructed without its payload on read.

AdminUpdateFeedbackRequest

class roboto.ai.agent_thread.AdminUpdateFeedbackRequest(/, **data)#View Source

Bases: pydantic.BaseModel

Triage fields editable by Roboto admins.

Fields omitted from the request (NotSet) are left unchanged. Fields explicitly set to None clear the column back to NULL. Fields set to a value overwrite the column.

Setting resolved to True additionally stamps resolved_at and resolved_by server-side; setting it back to False clears them.

Parameters

data Any

Attributes

AdminUpdateFeedbackRequest.admin_label

admin_label str | None | roboto.sentinels.NotSetType #

AdminUpdateFeedbackRequest.admin_note

admin_note str | None | roboto.sentinels.NotSetType #

AdminUpdateFeedbackRequest.model_config

model_config #

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

AdminUpdateFeedbackRequest.resolved

AgentClientContextEntry

class roboto.ai.agent_thread.AgentClientContextEntry(/, **data)#View Source

Bases: pydantic.BaseModel

The caller’s attached viewing context, carried on their own message.

Replaces a <ctx>{json}</ctx> marker persisted as a separate ROBOTO-role message. That shape made every consumer regex a JSON blob back out of prose it had serialized itself – four parsers across two languages – and relied on the ROBOTO role to keep the marker out of the chat view. It also put the context on a non-USER message, which the compression deletion pass collapses wholesale inside a resolved task’s interior, so the context silently disappeared from compressed history.

As a block on the user’s own message it is a typed field, invisible to text renderers (no text field), and safe from that collapse – USER turns are never dropped.

Parameters

data Any

Attributes

AgentClientContextEntry.content_type

content_type Literal[AgentContentType] #

AgentClientContextEntry.context

What the caller had open: attached datasets, files, and visualizer state.

AgentCompressionFillerContent

class roboto.ai.agent_thread.AgentCompressionFillerContent(/, **data)#View Source

Bases: pydantic.BaseModel

Filler standing in for a message the compression deletion pass emptied.

Carries no payload. A message reduced to nothing but tombstones keeps this single block instead of an empty content list, so it stays in the Bedrock payload at its original role and turn alternation survives without relocating content. The Bedrock boundary renders it as <Deleted in compression>. Produced only by the compression deletion pass and stored only at the DELETED tier; the verbatim original thread (what the SDK and UI render) never contains one.

Parameters

data Any

Attributes

AgentCompressionFillerContent.content_type

content_type Literal[AgentContentType] #

AgentContent

Type alias for all possible content types within agent messages.

AgentContentType

class roboto.ai.agent_thread.AgentContentType#View Source

Bases: roboto.compat.StrEnum

Enumeration of different types of content within agent messages.

Defines the various content types that can be included in agent messages.

Attributes

AgentContentType.CLIENT_CONTEXT

CLIENT_CONTEXT = 'client_context' #

What the caller was looking at when they composed the message this block sits on.

Superseded shape: the same payload used to be persisted as a whole ROBOTO-role message whose text was a <ctx>...</ctx> marker (ENG-2185). Readers still accept that form for threads written before this type existed; nothing emits it any more.

AgentContentType.COMPRESSION_FILLER

COMPRESSION_FILLER = 'compression_filler' #

Stand-in block kept in a message the deletion pass emptied entirely.

A message reduced to nothing but tombstones would break user/assistant alternation if it dropped from the payload. Replacing its content with this single filler keeps the message — and its role — in place. The model sees it as <Deleted in compression>. Only the cross-message deletion pass produces it, so it appears only inside a DELETED-tier compressed variant, never in the verbatim original thread the SDK and UI read.

AgentContentType.DELETED

DELETED = 'deleted' #

Tombstone marking a content block elided by compression.

Appears only inside a DELETED-tier compressed variant. Both producers — message-tier compression dropping a pure-filler text run, and the cross-message deletion pass dropping a whole tool exchange — store their output at the DELETED tier, so a message carrying one is always a DELETED-tier variant. Never in the verbatim original thread the SDK and UI read.

AgentContentType.ERROR

ERROR = 'error' #

Error information when message generation fails.

AgentContentType.TEXT

TEXT = 'text' #

Plain text content from users or AI responses.

AgentContentType.TOOL_RESULT

TOOL_RESULT = 'tool_result' #

Results returned from tool executions.

AgentContentType.TOOL_USE

TOOL_USE = 'tool_use' #

Tool invocation requests from the AI assistant.

AgentDeletedContent

class roboto.ai.agent_thread.AgentDeletedContent(/, **data)#View Source

Bases: pydantic.BaseModel

Tombstone for a content block removed by compression.

Carries no payload — its presence records that a block once occupied this slot, and it converts to None at the Bedrock boundary so the block drops from the LLM payload. Produced when compression drops a block — a pure-filler text run at message compression, or a whole redundant tool exchange in the cross-message deletion pass — and stored only at the DELETED tier, so a message carrying one is always a DELETED-tier variant. A message reduced to nothing but tombstones does not drop: it is replaced with a single AgentCompressionFillerContent so it keeps its role and turn. The verbatim original thread (what the SDK and UI render) never contains one.

Parameters

data Any

Attributes

AgentDeletedContent.content_type

content_type Literal[AgentContentType] #

AgentErrorContent

class roboto.ai.agent_thread.AgentErrorContent(/, **data)#View Source

Bases: pydantic.BaseModel

Error content within an agent message.

Used when message generation fails due to an error or is cancelled by the user.

Parameters

data Any

Attributes

AgentErrorContent.content_type

content_type Literal[AgentContentType] #

AgentErrorContent.error_code

error_code str | None = None #

Optional error code for programmatic handling.

AgentErrorContent.error_message

error_message str #

User-friendly error message describing what went wrong.

AgentErrorEvent

class roboto.ai.agent_thread.AgentErrorEvent(/, **data)#View Source

Bases: pydantic.BaseModel

Signals that message generation failed or was cancelled.

Parameters

data Any

Attributes

AgentErrorEvent.error_code

error_code str | None = None #

Optional error code for programmatic handling.

AgentErrorEvent.error_message

error_message str #

User-friendly error message describing what went wrong.

AgentEvent

type roboto.ai.agent_thread.AgentEvent = typing.Union[AgentStartTextEvent, AgentTextDeltaEvent, AgentTextEndEvent, AgentToolUseEvent, AgentToolResultEvent, AgentErrorEvent]#View Source

AgentGoalStatus

class roboto.ai.agent_thread.AgentGoalStatus#View Source

Bases: roboto.compat.StrEnum

Lifecycle of a per-turn declared goal.

Goals begin PENDING when registered. They transition to ACHIEVED when the corresponding achieve-tool reports success, or to FAILED when the runner’s corrective re-prompt budget for the turn is exhausted (or when the worker cannot construct an achieve-tool for the goal).

Attributes

AgentGoalStatus.ACHIEVED

ACHIEVED = 'achieved' #

Goal’s corresponding achieve-tool was invoked successfully.

AgentGoalStatus.FAILED

FAILED = 'failed' #

Goal could not be achieved within the turn’s retry budget.

AgentGoalStatus.PENDING

PENDING = 'pending' #

Goal has been registered but not yet completed.

AgentMessage

class roboto.ai.agent_thread.AgentMessage(/, **data)#View Source

Bases: pydantic.BaseModel

A single message within an agent thread.

Represents one message in the conversation, containing the sender role, content blocks, and generation status. Messages can contain multiple content blocks of different types (text, tool use, tool results).

Parameters

data Any

Attributes

AgentMessage.content

List of content blocks that make up this message.

AgentMessage.created

created datetime.datetime = None #

Timestamp when this message was created.

AgentMessage.is_complete()

is_complete()#View Source

Check if message generation is complete.

Returns

bool

True if the message status is COMPLETED, False otherwise.

AgentMessage.is_unsuccessful()

is_unsuccessful()#View Source

Check if message generation failed or was cancelled.

Returns

bool

True if the message status is FAILED or CANCELLED, False otherwise.

Attributes

AgentMessage.role

The role of the message sender (user, assistant, or roboto).

AgentMessage.status

Current generation status of this message.

AgentMessage.text()

classmethod text(text, role=AgentRole.USER)#View Source

Create a simple text message.

Convenience method for creating a message containing only text content.

Parameters

text str

The text content for the message.

The role of the message sender. Defaults to USER.

Returns

AgentMessage instance containing the text content.

AgentMessageStatus

class roboto.ai.agent_thread.AgentMessageStatus#View Source

Bases: roboto.compat.StrEnum

Enumeration of possible message generation states.

Tracks the lifecycle of message generation from initiation to completion.

Attributes

AgentMessageStatus.CANCELLED

CANCELLED = 'cancelled' #

Message generation was cancelled by the user.

AgentMessageStatus.COMPLETED

COMPLETED = 'completed' #

Message generation has finished and content is complete.

AgentMessageStatus.FAILED

FAILED = 'failed' #

Message generation failed due to an error.

AgentMessageStatus.GENERATING

GENERATING = 'generating' #

Message content is currently being generated.

AgentMessageStatus.NOT_STARTED

NOT_STARTED = 'not_started' #

Message has been queued but generation has not begun.

AgentMessageStatus.is_terminal()

is_terminal()#View Source

Check if the message generation is in a terminal state.

Returns

bool

True if the message is in a terminal state, False otherwise.

AgentRole

class roboto.ai.agent_thread.AgentRole#View Source

Bases: roboto.compat.StrEnum

Enumeration of possible roles in an agent thread.

Defines the different participants that can send messages in a thread.

Attributes

AgentRole.ASSISTANT

ASSISTANT = 'assistant' #

AI agent responding to user queries and requests.

AgentRole.ROBOTO

ROBOTO = 'roboto' #

Roboto system providing tool results and system information.

AgentRole.USER

USER = 'user' #

Human user sending messages to the agent.

AgentStartTextEvent

class roboto.ai.agent_thread.AgentStartTextEvent(/, **data)#View Source

Bases: pydantic.BaseModel

Signals the beginning of text generation in a chat response.

Parameters

data Any

AgentSubtask

class roboto.ai.agent_thread.AgentSubtask(/, **data)#View Source

Bases: pydantic.BaseModel

A lightweight checklist item under a top-level task.

Parameters

data Any

Attributes

AgentSubtask.status

Whether the sub-task is done.

AgentSubtask.title

title str #

Human-readable description of the sub-task.

AgentSubtaskStatus

class roboto.ai.agent_thread.AgentSubtaskStatus#View Source

Bases: roboto.compat.StrEnum

Lifecycle state of a sub-task.

Sub-tasks are never activated independently — they are implicitly active with their parent task — so they have no in_progress state.

Attributes

AgentSubtaskStatus.DONE

DONE = 'done' #

Completed.

AgentSubtaskStatus.PENDING

PENDING = 'pending' #

Not done yet.

AgentTask

class roboto.ai.agent_thread.AgentTask(/, **data)#View Source

Bases: pydantic.BaseModel

A top-level task the agent is tracking within a thread.

Parameters

data Any

Attributes

AgentTask.conclusion

conclusion str | None = None #

Outcome recorded when the task was completed. None until then.

AgentTask.description

description str | None = None #

Longer description delimiting the task’s scope and intent.

AgentTask.end

end AgentTaskBoundary | None = None #

Where the task was completed. None until then.

AgentTask.position

position int #

Display order among the thread’s top-level tasks.

AgentTask.start

start AgentTaskBoundary | None = None #

Where the task most recently became in_progress. None if never started.

AgentTask.status

Current lifecycle state.

AgentTask.subtasks

subtasks list[AgentSubtask] = None #

Ordered sub-tasks. A task cannot be completed until all of these are done.

AgentTask.task_id

task_id int #

Thread-monotonic id, stable for the life of the thread. The model references this id to start or complete the task.

AgentTask.title

title str #

Short title of the task.

AgentTaskBoundary

class roboto.ai.agent_thread.AgentTaskBoundary(/, **data)#View Source

Bases: pydantic.BaseModel

Position in the conversation where a top-level task’s active span begins or ends.

Identifies the assistant message and the content block within it whose tool call drove the transition, so callers can anchor a task to the part of the thread that worked on it.

Parameters

data Any

Attributes

AgentTaskBoundary.content_sequence_num

content_sequence_num int #

Index of the tool-call content block within that message.

AgentTaskBoundary.message_sequence_num

message_sequence_num int #

Index of the assistant message whose tool call stamped this boundary.

AgentTaskMinimal

class roboto.ai.agent_thread.AgentTaskMinimal(/, **data)#View Source

Bases: pydantic.BaseModel

Minimal acknowledgement returned by the task mutation tools.

The full list reaches the model through the per-turn injected context, so the mutation tools echo only the affected task’s id and resulting status.

Parameters

data Any

Attributes

AgentTaskMinimal.status

Resulting status of the affected task.

AgentTaskMinimal.task_id

task_id int #

Id of the affected task.

AgentTaskStatus

class roboto.ai.agent_thread.AgentTaskStatus#View Source

Bases: roboto.compat.StrEnum

Lifecycle state of a top-level agent task.

Attributes

AgentTaskStatus.COMPLETED

COMPLETED = 'completed' #

Finished; carries a AgentTask.conclusion.

AgentTaskStatus.IN_PROGRESS

IN_PROGRESS = 'in_progress' #

The single active task. At most one task per thread is in this state.

AgentTaskStatus.PENDING

PENDING = 'pending' #

Not started yet.

AgentTextContent

class roboto.ai.agent_thread.AgentTextContent(/, **data)#View Source

Bases: pydantic.BaseModel

Text content within an agent message.

Parameters

data Any

Attributes

AgentTextContent.text

text str #

The actual text content of the message.

AgentTextDeltaEvent

class roboto.ai.agent_thread.AgentTextDeltaEvent(/, **data)#View Source

Bases: pydantic.BaseModel

Contains incremental text content as the AI generates its response.

Parameters

data Any

Attributes

AgentTextDeltaEvent.text

text str #

Text fragment from the streaming response.

AgentTextEndEvent

class roboto.ai.agent_thread.AgentTextEndEvent(/, **data)#View Source

Bases: pydantic.BaseModel

Signals the completion of text generation in a chat response.

Parameters

data Any

AgentThread

class roboto.ai.agent_thread.AgentThread(record, roboto_client=None, client_tools=None)#View Source

An interactive AI agent session within the Roboto platform.

An AgentThread is a conversational interface with Roboto’s AI assistant, enabling users to ask questions, request data analysis, and interact with their robotics data through natural language. Sessions maintain conversation history and support streaming responses for real-time interaction.

The primary control-flow primitives are run() (drive the session forward with auto-dispatch of client-side tools) and events() (observe events as the agent generates without taking any actions).

Usage

Fire-and-forget with client-side tools:

from roboto.ai import AgentThread, client_tool
@client_tool
def remember(fact: str) -> str:
    """Store a fact in long-term memory."""
    ...
session = AgentThread.start("Remember my favorite color is blue.", client_tools=[remember])
session.run()

Observing events as they happen:

session = AgentThread.start("Explain machine learning.")
for event in session.events():
    if isinstance(event, AgentTextDeltaEvent):
        print(event.text, end="", flush=True)

Properties

AgentThread.client_tool_names

client_tool_names list[str] #

Names of client-side tools registered on this session with callbacks.

Return type: list[str]

AgentThread.events()

events(tick=0.2, timeout=None)#View Source

Yield events from the agent as they are generated.

Polls the session and yields AgentEvent objects as new content arrives. Does not auto-dispatch client-side tools — if the session reaches AgentThreadStatus.CLIENT_TOOL_TURN, the generator returns and the caller is expected to call submit_client_tool_results() (and then call events() again to continue). For automatic dispatch, use run().

Parameters

tick float

Polling interval in seconds between checks for new content.

timeout Optional[float]

Maximum time to wait in seconds. If None, waits indefinitely.

Yields

AgentEvent objects (AgentStartTextEvent, AgentTextDeltaEvent, AgentTextEndEvent, AgentToolUseEvent, AgentToolResultEvent, AgentErrorEvent) as they become available. Text events are scoped to a single message: an AgentTextEndEvent is emitted at the end of each message that carried text, so adjacent assistant messages produce separate start/end pairs.

Raises

TimeoutError

If timeout elapses before the session pauses.

Return type

collections.abc.Generator[roboto.ai.agent_thread.event.AgentEvent, None, None]

Usage

Stream text output as it arrives:

for event in session.events():
    if isinstance(event, AgentTextDeltaEvent):
        print(event.text, end="", flush=True)

AgentThread.fork()

fork(message_sequence_num)#View Source

Fork this session’s history up to a specific message into a new session owned by the caller.

Fork access mirrors read access: anyone who can read the source session can fork it. A ThreadVisibility.ORG-visible session can be forked by any member of its org; a ThreadVisibility.PRIVATE session can be forked only by its creator or a Roboto admin. The new session carries the source session’s org_id so tool calls resolve against the source org’s data, and is always created with ThreadVisibility.PRIVATE visibility, owned by the caller, regardless of the source’s visibility — so forks never appear in another member’s session list.

Parameters

message_sequence_num int

Highest message sequence number (inclusive) to copy.

Returns

A new AgentThread instance for the forked session.

Raises

If the caller is not a member of the source session’s org, or the source session is ThreadVisibility.PRIVATE and the caller is neither its creator nor a Roboto admin.

If message_sequence_num is out of range or points at a message still generating.

AgentThread.from_id()

classmethod from_id(thread_id, roboto_client=None, load_messages=True)#View Source

Retrieve an existing agent thread by its unique identifier.

Loads a previously created thread from the Roboto platform, allowing users to resume conversations and access message history.

Parameters

thread_id str

Unique identifier for the thread. Accepts current ath_* identifiers as well as legacy ags_* and ch_* identifiers.

roboto_client Optional[roboto.http.RobotoClient]

HTTP client for API communication. If None, uses the default client.

load_messages bool

Whether to load the thread’s messages. If False, the thread’s messages will be empty.

Returns

AgentThread instance representing the existing thread.

Raises

If the thread does not exist.

If the caller lacks permission to access the thread.

Usage

Resume an existing thread:

thread = AgentThread.from_id("ath_abc123")
print(f"Thread has {len(thread.messages)} messages")
# Thread has 5 messages

Properties

AgentThread.goals

Goals declared across this thread’s turns, oldest first.

Each entry is an AgentThreadGoalView SDK wrapper that delegates the goal-record fields and adds three resolved properties:

Returns an empty list when the thread declared no goals or when the underlying record was fetched without loading them — call refresh() if you expect goals to be present but the list is empty.

Wrappers are cached against the underlying goals-list identity: repeated reads return the same list (and the same per-goal wrappers) until a refresh installs a fresh goals list, at which point the cache is invalidated transparently. Callers may rely on identity comparison across reads against the same snapshot.

AgentThread.invoke_skill()

invoke_skill(skill_id, version=None)#View Source

Manually invoke a skill into this thread.

Thin wrapper around send() that builds a single-element invoke_skills list and sends it with no user message — kept for SDK ergonomics on the common “invoke exactly one skill” case.

Parameters

skill_id str

The skill to invoke.

version Optional[int]

Optional version number. Must exist on the skill (any version is invokable). If omitted, the latest (MAX(version)) version is used.

Returns

Self for method chaining.

Usage

Apply a skill’s latest version to the current turn:

thread.invoke_skill("sk_qa_review")

Apply a specific version of the skill:

thread.invoke_skill("sk_qa_review", version=2)

Properties

AgentThread.latest_message

The most recent message in the conversation, or None if no messages exist.

AgentThread.messages

Complete list of messages in the conversation in chronological order.

AgentThread.refresh()

refresh()#View Source

Update the session with the latest messages and status.

Fetches any new messages or status changes from the server and updates the local session state.

Returns

Self for method chaining.

AgentThread.register_client_tool()

register_client_tool(tool)#View Source

Register a client-side tool for auto-dispatch in subsequent turns.

The tool’s spec is not sent to the backend by this call; pass it via the client_tools= argument on send(), send_text(), or submit_client_tool_results() on the next outbound request.

Parameters

The ClientTool to register.

Returns

Self for method chaining.

AgentThread.run()

run(*, on_event=None, tick=0.2, timeout=None)#View Source

Drive the session forward until it is the user’s turn.

Polls the session, auto-dispatching any pending client-side tool invocations against the callbacks registered with this session (via start(), send(), or register_client_tool()). Returns once the session status is AgentThreadStatus.USER_TURN.

If the agent requests a client-side tool that has no registered callback, an error result is submitted automatically with a descriptive message so the agent can recover, and execution continues. If a registered callback raises, the exception is caught and also submitted as an error result.

Parameters

on_event Optional[OnEvent]

Optional callback invoked for each AgentEvent as the agent generates (text deltas, tool uses, tool results, start/end markers). Use this for progress display or logging.

tick float

Polling interval in seconds between status checks.

timeout Optional[float]

Total time budget in seconds across the whole loop. If None, waits indefinitely.

Returns

Self for method chaining.

Raises

TimeoutError

If the timeout budget is exhausted before the session reaches USER_TURN.

RuntimeError

If the session is in CLIENT_TOOL_TURN with no messages (i.e. a server state that should not be reachable), or if an unexpected AgentThreadStatus value is observed.

RobotoHttpException

Propagated from the underlying submit_client_tool_results() POST if the server rejects the submission (for example, a concurrent caller already answered the tool-use).

Usage

Fire-and-forget:

session = AgentThread.start("Remember my favorite color is blue.", client_tools=[remember])
session.run()

With progress logging:

def log(event):
    if isinstance(event, AgentToolUseEvent):
        print(f"[tool-use] {event.name}({event.input})")
session.run(on_event=log)

AgentThread.send()

send(message=None, *, client_context=None, client_tools=None, analysis_scope=None, goals=None, invoke_skills=None)#View Source

Send a structured message to the session.

Parameters

AgentMessage object containing the message content and metadata. Optional when at least one entry is provided in goals or invoke_skills; in that case the server synthesizes a minimal user message for the turn.

client_context Optional[roboto.ai.core.ClientViewingContext]

Optional ClientViewingContext describing what the calling client is currently viewing when this message was composed. Informational only; see AgentThread.start() for full semantics.

Optional client-side tools to add or update for this and subsequent turns. ClientTool callbacks are registered on the session for auto-dispatch.

analysis_scope Optional[roboto.ai.core.AnalysisScope]

Optional replacement AnalysisScope. When provided, overwrites the session’s current scope for all subsequent tool invocations. When None, the session’s existing scope (if any) is left untouched.

goals Optional[collections.abc.Sequence[roboto.ai.goals.AgentGoal]]

Optional structured goals to declare for this turn. The agent runner enforces achievement of every declared goal before completing the turn.

invoke_skills Optional[collections.abc.Sequence[roboto.ai.agent_thread.record.InvokeSkillSpec]]

Optional sequence of InvokeSkillSpec to invoke one or more stored skills as part of this turn. For each entry the server fabricates a load_skill tool_use/tool_result pair after message (if any). When message and goals are both omitted, the fabricated pairs alone trigger the turn. Latest (MAX(version)) is used when version is omitted on an entry.

Returns

Self for method chaining.

Raises

If the message format is invalid.

If the caller lacks permission to send messages.

If the thread was started on another surface (see AgentThreadRecord.origin). Fork it to keep going.

AgentThread.send_text()

send_text(text, *, client_context=None, client_tools=None, analysis_scope=None, goals=None)#View Source

Send a text message to the session.

Convenience method for sending a simple text message without needing to construct an AgentMessage.

Parameters

text str

Text content to send to the assistant.

client_context Optional[roboto.ai.core.ClientViewingContext]

Optional ClientViewingContext describing what the calling client is currently viewing.

Optional client-side tools to add or update.

analysis_scope Optional[roboto.ai.core.AnalysisScope]

Optional replacement AnalysisScope; see send() for update semantics.

goals Optional[collections.abc.Sequence[roboto.ai.goals.AgentGoal]]

Optional goals to declare for this turn. See send() for full semantics.

Returns

Self for method chaining.

Raises

If the text is empty or invalid.

If the caller lacks permission to send messages.

If the thread was started on another surface (see AgentThreadRecord.origin). Fork it to keep going.

AgentThread.set_pinned()

set_pinned(pinned)#View Source

Pin or unpin this thread for the calling user.

A pin is a personal bookmark: it promotes the thread to the top of the caller’s chat sidebar and is invisible to everyone else. Anyone who may read the thread may pin it, including teammates on a thread whose visibility is ORG.

Setting the state the thread already has is accepted and changes nothing; re-pinning in particular does not move the thread back to the top of the pinned block.

Leaves pinned_at untouched on this instance’s record: only the thread listing and search endpoints populate that field.

Parameters

pinned bool

True to pin, False to unpin.

Returns

Self for method chaining.

Raises

If the thread does not exist.

If the caller is not a member of the thread’s org, or the thread is private and the caller did not create it.

Usage

Pin a thread to the top of your sidebar:

thread.set_pinned(True)

Unpin it again:

thread.set_pinned(False)

AgentThread.set_visibility()

set_visibility(visibility)#View Source

Re-scope who may read this thread.

Only the thread’s creator may call this. Roboto admins can read every thread but cannot re-scope one they did not create. Setting the visibility a thread already has is accepted and leaves it unchanged.

Parameters

ThreadVisibility.ORG to open the thread to every member of the thread’s organization, or ThreadVisibility.PRIVATE to restrict it back to its creator.

Returns

Self for method chaining.

Raises

If the thread does not exist.

If the caller is not a member of the thread’s org, or is a member but did not create the thread.

AgentThread.start()

classmethod start(message=None, *, client_context=None, system_prompt=None, model_profile=None, org_id=None, client_tools=None, analysis_scope=None, goals=None, invoke_skills=None, available_skills=None, roboto_client=None)#View Source

Start a new agent session with an initial message.

Creates a new session and sends the initial message to begin the conversation. The AI assistant will process the message and generate a response, which can be driven to completion with run() or observed event-by-event with events().

Parameters

message Optional[Union[str, roboto.ai.agent_thread.record.AgentMessage, collections.abc.Sequence[roboto.ai.agent_thread.record.AgentMessage]]]

Initial message to start the conversation. Can be a text string, a single AgentMessage, or a sequence of AgentMessage objects for multi-turn initialization. Optional when at least one entry is provided in goals or invoke_skills; in those cases the server synthesizes a minimal user message — implicitly “achieve the goals” or “apply the invoked skills” — and the agent will work the turn from there.

client_context Optional[roboto.ai.core.ClientViewingContext]

Optional ClientViewingContext describing what the calling client (e.g. the Web UI) is currently displaying when this session is started. Lets the agent resolve deictic references like “this dataset” without the user spelling them out. Informational only; does not gate authorization or scope tools — see analysis_scope for that.

system_prompt Optional[str]

Optional system prompt to customize the AI assistant’s behavior for this conversation.

model_profile Optional[str]

Optional model profile ID (e.g. “standard”, “advanced”). Defaults to the deployment’s default profile.

org_id Optional[str]

Organization ID to create the session in. If None, uses the caller’s default organization.

Optional list of client-side tools to make available to the agent. Accepts ClientTool instances (which include a callback for auto-dispatch) and bare ClientToolSpec objects (which describe the tool but require the caller to submit results manually).

analysis_scope Optional[roboto.ai.core.AnalysisScope]

Optional AnalysisScope for the session (e.g. a time window or topic-pattern filter). When provided, the scope is persisted on the session and delivered to every tool invocation on the server side. Individual tools opt in to honoring the scope as they are adopted.

goals Optional[collections.abc.Sequence[roboto.ai.goals.AgentGoal]]

Optional structured goals to declare for the first turn. When provided, message may be omitted; the server will synthesize a minimal user message and the agent runner will enforce achievement of every declared goal before completing the turn.

invoke_skills Optional[collections.abc.Sequence[roboto.ai.agent_thread.record.InvokeSkillSpec]]

Optional sequence of InvokeSkillSpec to invoke one or more stored skills at session start, in order. For each entry the server fabricates a load_skill tool_use/tool_result pair after any seeded message; with no message and no goals, the fabricated pairs alone seed the conversation. Each skill must be visible to the caller (org skill or own private skill); the version must exist on the skill but is not required to be the latest one. Latest (MAX(version)) is used when version is omitted on an entry.

available_skills Optional[collections.abc.Sequence[roboto.ai.agent_thread.record.AvailableSkillSpec]]

Optional explicit set of skills the AI may auto-invoke during this session, replacing the registry it would otherwise derive from the caller’s skill subscriptions. None (the default) keeps the subscription-derived behavior; an empty list gives the AI no auto-invokable skills; a non-empty list of AvailableSkillSpec makes exactly those skill versions auto-invokable and ignores subscriptions and per-user ai_version pins. Each entry may reference any org skill or the caller’s own private skill (visibility only — no subscription needed), at any version. This is session configuration, not a turn trigger: it does not by itself satisfy the “needs a message, goal, or invoked skill” requirement. Distinct from invoke_skills, which seeds skill bodies into the opening transcript.

roboto_client Optional[roboto.http.RobotoClient]

HTTP client for API communication. If None, uses the default client.

Returns

AgentThread instance representing the newly created session.

Raises

If the message format is invalid.

If the caller lacks permission to create sessions.

Usage

Start and drive a session with client-side tools:

from roboto.ai import client_tool
@client_tool
def recall(query: str) -> str:
    """Search long-term memory for facts matching a query."""
    ...
session = AgentThread.start("What do you remember?", client_tools=[recall])
session.run()

Properties

AgentThread.status

Current status of the thread.

AgentThread.submit_client_tool_results()

submit_client_tool_results(results, client_tools=None)#View Source

Submit results of client-side tool execution to resume the session.

On success the server has persisted every submitted tool_result and queued a new worker turn; the local record.status flips to ROBOTO_TURN to match. See the inline comment for why the next delta poll cannot communicate that transition on its own.

Parameters

results collections.abc.Sequence[roboto.ai.agent_thread.record.ClientToolResult]

Tool results from client-side execution.

Optional updated client-side tools for the next invocation. ClientTool callbacks are registered on the session for auto-dispatch.

Returns

Self for method chaining.

AgentThread.submit_feedback()

submit_feedback(message_sequence_num, sentiment, categories=None, notes=None)#View Source

Submit structured feedback on a specific assistant message in this session.

Persists categorized feedback so it can be reviewed by Roboto operators. Re-submitting from the same user on the same message overwrites sentiment, categories, and notes on the existing row rather than creating a duplicate; the feedback_id and original created/created_by are preserved.

Parameters

message_sequence_num int

Zero-indexed position of the assistant message being rated.

Overall rating direction.

Zero or more categories describing the feedback. Must match sentiment. FeedbackCategory.OTHER is always permitted but requires notes.

notes Optional[str]

Free-text notes. Required when FeedbackCategory.OTHER is among the categories.

Returns

The persisted feedback as a UserFeedbackRecord. Admin triage columns (admin_label, admin_note, resolved, resolved_by, resolved_at) are intentionally not part of this shape — they are only visible through the admin API.

Raises

If the message is still generating, the sentiment/category combination is invalid, or OTHER is used without notes.

If the caller cannot access this session.

If message_sequence_num is out of range.

Properties

AgentThread.tasks

The thread’s task list, ordered by position.

Returns an empty list when the thread has no tasks or when the underlying record was fetched without loading them — call refresh() if you expect tasks but the list is empty.

AgentThread.thread_id

thread_id str #

Unique identifier for this thread.

Return type: str

AgentThread.transcript

transcript str #

Human-readable transcript of the entire conversation.

Returns a formatted string containing all messages in the conversation, with role indicators and message content clearly separated.

Return type: str

AgentThread.unregister_client_tool()

unregister_client_tool(name)#View Source

Remove a previously registered client-tool callback.

This only removes the local callback. The backend was told about the tool in StartAgentThreadRequest client_tools (or via a later send()) and may still emit tool_use events for it; once the callback is gone, run() will submit an error result for those invocations so the agent can recover. There is no server-side deregistration API.

The tool name remains recorded as a declared client tool on this session, so the dispatcher still treats it as client-side (and not as a server tool whose result the server will post).

Parameters

name str

Name of the client tool to unregister.

Returns

bool

True if a callback was removed, False if no callback was registered under name.

Properties

AgentThread.visibility

Read scope for this thread: creator-only (PRIVATE) or open to the thread’s org (ORG).

AgentThreadDelta

class roboto.ai.agent_thread.AgentThreadDelta(/, **data)#View Source

Bases: pydantic.BaseModel

Incremental update to an agent thread.

Contains only the changes since the last synchronization, used for efficient real-time updates without transferring the entire thread history.

Parameters

data Any

Attributes

AgentThreadDelta.continuation_token

continuation_token str #

Updated token for the next incremental synchronization.

AgentThreadDelta.goals

goals list[AgentThreadGoalRecord] | None = None #

Latest snapshot of every goal declared in the thread, ordered by allocation. None means there has been no change since the previous delta — clients should retain the snapshot they already hold. An empty list means the thread has no declared goals. A non-empty list is the authoritative current snapshot and replaces any prior value.

AgentThreadDelta.messages_by_idx

messages_by_idx dict[int, AgentMessage] #

New or updated messages indexed by their position in the conversation.

AgentThreadDelta.reset_from_message_sequence_num

reset_from_message_sequence_num int | None = None #

Discard held messages from this index onward before applying this delta.

A delta’s messages normally extend what a client already holds, because the server sends only content the client has not seen. That breaks when a turn is abandoned partway and regenerated: the replacement message reuses the same index, so what the client holds at that index is text from an attempt that no longer exists, and appending to it would splice the real reply onto a discarded draft.

When this field is set, the client must truncate its held messages to indices strictly below it, then apply messages_by_idx and adopt continuation_token as usual — the server has rewound the stream to that point and will resend it. None (the common case) means append as normal.

AgentThreadDelta.status

status AgentThreadStatus | None = None #

Updated status of the agent thread.

AgentThreadDelta.tasks

tasks list[roboto.ai.core.task.AgentTask] | None = None #

Latest snapshot of the thread’s task list, ordered by position. None means no change since the previous delta — clients should retain the snapshot they already hold. An empty list means the thread has no tasks. A non-empty list is the authoritative current snapshot and replaces any prior value.

AgentThreadDelta.title

title str | None = None #

Updated title of the agent thread.

AgentThreadGoalRecord

class roboto.ai.agent_thread.AgentThreadGoalRecord(/, **data)#View Source

Bases: pydantic.BaseModel

Customer-visible read shape of a goal declared on an agent thread.

Parameters

data Any

Attributes

AgentThreadGoalRecord.achieve_tool_use_id

achieve_tool_use_id str | None = None #

tool_use_id of the achieve-tool invocation associated with this goal. Populated by the turn runner on every achieve-tool attempt, so its final value depends on the goal’s terminal status:

  • ACHIEVED: the tool_use_id of the successful invocation.
  • FAILED: the tool_use_id of the last attempted invocation, or None if the LLM never invoked the achieve-tool before the retry budget exhausted.
  • PENDING: None (or the most recent attempt so far).

Use with AgentThread.goals and the GoalResult accessor on the SDK wrapper to locate the exact AgentToolUseContent / AgentToolResultContent pair without scanning by tool name.

AgentThreadGoalRecord.concluded_at

concluded_at datetime.datetime | None = None #

Timestamp when the goal transitioned to a terminal state (ACHIEVED or FAILED). None while the goal is still PENDING.

AgentThreadGoalRecord.created

created datetime.datetime #

Timestamp when the goal was registered.

AgentThreadGoalRecord.goal_data

goal_data dict[str, Any] #

The validated goal payload as JSON. Use to_agent_goal() to recover the typed model the caller declared.

AgentThreadGoalRecord.goal_type

goal_type str #

Discriminator selecting which AgentGoal model the goal_data payload conforms to (e.g. "dataset_summary").

AgentThreadGoalRecord.message_sequence_num

message_sequence_num int #

Index in the thread’s full messages list of the AgentRole.USER message that declared this goal. Use to render goals adjacent to the turn they were attached to.

AgentThreadGoalRecord.status

Current lifecycle state of the goal.

AgentThreadGoalRecord.to_agent_goal()

to_agent_goal()#View Source

Re-hydrate goal_data into the typed AgentGoal the caller declared.

Returns

The validated, discriminated AgentGoal instance — for "dataset_summary" rows, a DatasetSummaryAgentGoal; for "dataset_triage" rows, a DatasetTriageGoal; etc.

AgentThreadGoalView

class roboto.ai.agent_thread.AgentThreadGoalView(record, messages)#View Source

SDK-side wrapper around an AgentThreadGoalRecord.

Holds a back-reference to a messages list (the parent thread’s full message history) so the achieve-tool invocation can be located via achieve_tool_use_id without forcing the caller to do the lookup by hand.

The wrapper is value-like: instantiating it does not copy the underlying record. Callers should not mutate it; mutations on the parent thread (via run / refresh) are visible through the wrapper’s resolved properties because the messages list is shared.

Properties

AgentThreadGoalView.achieve_tool_result

The matching tool-result block for achieve_tool_use, if the runner persisted one before the turn terminated.

For a FAILED goal whose last attempt errored mid-flight (or whose tool_result chunk never landed), this returns None even when achieve_tool_use is non-null.

AgentThreadGoalView.achieve_tool_use

The achieve-tool invocation the LLM submitted for this goal.

Returns None when achieve_tool_use_id is None (no attempt has been recorded) or when no matching block is present in the thread’s messages — the latter can happen if the caller is holding a thread snapshot that pre-dates the achieve-tool being persisted. In that case, a refresh of the thread should bring the block into view.

AgentThreadGoalView.achieve_tool_use_id

achieve_tool_use_id str | None #

tool_use_id of the achieve-tool invocation associated with this goal — see AgentThreadGoalRecord.achieve_tool_use_id for the per-status semantics.

Return type: str | None

AgentThreadGoalView.concluded_at

concluded_at datetime.datetime | None #

Timestamp when the goal reached a terminal state, or None while still PENDING.

Return type: datetime.datetime | None

AgentThreadGoalView.created

created datetime.datetime #

Timestamp when the goal was registered.

Return type: datetime.datetime

AgentThreadGoalView.goal_data

goal_data dict[str, Any] #

The original goal-declaration payload. Use to_agent_goal() to re-hydrate into the typed AgentGoal model.

Return type: dict[str, Any]

AgentThreadGoalView.goal_type

goal_type str #

Discriminator selecting which AgentGoal model the goal was declared as. Equivalent to self.record.goal_type.

Return type: str

AgentThreadGoalView.message_sequence_num

message_sequence_num int #

Index of the user-role message that declared this goal.

Return type: int

AgentThreadGoalView.record

The underlying wire record. Useful for callers that want the unwrapped pydantic shape (e.g. for JSON serialization).

AgentThreadGoalView.result

Typed, per-goal-type result for the achieve-tool invocation.

Returns None when no terminal achieve-tool invocation is available (PENDING goal, or FAILED with no attempted invocation), when the matching tool_use cannot be located in the thread’s messages, or when the persisted input is malformed enough to fail validation against the achieve-input model. In all three cases callers can still inspect achieve_tool_use / achieve_tool_result directly for debugging.

The returned object is one of the concrete subclasses of GoalResult (e.g. DatasetSummaryGoalResult), so callers can isinstance-dispatch or simply read the typed fields. The status field reflects the goal’s terminal status, so the same accessor works for both ACHIEVED and FAILED outcomes.

AgentThreadGoalView.status

Current lifecycle state of the goal.

AgentThreadGoalView.to_agent_goal()

to_agent_goal()#View Source

Re-hydrate the goal declaration into its typed AgentGoal model. Delegates to AgentThreadGoalRecord.to_agent_goal().

AgentThreadRecord

class roboto.ai.agent_thread.AgentThreadRecord(/, **data)#View Source

Bases: pydantic.BaseModel

Complete record of an agent thread.

Contains all the persistent data for a thread including metadata, message history, and synchronization state.

Parameters

data Any

Attributes

AgentThreadRecord.continuation_token

continuation_token str #

Token used for incremental updates and synchronization.

AgentThreadRecord.created

created datetime.datetime #

Timestamp when this agent thread was created.

AgentThreadRecord.created_by

created_by str #

User ID of the person who created this agent thread.

AgentThreadRecord.created_by_principal

created_by_principal str | None = None #

Serialized RobotoPrincipal ("ptype:id") that started this thread, e.g. "user:jo@example.com" or "invocation:iv_123". Unlike created_by, this preserves whether the thread was driven by a person, a device, or an action invocation. None on threads created before this field existed.

AgentThreadRecord.created_from_agent_id

created_from_agent_id str | None = None #

If this thread was started via the agent launch flow, the id of the agent that produced it. None for threads started directly through POST /v1/ai/threads. Forks do not inherit this field — a fork is its own thread.

AgentThreadRecord.created_from_trigger_id

created_from_trigger_id str | None = None #

If a trigger’s start_agent target launched this thread, the id of that trigger. None for every other thread, including one launched from the same agent by a person. Set alongside created_from_agent_id, which names the agent; this names what decided to run it. Forks do not inherit this field — a fork is its own thread.

AgentThreadRecord.forked_from_message_sequence_num

forked_from_message_sequence_num int | None = None #

Message sequence number in the source thread that this fork was taken from.

Populated in tandem with forked_from_thread_id; both are None for threads that were not created as a fork.

AgentThreadRecord.forked_from_thread_id

forked_from_thread_id str | None = None #

If this thread was forked, the id of the source thread. None otherwise.

Deserialization also accepts the legacy forked_from_session_id spelling for backward compatibility.

AgentThreadRecord.goals

goals list[AgentThreadGoalRecord] | None = None #

Goals declared across this thread’s turns, ordered by the turn that declared them. None means goals were not loaded for this record; an empty list means they were loaded but the thread never declared any.

AgentThreadRecord.messages

messages list[AgentMessage] = None #

Complete list of messages in the conversation.

AgentThreadRecord.model_config

model_config #

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

AgentThreadRecord.model_profile

model_profile str | None = None #

Model profile used for this agent thread (e.g., ‘standard’, ‘advanced’).

AgentThreadRecord.org_id

org_id str #

Organization ID that owns this agent thread.

AgentThreadRecord.origin

origin ThreadOrigin | None = None #

The surface that owns this thread.

Set by Roboto when the thread is created; it cannot be supplied by a caller. A thread owned by a surface outside Roboto refuses new messages over the API, raising RobotoThreadReadOnlyException; reading, cancelling, and rating it stay open, and a fork does not inherit the origin, so the copy is writable.

None on threads created before this field existed, and equivalent to ThreadOrigin.API — every surface that predates the field was Roboto’s own. It stays nullable until those rows are backfilled.

AgentThreadRecord.pinned_at

pinned_at datetime.datetime | None = None #

When the calling user pinned this thread, or None if they have not.

A pin is a personal bookmark, so two users reading the same thread see different values here. Pin and unpin through roboto.ai.agent_thread.AgentThread.set_pinned(). Only GET /v1/ai/threads and POST /v1/ai/threads/search report it; reads of a single thread leave it None whatever the pin state.

AgentThreadRecord.status

Current status of this agent thread.

AgentThreadRecord.tasks

tasks list[roboto.ai.core.task.AgentTask] | None = None #

The thread’s task list, ordered by AgentTask.position. None means tasks were not loaded for this record; an empty list means they were loaded but the thread has none.

AgentThreadRecord.thread_id

thread_id str = None #

Unique identifier for this agent thread.

Deserialization also accepts the legacy session_id and chat_id spellings for backward compatibility; the canonical attribute name is thread_id.

AgentThreadRecord.title

title str | None = None #

Title of this agent thread.

AgentThreadRecord.visibility

visibility ThreadVisibility #

Who can read this thread. PRIVATE (the default) restricts reads to the created_by user and Roboto admins; ORG opens the thread to every member of org_id.

AgentThreadStatus

class roboto.ai.agent_thread.AgentThreadStatus#View Source

Bases: roboto.compat.StrEnum

Enumeration of possible agent thread states.

Tracks the overall status of an agent thread from creation to termination.

Attributes

AgentThreadStatus.CLIENT_TOOL_TURN

CLIENT_TOOL_TURN = 'client_tool_turn' #

Client must execute pending tool uses and submit results.

AgentThreadStatus.GOALS_FAILED

GOALS_FAILED = 'goals_failed' #

The agent runner exhausted its corrective re-prompt budget without achieving every declared goal for the most-recent turn. Signals to clients that the thread needs human intervention before it can continue.

AgentThreadStatus.NOT_STARTED

NOT_STARTED = 'not_started' #

Thread has been created but no messages have been sent.

AgentThreadStatus.ROBOTO_TURN

ROBOTO_TURN = 'roboto_turn' #

Roboto is generating a message.

AgentThreadStatus.USER_TURN

USER_TURN = 'user_turn' #

User has the turn to send a message.

AgentThreadSubject

class roboto.ai.agent_thread.AgentThreadSubject(/, **data)#View Source

Bases: pydantic.BaseModel

Canonical record of an entity an AgentThread applies to.

Used to answer “which agent threads are about this dataset / file?” — the dataset detail page’s Agent Threads tab is one consumer; future surfaces that want to discover threads by entity will use the same record.

Parameters

data Any

Attributes

AgentThreadSubject.association_id

association_id str #

Identifier of the entity the thread applies to — e.g. a dataset id or file id.

AgentThreadSubject.model_config

model_config #

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

AgentThreadSubject.note

note str #

Short, free-form explanation of how the subject came to be attached (e.g. why this thread applies to that entity).

AgentToolDetailResponse

class roboto.ai.agent_thread.AgentToolDetailResponse(/, **data)#View Source

Bases: pydantic.BaseModel

Unsanitized tool request and response details for an agent tool invocation.

Parameters

data Any

Attributes

AgentToolDetailResponse.tool_result

AgentToolDetailResponse.tool_use

AgentToolResultContent

class roboto.ai.agent_thread.AgentToolResultContent(/, **data)#View Source

Bases: pydantic.BaseModel

Tool execution result content within an agent message.

Parameters

data Any

Attributes

AgentToolResultContent.content_type

content_type Literal[AgentContentType] #

AgentToolResultContent.payload

payload str | dict[str, Any] | list[Any] | None = None #

What the tool returned: free-form text, or a JSON object or array of structured data.

Independent of any model provider’s wire format (provider-agnostic, like AgentToolUseContent.input). None on results written before this field existed; use resolve_payload() to read old and new results uniformly.

AgentToolResultContent.raw_response

raw_response dict[str, Any] | None = None #

Legacy provider-formatted response envelope (Bedrock toolResult shape).

Kept so results written before payload existed remain readable; deprecated for new readers, who should call resolve_payload() instead of parsing this field.

Populated only where the server-side envelope is in scope: threads read over the API arrive with this field stripped (and, for legacy rows, with payload synthesized from it server-side before the strip), so API/SDK readers should not expect it.

AgentToolResultContent.resolve_payload()

resolve_payload()#View Source

Return what the tool returned, whichever field carries it.

Prefers payload. Results written before that field existed carry only the legacy envelope, from which the equivalent value is reconstructed via synthesize_tool_result_payload().

Returns

Optional[Union[str, dict[str, Any], list[Any]]]

The tool’s text or JSON output, or None when neither field carries a recognizable value.

Attributes

AgentToolResultContent.runtime_ms

runtime_ms int #

Wall-clock execution time of the tool in milliseconds.

AgentToolResultContent.status

status str #

Outcome of the tool execution (e.g. ‘success’, ‘error’).

AgentToolResultContent.tool_name

tool_name str #

Name of the tool that was executed.

AgentToolResultContent.tool_use_id

tool_use_id str #

Identifier of the tool invocation this result corresponds to.

AgentToolResultEvent

class roboto.ai.agent_thread.AgentToolResultEvent(/, **data)#View Source

Bases: pydantic.BaseModel

Contains the result of a tool invocation.

Parameters

data Any

Attributes

AgentToolResultEvent.name

name str #

Name of the tool that was invoked.

AgentToolResultEvent.output

output dict[str, Any] | None = None #

Raw tool output payload (from the underlying tool_result’s raw_response). May be None for errored invocations or for tools that return no data.

AgentToolResultEvent.runtime_ms

runtime_ms int | None = None #

Wall-clock execution time of the tool in milliseconds, as reported by the tool-result content. None only if the underlying content omits it.

AgentToolResultEvent.success

success bool #

Whether the tool invocation succeeded.

AgentToolResultEvent.tool_use_id

tool_use_id str #

Unique identifier for this tool invocation.

AgentToolUseContent

class roboto.ai.agent_thread.AgentToolUseContent(/, **data)#View Source

Bases: pydantic.BaseModel

Tool usage request content within an agent message.

Parameters

data Any

Attributes

AgentToolUseContent.content_type

content_type Literal[AgentContentType] #

AgentToolUseContent.input

input dict[str, Any] | None = None #

Parsed tool input parameters chosen by the LLM (provider-agnostic).

AgentToolUseContent.raw_request

raw_request dict[str, Any] | None = None #

Raw, unparsed request payload for this tool invocation.

AgentToolUseContent.tool_name

tool_name str #

Name of the tool the LLM is requesting to invoke.

AgentToolUseContent.tool_use_id

tool_use_id str #

Unique identifier for this tool invocation, used to correlate with its result.

AgentToolUseEvent

class roboto.ai.agent_thread.AgentToolUseEvent(/, **data)#View Source

Bases: pydantic.BaseModel

Signals that the AI is invoking a tool to gather information.

Parameters

data Any

Attributes

AgentToolUseEvent.input

input dict[str, Any] | None = None #

Parsed tool input parameters chosen by the LLM.

AgentToolUseEvent.name

name str #

Name of the tool being invoked.

AgentToolUseEvent.tool_use_id

tool_use_id str #

Unique identifier for this tool invocation.

AnalysisScope

class roboto.ai.agent_thread.AnalysisScope(/, **data)#View Source

Bases: pydantic.BaseModel

The slice of data an agent is expected to analyze.

An AnalysisScope is delivered to every AgentTool invocation on the server side. Individual tools opt in to honoring the scope as they are adopted; this SDK type carries the configuration, it does not itself enforce anything. Fields set to None are unconstrained on that dimension; an AnalysisScope with every field None is equivalent to no scope at all.

Parameters

data Any

Attributes

AnalysisScope.end_time

end_time int | None = None #

Upper bound (inclusive) of the analysis window, expressed as nanoseconds since the Unix epoch.

AnalysisScope.merge()

merge(override)#View Source

Zipper-merge override onto this scope, dimension by dimension.

For each field, the override value wins when it is set (not None); otherwise this scope’s value carries through. Used to reconcile an invoke-time scope (override) against an authored template scope (self): a launch can override individual dimensions while inheriting the rest.

Parameters

override AnalysisScope

Return type

Attributes

AnalysisScope.start_time

start_time int | None = None #

Lower bound (inclusive) of the analysis window, expressed as nanoseconds since the Unix epoch.

AvailableSkillSpec

class roboto.ai.agent_thread.AvailableSkillSpec(/, **data)#View Source

Bases: pydantic.BaseModel

One entry in a thread’s explicit AI-invokable skill set.

Embedded in StartAgentThreadRequest.available_skills. Unlike InvokeSkillSpec — which seeds a skill into the opening transcript as a turn trigger — this struct only declares that a skill version is available for the AI to auto-invoke via its load_skill tool during the thread. The route layer resolves access (org visibility, private gating) and version selection; this struct only carries the caller’s choice.

Defined locally to avoid the roboto.ai <-> roboto.domain.skills import cycle.

Parameters

data Any

Attributes

AvailableSkillSpec.skill_id

skill_id str #

Target skill’s ID. Must be visible to the caller — an org-shared skill, or the caller’s own private skill. A subscription is not required; the per-thread set bypasses subscription state entirely.

AvailableSkillSpec.version

version int | None = None #

Optional. If omitted, resolves to the skill’s latest (MAX(version)) row.

Pins the exact version exposed to the AI for this thread. Subscriptions and their per-user ai_version pins are ignored when a thread carries an explicit available_skills set.

ClientTool

class roboto.ai.agent_thread.ClientTool(fn, *, name, description, input_schema)#View Source

A client-side tool with an execution callback.

Wraps a Python callable as a tool that the Roboto agent can request the client to execute. The tool’s JSON schema is inferred from the callable’s type hints; the tool description and per-parameter descriptions are taken from the function’s Google-style docstring unless passed explicitly.

Most callers build ClientTools via the client_tool() decorator or ClientTool.from_function() rather than instantiating this class directly.

Usage

Using the decorator — descriptions come from the docstring:

@client_tool
def remember(fact: str, tags: Optional[list[str]] = None) -> str:
    """Store a fact in long-term memory.

    Args:
        fact: A standalone sentence worth remembering.
        tags: Optional tags for later retrieval.
    """
    ...

Using Annotated[T, Field(...)] instead (takes precedence over the docstring):

from typing import Annotated
from pydantic import Field
@client_tool
def recall(
    query: Annotated[str, Field(description="Substring to search for.")],
) -> str:
    """Search long-term memory."""
    ...

Using the factory with explicit overrides:

tool = ClientTool.from_function(
    my_fn,
    name="store_fact",
    description="Store a fact in long-term memory.",
)

Parameters

fn collections.abc.Callable[..., Any]
name str
description str
input_schema dict[str, Any]

ClientTool.from_function()

classmethod from_function(fn, *, name=None, description=None, input_schema=None)#View Source

Build a ClientTool from a Python callable.

The tool’s name defaults to fn.__name__. The tool description defaults to the summary-and-body of fn’s docstring (everything before the first Google-style section header like Args: or Returns:). Per-parameter descriptions are pulled from the docstring’s Args: section, and can be overridden with typing.Annotated[T, pydantic.Field(description="...")] or param: T = pydantic.Field(description="...").

Parameters

fn collections.abc.Callable[..., Any]

The callable to invoke when the tool is dispatched.

name Optional[str]

Override for the tool name (default: fn.__name__).

description Optional[str]

Override for the tool description (default: the docstring’s summary-and-body). Required if fn has no docstring.

input_schema Optional[dict[str, Any]]

Override for the input JSON Schema (default: inferred from fn’s type hints and docstring).

Returns

A ClientTool wrapping the given callable.

Raises

ValueError

If the description cannot be resolved, or if input_schema is not provided and the signature cannot be automatically converted (e.g. uses *args or **kwargs).

Properties

ClientTool.name

name str #

Tool name surfaced to the LLM.

Return type: str

ClientTool.spec

Declarative spec sent to the Roboto backend.

ClientToolResult

class roboto.ai.agent_thread.ClientToolResult(/, **data)#View Source

Bases: pydantic.BaseModel

Result of executing a client-side tool.

Parameters

data Any

Attributes

ClientToolResult.output

output dict[str, Any] | None = None #

Structured output returned by the tool.

ClientToolResult.runtime_ms

runtime_ms int #

Wall-clock execution time of the tool in milliseconds.

ClientToolResult.status

Outcome of the tool execution.

ClientToolResult.tool_name

tool_name str #

Name of the tool that was executed.

ClientToolResult.tool_use_id

tool_use_id str #

Identifier of the tool invocation this result corresponds to.

ClientToolResultStatus

class roboto.ai.agent_thread.ClientToolResultStatus#View Source

Bases: roboto.compat.StrEnum

Outcome of executing a client-side tool.

Attributes

ClientToolResultStatus.DECLINED

DECLINED = 'declined' #

ClientToolResultStatus.ERROR

ERROR = 'error' #

ClientToolResultStatus.SUCCESS

SUCCESS = 'success' #

ClientToolSpec

class roboto.ai.agent_thread.ClientToolSpec(/, **data)#View Source

Bases: pydantic.BaseModel

Declarative specification for a client-side tool.

Unlike AgentTool (which is an ABC with a __call__ method for server-side execution), ClientToolSpec is a plain data model. The backend includes it in the LLM’s tool list but never executes it — the client is responsible for execution and submitting the result.

Parameters

data Any

Attributes

ClientToolSpec.description

description str #

ClientToolSpec.input_schema

input_schema dict[str, Any] #

ClientToolSpec.name

name str #

FeedbackCategory

class roboto.ai.agent_thread.FeedbackCategory#View Source

Bases: roboto.compat.StrEnum

Taxonomy of feedback categories.

Which categories are valid depends on sentiment (see category_is_valid_for_sentiment()). OTHER is always permitted; when OTHER is among the selected categories, notes is required.

Attributes

FeedbackCategory.CORRECT

CORRECT = 'correct' #

FeedbackCategory.FORMATTING

FORMATTING = 'formatting' #

FeedbackCategory.GOOD_TOOL_USE

GOOD_TOOL_USE = 'good_tool_use' #

FeedbackCategory.HELPFUL

HELPFUL = 'helpful' #

FeedbackCategory.INCOMPLETE

INCOMPLETE = 'incomplete' #

FeedbackCategory.INCORRECT

INCORRECT = 'incorrect' #

FeedbackCategory.OTHER

OTHER = 'other' #

FeedbackCategory.REFUSED_VALID_REQUEST

REFUSED_VALID_REQUEST = 'refused_valid_request' #

FeedbackCategory.SLOW

SLOW = 'slow' #

FeedbackCategory.TOOL_FAILURE

TOOL_FAILURE = 'tool_failure' #

FeedbackCategory.UNSAFE

UNSAFE = 'unsafe' #

FeedbackSentiment

class roboto.ai.agent_thread.FeedbackSentiment#View Source

Bases: roboto.compat.StrEnum

Overall rating direction for a piece of AI-chat feedback.

Attributes

FeedbackSentiment.NEGATIVE

NEGATIVE = 'negative' #

FeedbackSentiment.POSITIVE

POSITIVE = 'positive' #

ForkAgentThreadRequest

class roboto.ai.agent_thread.ForkAgentThreadRequest(/, **data)#View Source

Bases: pydantic.BaseModel

Request payload for forking an agent thread at a specific message.

Parameters

data Any

Attributes

ForkAgentThreadRequest.message_sequence_num

message_sequence_num int #

Highest message sequence number (inclusive) to copy into the new thread.

InvokeSkillSpec

class roboto.ai.agent_thread.InvokeSkillSpec(/, **data)#View Source

Bases: pydantic.BaseModel

Spec for invoking a skill as part of a turn trigger.

Embedded in SendMessageRequest and StartAgentThreadRequest. The route layer resolves access (org visibility, private gating) and version selection; this struct only carries the caller’s choice.

Defined locally to avoid the roboto.ai ↔ roboto.domain.skills import cycle.

Parameters

data Any

Attributes

InvokeSkillSpec.skill_id

skill_id str #

Target skill’s ID. Must be visible to the caller (own private skill or an org-shared skill); the route layer rejects unauthorized invocations before the spec reaches the service.

InvokeSkillSpec.version

version int | None = None #

Optional. If omitted, resolves to the skill’s latest (MAX(version)) row.

Note this is the structural latest, not the caller’s pinned ai_version: a manually-invoked chip runs MAX(version) regardless of what the caller’s AI auto-invoke is pinned to. Use Skill.set_ai_version() to control the pin separately.

NEGATIVE_CATEGORIES

roboto.ai.agent_thread.NEGATIVE_CATEGORIES: frozenset[FeedbackCategory]#View Source

POSITIVE_CATEGORIES

roboto.ai.agent_thread.POSITIVE_CATEGORIES: frozenset[FeedbackCategory]#View Source

PinThreadRequest

class roboto.ai.agent_thread.PinThreadRequest(/, **data)#View Source

Bases: pydantic.BaseModel

Request body for POST /v1/ai/threads/<thread_id>/pin, which pins or unpins a thread for the caller.

Parameters

data Any

Attributes

PinThreadRequest.pinned

pinned bool #

Pin state the thread should have for the calling user after the call.

RobotoAgentGoalsFailedException

exception roboto.ai.agent_thread.RobotoAgentGoalsFailedException(thread_id)#View Source

Bases: RuntimeError

Raised by AgentThread.run() when a goal-bearing turn exhausts its corrective re-prompt budget without satisfying every declared goal.

Inherits RuntimeError rather than RobotoException because this is a strictly client-side condition — the session is paused, not errored on the wire — and the project’s Roboto*Exception hierarchy is reserved for exceptions cast from HTTP status codes by the SDK’s response layer. The typed shape still lets callers distinguish “the agent gave up on declared goals” from “I have a bug in my client state machine,” which is what motivated lifting it out of the opaque RuntimeError run() used to raise on unexpected statuses.

The session is in AgentThreadStatus.GOALS_FAILED; inspect AgentThread.messages and AgentThreadRecord.goals for detail about which goals failed and why.

Parameters

thread_id str

Attributes

RobotoAgentGoalsFailedException.thread_id

thread_id #

SendMessageRequest

class roboto.ai.agent_thread.SendMessageRequest(/, **data)#View Source

Bases: pydantic.BaseModel

Request payload for sending a message to an agent thread.

Contains the message content and optional context for the AI assistant.

Parameters

data Any

Attributes

SendMessageRequest.analysis_scope

analysis_scope roboto.ai.core.AnalysisScope | None = None #

Optional replacement analysis scope. When provided, overwrites the thread’s current analysis scope; the new scope takes effect for this turn’s tool invocations and every turn thereafter. When None, the thread’s existing analysis scope is left untouched (there is currently no wire-format way to clear a scope via send).

SendMessageRequest.client_context

client_context roboto.ai.core.ClientViewingContext | None = None #

Optional ClientViewingContext describing what the client was viewing when this message was composed. Wire field is client_context; the legacy context alias is accepted during the migration window and will be dropped in a future release.

SendMessageRequest.client_tools

client_tools list[roboto.ai.core.record.ClientToolSpec] | None = None #

Optional client-side tools available for this invocation.

SendMessageRequest.goals

goals list[roboto.ai.goals.AgentGoal] | None = None #

Goals declared for this turn. The agent runner enforces achievement: it gates a per-turn achieve-tool against each goal and re-prompts until every goal is satisfied or a per-turn retry budget is exhausted. May be omitted; when present, message becomes optional. Capped at MAX_GOALS_PER_TURN entries (see the constant for rationale).

SendMessageRequest.invoke_skills

invoke_skills list[InvokeSkillSpec] = None #

Skills to invoke as part of this turn, in order. The server fabricates one LoadSkillTool tool_use + tool_result pair per entry and appends them to the transcript after message (if any). When message is empty and no goals are declared, the fabricated pairs become the turn trigger themselves — useful for chip-only invocations that don’t carry a typed prompt. Pass a single-element list for the common “invoke one skill” case.

SendMessageRequest.message

Message content to send. May be omitted when at least one goal is declared in goals; in that case the server synthesizes a minimal user message so the LLM has a turn-initiating prompt.

StartAgentThreadRequest

class roboto.ai.agent_thread.StartAgentThreadRequest(/, **data)#View Source

Bases: pydantic.BaseModel

Request payload for starting a new agent thread.

Contains the initial messages and configuration for creating a new conversation.

Parameters

data Any

Attributes

StartAgentThreadRequest.analysis_scope

analysis_scope roboto.ai.core.AnalysisScope | None = None #

Optional analysis scope for the thread. Delivered to every tool invocation on the server side; individual tools opt in to honoring it. None means no scope.

StartAgentThreadRequest.available_skills

available_skills list[AvailableSkillSpec] | None = None #

Explicit set of skills the AI may auto-invoke during this thread, replacing the subscription-derived load_skill registry.

Tri-state:

  • None (the default) — the AI’s load_skill registry is derived per turn from the caller’s skill subscriptions, as usual.
  • [] — the AI has no auto-invokable skills for this thread.
  • a non-empty list — exactly these skill versions are auto-invokable; the caller’s subscriptions and per-user ai_version pins are ignored.

Each entry may reference any org-shared skill or the caller’s own private skill (visibility only — no subscription required), at any version. One version per skill: duplicate skill_id entries are rejected. Resolved once at thread start and frozen onto the thread; later subscription changes and skill-body edits do not propagate into it. Capped at MAX_AVAILABLE_SKILLS entries.

Distinct from invoke_skills: available_skills configures what the AI can reach for, while invoke_skills seeds skill bodies into the opening transcript as a turn trigger. It is configuration, not a trigger — a request carrying only available_skills and no messages / goals / invoke_skills is still rejected.

StartAgentThreadRequest.client_context

client_context roboto.ai.core.ClientViewingContext | None = None #

Optional ClientViewingContext describing what the client was viewing when this thread was started. Wire field is client_context; the legacy context alias is accepted during the migration window and will be dropped in a future release.

StartAgentThreadRequest.client_tools

client_tools list[roboto.ai.core.record.ClientToolSpec] | None = None #

Optional client-side tools available for this invocation.

StartAgentThreadRequest.goals

goals list[roboto.ai.goals.AgentGoal] | None = None #

Goals declared for the first turn. The agent runner enforces achievement: it gates a per-turn achieve-tool against each goal and re-prompts until every goal is satisfied or a per-turn retry budget is exhausted. May be omitted; when present, messages may be empty. Capped at MAX_GOALS_PER_TURN entries (see the constant for rationale).

StartAgentThreadRequest.invoke_skills

invoke_skills list[InvokeSkillSpec] = None #

Skills to invoke at thread start, in order. The server fabricates one LoadSkillTool tool_use + tool_result pair per entry and appends them to the transcript after any seeded messages. When messages is empty and no goals are declared, the fabricated pairs become the thread seed. Pass a single-element list for the common “invoke one skill” case.

StartAgentThreadRequest.messages

Initial messages to start the conversation with. May be empty when at least one goal is declared in goals; in that case the server synthesizes a minimal user message so the LLM has a turn-initiating prompt.

StartAgentThreadRequest.model_config

model_config #

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

StartAgentThreadRequest.model_profile

model_profile str | None = None #

Optional model profile ID for the thread (e.g. ‘standard’, ‘advanced’).

StartAgentThreadRequest.system_prompt

system_prompt str | None = None #

Optional system prompt to customize AI assistant behavior.

StartAgentThreadRequest.visibility

Who may read the resulting thread after it is created. PRIVATE (the default) restricts reads to the creator and Roboto admins; ORG lets any member of the thread’s org read it and makes the thread visible to org members on POST /v1/ai/threads/search. The default is PRIVATE so that a thread started via POST /v1/ai/threads does not leak to the rest of the org until the caller opts in; threads created through the agent launch flow default to ORG instead, since agents exist to share workflows across teammates.

SubmitFeedbackRequest

class roboto.ai.agent_thread.SubmitFeedbackRequest(/, **data)#View Source

Bases: pydantic.BaseModel

Request body for submitting feedback on an assistant message.

One row per (session, message, user); resubmitting replaces the previous sentiment/categories/notes rather than adding a new row.

Parameters

data Any

Attributes

SubmitFeedbackRequest.categories

categories list[FeedbackCategory] = None #

Categories describing what was good or bad. Semantically a set: duplicates are dropped and the persisted order is enum-value sort, not request order.

SubmitFeedbackRequest.notes

notes str | None = None #

Free-text notes. Whitespace-only input is normalised to None before persistence. Required when FeedbackCategory.OTHER is selected.

SubmitFeedbackRequest.sentiment

Overall rating direction.

SubmitToolResultsRequest

class roboto.ai.agent_thread.SubmitToolResultsRequest(/, **data)#View Source

Bases: pydantic.BaseModel

Request payload for submitting client-side tool execution results.

Parameters

data Any

Attributes

SubmitToolResultsRequest.client_tools

client_tools list[roboto.ai.core.record.ClientToolSpec] | None = None #

Optional updated client-side tools for the next invocation.

SubmitToolResultsRequest.tool_results

tool_results list[ClientToolResult] #

Tool results from client-side execution.

ThreadOrigin

class roboto.ai.agent_thread.ThreadOrigin#View Source

Bases: roboto.compat.StrEnum

The surface an AgentThreadRecord was started from, and the one that owns it.

Distinct from AgentThreadRecord.created_by_principal, which records who started a thread. This records where the conversation lives, which is what decides whether it can be added to over the API.

Import as roboto.ai.agent_thread.ThreadOrigin.

Attributes

ThreadOrigin.API

API = 'api' #

Started through Roboto itself – the web app, the CLI, the SDK, or a direct REST call.

These surfaces read the thread back from Roboto rather than mirroring it somewhere else, so there is nothing for a new turn to fall out of sync with and the thread stays writable.

ThreadOrigin.SLACK

SLACK = 'slack' #

Started by mentioning @Roboto in Slack, and mirrored into that Slack conversation.

ThreadVisibility

class roboto.ai.agent_thread.ThreadVisibility#View Source

Bases: roboto.compat.StrEnum

Read-scope for an AgentThreadRecord.

Set when the thread is created, and changed afterwards only by the thread’s creator, via roboto.ai.agent_thread.AgentThread.set_visibility() (POST /v1/ai/threads/<thread_id>/visibility). Roboto admins read every thread but cannot re-scope one they did not create.

Import as roboto.ai.agent_thread.ThreadVisibility.

Attributes

ThreadVisibility.ORG

ORG = 'org' #

Any member of the thread’s organization (and Roboto admins) may read the thread.

Default for threads produced by the agent launch flow, since agents exist to share workflows across teammates. Forks of an ORG thread do not inherit visibility — every fork lands as PRIVATE.

ThreadVisibility.PRIVATE

PRIVATE = 'private' #

Only the creating user (and Roboto admins) may read the thread.

Default for threads created via POST /v1/ai/threads so an in-flight experiment does not leak to the rest of the org until the caller opts in.

UpdateThreadVisibilityRequest

class roboto.ai.agent_thread.UpdateThreadVisibilityRequest(/, **data)#View Source

Bases: pydantic.BaseModel

Request payload for re-scoping who may read an agent thread.

Sent to POST /v1/ai/threads/<thread_id>/visibility by roboto.ai.agent_thread.AgentThread.set_visibility().

Parameters

data Any

Attributes

UpdateThreadVisibilityRequest.visibility

Read-scope the thread has once the call returns.

UserFeedbackRecord

class roboto.ai.agent_thread.UserFeedbackRecord(/, **data)#View Source

Bases: pydantic.BaseModel

Customer-facing projection of AgentFeedbackRecord.

Excludes every admin-only column (admin_label, admin_note, resolved, resolved_by, resolved_at) so routes that serve an end user never return internal triage state. Use this as the return type of any non-admin endpoint that surfaces feedback, including the response to the submitter’s own submit call.

The admin endpoints continue to return AgentFeedbackRecord directly.

Parameters

data Any

Attributes

UserFeedbackRecord.categories

categories list[FeedbackCategory] = None #

Categories describing the feedback. May be empty.

UserFeedbackRecord.created

created datetime.datetime #

When this feedback was first submitted.

UserFeedbackRecord.created_by

created_by str #

User id of the submitter.

UserFeedbackRecord.feedback_id

feedback_id str #

Unique identifier for this feedback entry.

UserFeedbackRecord.from_admin_record()

classmethod from_admin_record(record)#View Source

Project an AgentFeedbackRecord down to the user-facing shape.

Use at the boundary of any non-admin route that materialises a full admin record from persistence — the projection guarantees no admin triage column accidentally escapes to a customer response.

The projection is driven by cls.model_fields rather than a hand list of columns: a new submitter-controlled column added to both records flows through automatically, and a new admin-only column on AgentFeedbackRecord is silently dropped here (which is what we want for privacy). The matching test asserts the admin-only field set has not drifted unexpectedly.

Parameters

record AgentFeedbackRecord

Attributes

UserFeedbackRecord.message_sequence_num

message_sequence_num int #

Zero-indexed position of the assistant message within the session.

UserFeedbackRecord.modified

modified datetime.datetime #

When the submitter last updated this feedback.

UserFeedbackRecord.modified_by

modified_by str #

User id of the submitter’s last edit.

UserFeedbackRecord.notes

notes str | None = None #

Free-text notes from the submitter, if any.

UserFeedbackRecord.org_id

org_id str #

Org the session belonged to at the time of submission.

UserFeedbackRecord.sentiment

Overall rating direction.

UserFeedbackRecord.thread_id

thread_id str #

Session the feedback was submitted against.

category_is_valid_for_sentiment()

roboto.ai.agent_thread.category_is_valid_for_sentiment(category, sentiment)#View Source

Report whether category is a permitted choice under sentiment.

FeedbackCategory.OTHER is always permitted.

Parameters

Return type

bool

client_tool()

roboto.ai.agent_thread.client_tool(fn: collections.abc.Callable[..., Any], /) → ClientTool#View Source
roboto.ai.agent_thread.client_tool(*, name: str | None = None, description: str | None = None, input_schema: dict[str, Any] | None = None) → collections.abc.Callable[[collections.abc.Callable[..., Any]], ClientTool]

Decorator that converts a function into a ClientTool.

Usable bare (@client_tool) or with keyword overrides (@client_tool(description="...")). See ClientTool.from_function() for how descriptions are resolved.

Usage

Bare — infers everything from the function, including per-parameter descriptions from the docstring’s Args: section:

@client_tool
def remember(fact: str) -> str:
    """Store a fact in long-term memory.

    Args:
        fact: A standalone sentence worth remembering.
    """
    ...

With overrides:

@client_tool(name="store_fact", description="Persist a fact.")
def _store(fact: str) -> str: ...

Was this page helpful?