roboto.ai
Submodules
Package Contents
AISummary
Bases: pydantic.BaseModel
A wire-transmissible representation of an AI summary
Parameters
data AnyAttributes
AgentThread
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)Parameters
roboto_client Optional[roboto.client_tools Optional[collections.Properties
AgentThread.client_tool_names
Names of client-side tools registered on this session with callbacks.
AgentThread.events()
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 floatPolling 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
TimeoutErrorIf timeout elapses before the session pauses.
Return type
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 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 intHighest 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()
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 strUnique identifier for the thread. Accepts current ath_* identifiers as well as legacy ags_* and ch_* identifiers.
roboto_client Optional[roboto.HTTP client for API communication. If None, uses the default client.
load_messages boolWhether 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 messagesProperties
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:
AgentThreadGoalView.achieve_tool_use— the LLM’s achieve-tool invocation, located byachieve_tool_use_id.AgentThreadGoalView.achieve_tool_result— the matching tool result block, if one was persisted.AgentThreadGoalView.result— the typed, per-goal-typeGoalResult(with the LLM’s submitted payload parsed into typed fields).
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()
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 strThe 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()
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 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()
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 floatPolling 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
TimeoutErrorIf the timeout budget is exhausted before the session reaches USER_TURN.
RuntimeErrorIf 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.
RobotoHttpExceptionPropagated 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 a structured message to the session.
Parameters
message Optional[roboto.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.Optional ClientViewingContext describing what the calling client is currently viewing when this message was composed. Informational only; see AgentThread.start() for full semantics.
client_tools Optional[collections.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.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.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.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 a text message to the session.
Convenience method for sending a simple text message without needing to construct an AgentMessage.
Parameters
text strText content to send to the assistant.
client_context Optional[roboto.Optional ClientViewingContext describing what the calling client is currently viewing.
client_tools Optional[collections.Optional client-side tools to add or update.
analysis_scope Optional[roboto.Optional replacement AnalysisScope; see send() for update semantics.
goals Optional[collections.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()
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 boolTrue 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()
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()
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.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.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.
client_tools Optional[collections.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.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.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.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.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 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 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.Tool results from client-side execution.
client_tools Optional[collections.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 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 intZero-indexed position of the assistant message being rated.
Overall rating direction.
categories Optional[list[roboto.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.transcript
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.
AgentThread.unregister_client_tool()
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 strName of the client tool to unregister.
Returns
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).
AgentThreadRecord
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 AnyAttributes
AgentThreadRecord.continuation_token
Token used for incremental updates and synchronization.
AgentThreadRecord.created_by_principal
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
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
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
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
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 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
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 used for this agent thread (e.g., ‘standard’, ‘advanced’).
AgentThreadRecord.origin
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
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.tasks
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
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.visibility
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.
ClientTool
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.name strdescription strinput_schema dict[str, Any]ClientTool.from_function()
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.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
ValueErrorIf 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.spec
Declarative spec sent to the Roboto backend.
ClientToolResult
Bases: pydantic.BaseModel
Result of executing a client-side tool.
Parameters
data AnyAttributes
ClientToolResult.tool_use_id
Identifier of the tool invocation this result corresponds to.
ClientToolResultStatus
ClientToolSpec
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 AnyPromptRequest
Bases: pydantic.BaseModel
A generic request intended for a natural-language powered endpoint which accepts a human-readable prompt.
Parameters
data AnyAttributes
SetSummaryRequest
Bases: pydantic.BaseModel
A request to set the summary of an entity.
Parameters
data AnyAttributes
client_tool()
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: ...