roboto.ai#

Submodules#

Package Contents#

class roboto.ai.AISummary(/, **data)#

Bases: pydantic.BaseModel

A wire-transmissible representation of an AI summary

Parameters:

data (Any)

created: datetime.datetime#

The time at which the summary was created.

status: AISummaryStatus#

The status of the summary.

summary_id: str#

The ID of the summary.

text: str#

The text of the summary.

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

An interactive AI agent session within the Roboto platform.

An AgentSession 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 AgentSession, client_tool
>>> @client_tool
... def remember(fact: str) -> str:
...     """Store a fact in long-term memory."""
...     ...
>>> session = AgentSession.start("Remember my favorite color is blue.", client_tools=[remember])
>>> session.run()

Observing events as they happen:

>>> session = AgentSession.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 AgentSessionStatus.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_session.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)
classmethod from_id(session_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:

  • session_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:

AgentSession instance representing the existing session.

Raises:
Return type:

AgentSession

Examples

Resume an existing session:

>>> session = AgentSession.from_id("ags_abc123")
>>> print(f"Session has {len(session.messages)} messages")
Session has 5 messages
property latest_message: roboto.ai.agent_session.record.AgentMessage | None#

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

Return type:

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

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

Complete list of messages in the conversation in chronological order.

Return type:

list[roboto.ai.agent_session.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:

AgentSession

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_session.client_tool.ClientTool) – The ClientTool to register.

Returns:

Self for method chaining.

Return type:

AgentSession

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 AgentSessionStatus.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 AgentSessionStatus 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:

AgentSession

Examples

Fire-and-forget:

>>> session = AgentSession.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, *, context=None, client_tools=None, analysis_scope=None)#

Send a structured message to the session.

Parameters:
Returns:

Self for method chaining.

Raises:
Return type:

AgentSession

send_text(text, *, context=None, client_tools=None, analysis_scope=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:

AgentSession

property session_id: str#

Unique identifier for this session.

Return type:

str

classmethod start(message, *, context=None, system_prompt=None, model_profile=None, org_id=None, client_tools=None, analysis_scope=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 (Union[str, roboto.ai.agent_session.record.AgentMessage, collections.abc.Sequence[roboto.ai.agent_session.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.

  • context (Optional[roboto.ai.core.RobotoLLMContext]) – Optional context to scope the AI assistant’s knowledge for this conversation (e.g., specific datasets or resources).

  • 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_session.client_tool.ClientTool, roboto.ai.agent_session.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.

  • roboto_client (Optional[roboto.http.RobotoClient]) – HTTP client for API communication. If None, uses the default client.

Returns:

AgentSession instance representing the newly created session.

Raises:
Return type:

AgentSession

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 = AgentSession.start("What do you remember?", client_tools=[recall])
>>> session.run()
property status: roboto.ai.agent_session.record.AgentSessionStatus#

Current status of the session.

Return type:

roboto.ai.agent_session.record.AgentSessionStatus

submit_client_tool_results(results, client_tools=None)#

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

Parameters:
Returns:

Self for method chaining.

Return type:

AgentSession

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 StartAgentSessionRequest 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

class roboto.ai.AgentSessionRecord(/, **data)#

Bases: pydantic.BaseModel

Complete record of an agent session.

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

Parameters:

data (Any)

property chat_id: str#

Backwards-compatible alias — serialized as chat_id in API responses.

Return type:

str

continuation_token: str#

Token used for incremental updates and synchronization.

created: datetime.datetime#

Timestamp when this agent session was created.

created_by: str#

User ID of the person who created this agent session.

messages: list[AgentMessage] = None#

Complete list of messages in the conversation.

model_profile: str | None = None#

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

org_id: str#

Organization ID that owns this agent session.

session_id: str = None#

Unique identifier for this agent session.

status: AgentSessionStatus#

Current status of this agent session.

title: str | None = None#

Title of this agent session.

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

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.

Examples

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[Ellipsis, Any])

  • name (str)

  • description (str)

  • input_schema (dict[str, Any])

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

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[Ellipsis, 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).

Return type:

ClientTool

property name: str#

Tool name surfaced to the LLM.

Return type:

str

property spec: roboto.ai.agent_session.record.ClientToolSpec#

Declarative spec sent to the Roboto backend.

Return type:

roboto.ai.agent_session.record.ClientToolSpec

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

Bases: pydantic.BaseModel

Result of executing a client-side tool.

Parameters:

data (Any)

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

Structured output returned by the tool.

runtime_ms: int#

Wall-clock execution time of the tool in milliseconds.

status: ClientToolResultStatus#

Outcome of the tool execution.

tool_name: str#

Name of the tool that was executed.

tool_use_id: str#

Identifier of the tool invocation this result corresponds to.

class roboto.ai.ClientToolResultStatus#

Bases: roboto.compat.StrEnum

Outcome of executing a client-side tool.

DECLINED = 'declined'#
ERROR = 'error'#
SUCCESS = 'success'#
class roboto.ai.ClientToolSpec(/, **data)#

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)

description: str#
input_schema: dict[str, Any]#
name: str#
class roboto.ai.PromptRequest(/, **data)#

Bases: pydantic.BaseModel

A generic request intended for a natural-language powered endpoint which accepts a human-readable prompt.

Parameters:

data (Any)

prompt: str#

The prompt to send to the AI model.

class roboto.ai.SetSummaryRequest(/, **data)#

Bases: pydantic.BaseModel

A request to set the summary of an entity.

Parameters:

data (Any)

summary: str#

The summary to set.

On this page