roboto.ai.agent_session#

Submodules#

Package Contents#

class roboto.ai.agent_session.AdminUpdateFeedbackRequest(/, **data)#

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)

admin_label: str | None | roboto.sentinels.NotSetType#
admin_note: str | None | roboto.sentinels.NotSetType#
model_config#

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

resolved: bool | roboto.sentinels.NotSetType#
type roboto.ai.agent_session.AgentContent = AgentTextContent | AgentToolUseContent | AgentToolResultContent | AgentErrorContent#

Type alias for all possible content types within agent messages.

class roboto.ai.agent_session.AgentContentType#

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.

ERROR = 'error'#

Error information when message generation fails.

TEXT = 'text'#

Plain text content from users or AI responses.

TOOL_RESULT = 'tool_result'#

Results returned from tool executions.

TOOL_USE = 'tool_use'#

Tool invocation requests from the AI assistant.

class roboto.ai.agent_session.AgentErrorContent(/, **data)#

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)

content_type: Literal[AgentContentType]#
error_code: str | None = None#

Optional error code for programmatic handling.

error_message: str#

User-friendly error message describing what went wrong.

class roboto.ai.agent_session.AgentErrorEvent(/, **data)#

Bases: pydantic.BaseModel

Signals that message generation failed or was cancelled.

Parameters:

data (Any)

error_code: str | None = None#

Optional error code for programmatic handling.

error_message: str#

User-friendly error message describing what went wrong.

type roboto.ai.agent_session.AgentEvent = Union[AgentStartTextEvent, AgentTextDeltaEvent, AgentTextEndEvent, AgentToolUseEvent, AgentToolResultEvent, AgentErrorEvent]#
class roboto.ai.agent_session.AgentGoalStatus#

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

ACHIEVED = 'achieved'#

Goal’s corresponding achieve-tool was invoked successfully.

FAILED = 'failed'#

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

PENDING = 'pending'#

Goal has been registered but not yet completed.

class roboto.ai.agent_session.AgentMessage(/, **data)#

Bases: pydantic.BaseModel

A single message within an agent session.

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)

content: list[AgentContent]#

List of content blocks that make up this message.

created: datetime.datetime = None#

Timestamp when this message was created.

is_complete()#

Check if message generation is complete.

Returns:

True if the message status is COMPLETED, False otherwise.

Return type:

bool

is_unsuccessful()#

Check if message generation failed or was cancelled.

Returns:

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

Return type:

bool

role: AgentRole#

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

status: AgentMessageStatus#

Current generation status of this message.

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

Create a simple text message.

Convenience method for creating a message containing only text content.

Parameters:

  • text (str) – The text content for the message.

  • role (AgentRole) – The role of the message sender. Defaults to USER.

Returns:

AgentMessage instance containing the text content.

Return type:

AgentMessage

class roboto.ai.agent_session.AgentMessageStatus#

Bases: roboto.compat.StrEnum

Enumeration of possible message generation states.

Tracks the lifecycle of message generation from initiation to completion.

CANCELLED = 'cancelled'#

Message generation was cancelled by the user.

COMPLETED = 'completed'#

Message generation has finished and content is complete.

FAILED = 'failed'#

Message generation failed due to an error.

GENERATING = 'generating'#

Message content is currently being generated.

NOT_STARTED = 'not_started'#

Message has been queued but generation has not begun.

is_terminal()#

Check if the message generation is in a terminal state.

Returns:

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

Return type:

bool

class roboto.ai.agent_session.AgentRole#

Bases: roboto.compat.StrEnum

Enumeration of possible roles in an agent session.

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

ASSISTANT = 'assistant'#

AI agent responding to user queries and requests.

ROBOTO = 'roboto'#

Roboto system providing tool results and system information.

USER = 'user'#

Human user sending messages to the agent.

class roboto.ai.agent_session.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)
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 AgentSession 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:

AgentSession

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

Send a structured message to the session.

Parameters:

  • message (Optional[roboto.ai.agent_session.record.AgentMessage]) – AgentMessage object containing the message content and metadata. Optional when at least one entry is provided in goals; 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 AgentSession.start() for full semantics.

  • client_tools (Optional[collections.abc.Sequence[Union[roboto.ai.agent_session.client_tool.ClientTool, roboto.ai.agent_session.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.

Returns:

Self for method chaining.

Raises:
Return type:

AgentSession

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:

AgentSession

property session_id: str#

Unique identifier for this session.

Return type:

str

classmethod start(message=None, *, client_context=None, system_prompt=None, model_profile=None, org_id=None, client_tools=None, analysis_scope=None, goals=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_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. Optional when at least one entry is provided in goals; in that case the server synthesizes a minimal user message — implicitly “achieve the goals” — and the agent will work to achieve every declared goal during the first turn.

  • 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_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.

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

  • 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

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_session.feedback.FeedbackSentiment) – Overall rating direction.

  • categories (Optional[list[roboto.ai.agent_session.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_session.feedback.UserFeedbackRecord

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.agent_session.AgentSessionDelta(/, **data)#

Bases: pydantic.BaseModel

Incremental update to an agent session.

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

Parameters:

data (Any)

continuation_token: str#

Updated token for the next incremental synchronization.

goals: list[AgentSessionGoalRecord] | None = None#

Latest snapshot of every goal declared in the session, 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 session has no declared goals. A non-empty list is the authoritative current snapshot and replaces any prior value.

messages_by_idx: dict[int, AgentMessage]#

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

status: AgentSessionStatus | None = None#

Updated status of the agent session.

title: str | None = None#

Updated title of the agent session.

class roboto.ai.agent_session.AgentSessionGoalRecord(/, **data)#

Bases: pydantic.BaseModel

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

Parameters:

data (Any)

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.

created: datetime.datetime#

Timestamp when the goal was registered.

goal_data: dict[str, Any]#

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

goal_type: str#

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

message_sequence_num: int#

Index in the session’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.

status: AgentGoalStatus#

Current lifecycle state of the goal.

to_agent_goal()#

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.

Return type:

roboto.ai.goals.AgentGoal

class roboto.ai.agent_session.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.

forked_from_message_sequence_num: int | None = None#

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

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

forked_from_session_id: str | None = None#

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

goals: list[AgentSessionGoalRecord] | None = None#

Goals declared across this session’s turns, ordered by allocation (i.e. by goal_index). None means goals were not loaded for this record (full-session reads populate the field; lightweight or legacy reads may omit it). An empty list means goals were loaded but the session never declared any.

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.agent_session.AgentSessionStatus#

Bases: roboto.compat.StrEnum

Enumeration of possible agent session states.

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

CLIENT_TOOL_TURN = 'client_tool_turn'#

Client must execute pending tool uses and submit results.

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 session needs human intervention before it can continue.

NOT_STARTED = 'not_started'#

Session has been created but no messages have been sent.

ROBOTO_TURN = 'roboto_turn'#

Roboto is generating a message.

USER_TURN = 'user_turn'#

User has the turn to send a message.

class roboto.ai.agent_session.AgentStartTextEvent(/, **data)#

Bases: pydantic.BaseModel

Signals the beginning of text generation in a chat response.

Parameters:

data (Any)

class roboto.ai.agent_session.AgentTextContent(/, **data)#

Bases: pydantic.BaseModel

Text content within an agent message.

Parameters:

data (Any)

text: str#

The actual text content of the message.

class roboto.ai.agent_session.AgentTextDeltaEvent(/, **data)#

Bases: pydantic.BaseModel

Contains incremental text content as the AI generates its response.

Parameters:

data (Any)

text: str#

Text fragment from the streaming response.

class roboto.ai.agent_session.AgentTextEndEvent(/, **data)#

Bases: pydantic.BaseModel

Signals the completion of text generation in a chat response.

Parameters:

data (Any)

class roboto.ai.agent_session.AgentToolDetailResponse(/, **data)#

Bases: pydantic.BaseModel

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

Parameters:

data (Any)

tool_result: roboto.ai.core.record.AgentToolResultContent#
tool_use: roboto.ai.core.record.AgentToolUseContent#
class roboto.ai.agent_session.AgentToolResultContent(/, **data)#

Bases: pydantic.BaseModel

Tool execution result content within an agent message.

Parameters:

data (Any)

content_type: Literal[AgentContentType]#
raw_response: dict[str, Any] | None = None#

Raw, unparsed response payload from tool execution.

runtime_ms: int#

Wall-clock execution time of the tool in milliseconds.

status: str#

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

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.agent_session.AgentToolResultEvent(/, **data)#

Bases: pydantic.BaseModel

Contains the result of a tool invocation.

Parameters:

data (Any)

name: str#

Name of the tool that was invoked.

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.

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.

success: bool#

Whether the tool invocation succeeded.

tool_use_id: str#

Unique identifier for this tool invocation.

class roboto.ai.agent_session.AgentToolUseContent(/, **data)#

Bases: pydantic.BaseModel

Tool usage request content within an agent message.

Parameters:

data (Any)

content_type: Literal[AgentContentType]#
input: dict[str, Any] | None = None#

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

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

Raw, unparsed request payload for this tool invocation.

tool_name: str#

Name of the tool the LLM is requesting to invoke.

tool_use_id: str#

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

class roboto.ai.agent_session.AgentToolUseEvent(/, **data)#

Bases: pydantic.BaseModel

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

Parameters:

data (Any)

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

Parsed tool input parameters chosen by the LLM.

name: str#

Name of the tool being invoked.

tool_use_id: str#

Unique identifier for this tool invocation.

class roboto.ai.agent_session.AnalysisScope(/, **data)#

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)

end_time: int | None = None#

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

start_time: int | None = None#

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

class roboto.ai.agent_session.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.agent_session.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.agent_session.ClientToolResultStatus#

Bases: roboto.compat.StrEnum

Outcome of executing a client-side tool.

DECLINED = 'declined'#
ERROR = 'error'#
SUCCESS = 'success'#
class roboto.ai.agent_session.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.agent_session.FeedbackCategory#

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.

CORRECT = 'correct'#
FORMATTING = 'formatting'#
GOOD_TOOL_USE = 'good_tool_use'#
HELPFUL = 'helpful'#
INCOMPLETE = 'incomplete'#
INCORRECT = 'incorrect'#
OTHER = 'other'#
REFUSED_VALID_REQUEST = 'refused_valid_request'#
SLOW = 'slow'#
TOOL_FAILURE = 'tool_failure'#
UNSAFE = 'unsafe'#
class roboto.ai.agent_session.FeedbackSentiment#

Bases: roboto.compat.StrEnum

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

NEGATIVE = 'negative'#
POSITIVE = 'positive'#
class roboto.ai.agent_session.ForkChatRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload for forking a chat at a specific message.

Parameters:

data (Any)

message_sequence_num: int#

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

roboto.ai.agent_session.NEGATIVE_CATEGORIES: frozenset[FeedbackCategory]#
roboto.ai.agent_session.POSITIVE_CATEGORIES: frozenset[FeedbackCategory]#
exception roboto.ai.agent_session.RobotoAgentGoalsFailedException(session_id)#

Bases: RuntimeError

Raised by AgentSession.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 AgentSessionStatus.GOALS_FAILED; inspect AgentSession.messages and AgentSessionRecord.goals for detail about which goals failed and why.

Parameters:

session_id (str)

session_id#
class roboto.ai.agent_session.SendMessageRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload for sending a message to an agent session.

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

Parameters:

data (Any)

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

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

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.

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

Optional client-side tools available for this invocation.

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

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

Type:

Goals declared for this turn. The agent runner enforces achievement

message: roboto.ai.core.record.AgentMessage | None = None#

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.

class roboto.ai.agent_session.StartAgentSessionRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload for starting a new agent session.

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

Parameters:

data (Any)

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

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

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

Optional ClientViewingContext describing what the client was viewing when this session 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.

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

Optional client-side tools available for this invocation.

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

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

Type:

Goals declared for the first turn. The agent runner enforces achievement

messages: list[roboto.ai.core.record.AgentMessage] = None#

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.

model_profile: str | None = None#

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

system_prompt: str | None = None#

Optional system prompt to customize AI assistant behavior.

class roboto.ai.agent_session.SubmitFeedbackRequest(/, **data)#

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)

categories: list[FeedbackCategory] = None#

duplicates are dropped and the persisted order is enum-value sort, not request order.

Type:

Categories describing what was good or bad. Semantically a set

notes: str | None = None#

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

sentiment: FeedbackSentiment#

Overall rating direction.

class roboto.ai.agent_session.SubmitToolResultsRequest(/, **data)#

Bases: pydantic.BaseModel

Request payload for submitting client-side tool execution results.

Parameters:

data (Any)

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

Optional updated client-side tools for the next invocation.

tool_results: list[ClientToolResult]#

Tool results from client-side execution.

class roboto.ai.agent_session.UserFeedbackRecord(/, **data)#

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)

categories: list[FeedbackCategory] = None#

Categories describing the feedback. May be empty.

created: datetime.datetime#

When this feedback was first submitted.

created_by: str#

User id of the submitter.

feedback_id: str#

Unique identifier for this feedback entry.

classmethod from_admin_record(record)#

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)

Return type:

UserFeedbackRecord

message_sequence_num: int#

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

modified: datetime.datetime#

When the submitter last updated this feedback.

modified_by: str#

User id of the submitter’s last edit.

notes: str | None = None#

Free-text notes from the submitter, if any.

org_id: str#

Org the session belonged to at the time of submission.

sentiment: FeedbackSentiment#

Overall rating direction.

session_id: str#

Session the feedback was submitted against.

roboto.ai.agent_session.category_is_valid_for_sentiment(category, sentiment)#

Report whether category is a permitted choice under sentiment.

FeedbackCategory.OTHER is always permitted.

Parameters:
Return type:

bool

roboto.ai.agent_session.client_tool(fn: collections.abc.Callable[Ellipsis, Any], /) → ClientTool#
roboto.ai.agent_session.client_tool(*, name: str | None = None, description: str | None = None, input_schema: dict[str, Any] | None = None) → collections.abc.Callable[[collections.abc.Callable[Ellipsis, 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.

Note

This function is declared with two typing.overload() stubs above so that type checkers see the decorated name as a ClientTool regardless of call form. The overloads carry no runtime behavior; the implementation below handles both shapes.

Examples

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: ...
On this page