roboto.ai.agent_thread.agent_thread#

Module Contents#

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

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).

Examples

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:
property client_tool_names: list[str]#

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

Return type:

list[str]

events(tick=0.2, timeout=None)#

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]

Examples

Stream text output as it arrives:

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

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

Available to the session’s creator (forking their own session) and to Roboto admins (is_roboto_admin) forking anyone’s session. The new session carries the source session’s org_id so tool calls resolve against the source org’s data. The new session is owned by the caller, so admin forks of a customer session never appear in the customer’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:

  • RobotoUnauthorizedException – If the caller is not a member of the source session’s org, or is a member but is neither the source session’s creator nor a Roboto admin.

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

Return type:

AgentThread

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

Retrieve an existing agent session by its unique identifier.

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

Parameters:

  • thread_id (str) – Unique identifier for the session. Accepts both ags_* and legacy 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 session’s messages. If False, the session’s messages will be empty.

Returns:

AgentThread instance representing the existing session.

Raises:
Return type:

AgentThread

Examples

Resume an existing session:

>>> session = AgentThread.from_id("ags_abc123")
>>> print(f"Session has {len(session.messages)} messages")
Session has 5 messages
invoke_skill(skill_id, version=None)#

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.

Return type:

AgentThread

Examples

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)
property latest_message: roboto.ai.agent_thread.record.AgentMessage | None#

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

Return type:

Optional[roboto.ai.agent_thread.record.AgentMessage]

property messages: list[roboto.ai.agent_thread.record.AgentMessage]#

Complete list of messages in the conversation in chronological order.

Return type:

list[roboto.ai.agent_thread.record.AgentMessage]

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.

Return type:

AgentThread

register_client_tool(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:

tool (roboto.ai.agent_thread.client_tool.ClientTool) – The ClientTool to register.

Returns:

Self for method chaining.

Return type:

AgentThread

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

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).

Return type:

AgentThread

Examples

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)
send(message=None, *, client_context=None, client_tools=None, analysis_scope=None, goals=None, invoke_skills=None)#

Send a structured message to the session.

Parameters:

  • message (Optional[roboto.ai.agent_thread.record.AgentMessage]) – 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.

  • client_tools (Optional[collections.abc.Sequence[Union[roboto.ai.agent_thread.client_tool.ClientTool, roboto.ai.agent_thread.record.ClientToolSpec]]]) – 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:
Return type:

AgentThread

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

Send a text message to the session.

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

Parameters:
Returns:

Self for method chaining.

Raises:
Return type:

AgentThread

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)#

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.

  • client_tools (Optional[collections.abc.Sequence[Union[roboto.ai.agent_thread.client_tool.ClientTool, roboto.ai.agent_thread.record.ClientToolSpec]]]) – 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:
Return type:

AgentThread

Examples

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()
property status: roboto.ai.agent_thread.record.AgentThreadStatus#

Current status of the thread.

Return type:

roboto.ai.agent_thread.record.AgentThreadStatus

submit_client_tool_results(results, client_tools=None)#

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:
Returns:

Self for method chaining.

Return type:

AgentThread

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

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.

  • sentiment (roboto.ai.agent_thread.feedback.FeedbackSentiment) – Overall rating direction.

  • categories (Optional[list[roboto.ai.agent_thread.feedback.FeedbackCategory]]) – 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:
Return type:

roboto.ai.agent_thread.feedback.UserFeedbackRecord

property thread_id: str#

Unique identifier for this thread.

Return type:

str

property 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

unregister_client_tool(name)#

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:

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

Return type:

bool

roboto.ai.agent_thread.agent_thread.OnEvent#
exception roboto.ai.agent_thread.agent_thread.RobotoAgentGoalsFailedException(thread_id)#

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)

thread_id#
On this page