roboto.ai.agent#

Reusable, parameterized agent definitions.

An agent captures a parameterized StartAgentThreadRequest alongside the variables it declares, so the same workflow (triage, summary, etc.) can be re-invoked against new subjects without re-authoring the request. Variables appear as {{name}} placeholders in any string leaf of the request body and are substituted client-side before the request is sent — unresolved or unknown placeholders raise rather than reach the service.

Submodules#

Package Contents#

class roboto.ai.agent.Agent(record, roboto_client=None)#

A reusable, parameterized factory for AgentThread.

An Agent captures a StartAgentThreadRequest body with {{name}} placeholders alongside the TemplateVariable declarations that those placeholders bind to. Invoking the agent substitutes caller-supplied values into the body, runs invoke-time existence checks for typed values (dataset ids, device ids), and starts a session via the same code path as AgentThread.start().

The wrapper is to AgentRecord what Action is to ActionRecord: the record carries the wire-shape data, this class carries the behaviors.

Note

Agents cannot be instantiated directly through the constructor when starting from a name or an id; use create(), from_id(), or from_name() to obtain Agent instances.

Parameters:
property agent_id: str#

Platform-issued agt_<short> handle. Stable across renames.

Return type:

str

classmethod create(name, request_template, variables=None, description=None, caller_org_id=None, roboto_client=None)#

Persist a new agent in the caller’s organization.

The platform fills agent_id, org_id, created, created_by, modified, and modified_by from the caller’s identity. Every other field comes from this method’s arguments.

Parameters:

  • name (str) – Human-readable display name. Unique per (org_id, name); a second agent with the same name in the same org is rejected with RobotoConflictException.

  • request_template (roboto.ai.agent_thread.StartAgentThreadRequest) – The StartAgentThreadRequest to clone at invoke time. Any string leaf may carry {{name}} placeholders; every referenced placeholder must appear in variables (and vice versa).

  • variables (Optional[collections.abc.Sequence[roboto.ai.agent.record.TemplateVariable]]) – TemplateVariable declarations. The set of names here must equal the set of placeholder names referenced anywhere in request_template; a mismatch fails server-side validation with a 400.

  • description (Optional[str]) – Free-form text rendered on the agent’s detail and library pages.

  • caller_org_id (Optional[str]) – Organization to create the agent in. If omitted and the caller belongs to exactly one organization, that organization is used; required when the caller belongs to multiple organizations.

  • roboto_client (Optional[roboto.http.RobotoClient]) – Optional RobotoClient instance. If omitted, the default client configuration is used.

Returns:

The newly created Agent instance.

Raises:
Return type:

Agent

Examples

Create an agent with one dataset-typed variable:

>>> from roboto.ai.agent import Agent, TemplateVariable, TemplateVariableType
>>> from roboto.ai.agent_thread import StartAgentThreadRequest, AgentMessage
>>> request_template = StartAgentThreadRequest(
...     messages=[AgentMessage.text("Triage dataset {{dataset_id}}.")],
... )
>>> agent = Agent.create(
...     name="triage",
...     request_template=request_template,
...     variables=[
...         TemplateVariable(name="dataset_id", type=TemplateVariableType.DATASET_ID),
...     ],
...     description="Apply the triage label vocabulary to a dataset.",
... )
property created: datetime.datetime#

Wall-clock time the agent was first persisted.

Return type:

datetime.datetime

property created_by: str#

User id of the original author.

Return type:

str

delete()#

Permanently delete this agent.

Sessions previously launched from this agent are unaffected — their created_from_agent_id references this agent’s id independent of any foreign key, so they remain readable. Creating a new agent with the same name in the same org is allowed after deletion.

Raises:
Return type:

None

Examples

>>> agent = Agent.from_name("retired_triage")
>>> agent.delete()
property description: str | None#

Free-form text rendered on the agent’s detail and library pages.

Return type:

Optional[str]

classmethod for_org(org_id=None, roboto_client=None)#

Iterate every Agent declared in an organization.

Yields agents newest-first (by created) and paginates transparently; callers consume the generator without worrying about next_token.

Parameters:

  • org_id (Optional[str]) – Organization to list agents for. If omitted and the caller belongs to exactly one organization, that organization is used.

  • roboto_client (Optional[roboto.http.RobotoClient]) – Optional RobotoClient; defaults to the ambient configuration.

Yields:

Agent instances, one per row, in newest-first order.

Return type:

collections.abc.Generator[Agent, None, None]

Examples

Print every agent’s name:

>>> for agent in Agent.for_org():
...     print(agent.name)

Count agents in a specific org:

>>> count = sum(1 for _ in Agent.for_org(org_id="og_abc123"))
classmethod from_id(agent_id, roboto_client=None)#

Load an Agent by its canonical id.

agent_id is platform-unique, so the SDK does not need the caller to declare which org the agent lives in — the route looks up the record and authorizes the caller against the record’s org_id.

Parameters:

  • agent_id (str) – The platform-issued agt_<short> identifier.

  • roboto_client (Optional[roboto.http.RobotoClient]) – Optional RobotoClient; defaults to the ambient configuration.

Returns:

The Agent instance.

Raises:

RobotoNotFoundException – The agent does not exist or belongs to an organization the caller cannot read.

Return type:

Agent

Examples

>>> agent = Agent.from_id("agt_abc123")
classmethod from_name(name, roboto_client=None, owner_org_id=None)#

Load an Agent by its (org_id, name) handle.

Mirrors from_name(): name + owning org are jointly unique, so this returns at most one agent. Unlike from_id() (which derives the org from the looked-up record), name lookup needs the org up-front and goes through the X-Roboto-Resource-Owner-Id header. agent_id (the canonical handle) stays stable across renames, so a script that pins by name follows renames; a script that pins by id does not.

Parameters:

  • name (str) – The agent’s name.

  • roboto_client (Optional[roboto.http.RobotoClient]) – Optional RobotoClient; defaults to the ambient configuration.

  • owner_org_id (Optional[str]) – Organization that owns the agent. If omitted and the caller belongs to one organization, that organization is used.

Returns:

The Agent instance.

Raises:

RobotoNotFoundException – No agent with that name exists in the resolved organization.

Return type:

Agent

Examples

>>> agent = Agent.from_name("triage")
launch(values=None, visibility=ThreadVisibility.ORG)#

Resolve placeholders and start an AgentThread.

The platform substitutes values into request_template, runs launch-time existence checks on typed values (DATASET_ID / DEVICE_ID), and creates a thread via the same code path as a bare AgentThread.start(). The resulting thread is owned by this agent’s org_id (not the caller’s default org), and its created_from_agent_id is stamped with this agent’s agent_id so the agent detail page can list “threads launched from here”.

Parameters:

  • values (Optional[dict[str, str]]) – Variable name to caller-supplied value. Keys must match the names declared in variables; values are coerced to str before splicing into placeholders. Omit when every declared variable carries a default.

  • visibility (roboto.ai.agent_thread.ThreadVisibility) – ThreadVisibility for the new thread. Defaults to ORG because agents exist to share workflows; the caller passes PRIVATE to opt out per launch. Overrides whatever the authored request_template carries.

Returns:

An AgentThread wrapping the freshly created thread. The thread is fully hydrated (messages, status, continuation token) and ready for AgentThread.run() or AgentThread.events().

Raises:
Return type:

roboto.ai.agent_thread.AgentThread

Examples

>>> agent = Agent.from_name("triage")
>>> thread = agent.launch(values={"dataset_id": "ds_abc123"})
>>> thread.run()
property modified: datetime.datetime#

Wall-clock time of the most recent successful update.

Return type:

datetime.datetime

property modified_by: str#

User id who applied the most recent update; equals created_by for records never edited since creation.

Return type:

str

property name: str#

Human-readable display name; unique within org_id.

Return type:

str

property org_id: str#

Organization that owns this agent.

Return type:

str

property record: roboto.ai.agent.record.AgentRecord#

Underlying AgentRecord.

Wire shape used in API requests; may evolve over time. Prefer the public Agent API unless you need direct record access.

Return type:

roboto.ai.agent.record.AgentRecord

property request_template: roboto.ai.agent_thread.StartAgentThreadRequest#

The StartAgentThreadRequest cloned at invoke time.

Return type:

roboto.ai.agent_thread.StartAgentThreadRequest

to_dict()#

Return the JSON-serializable form of the underlying record.

Return type:

dict[str, Any]

update(name=NotSet, description=NotSet, request_template=NotSet, variables=NotSet)#

Apply a partial update.

Each parameter defaults to NotSet so omitting a field leaves it untouched on the server, while passing None for description clears it. request_template and variables are co-validated server-side after the merge: the post-update placeholder set must equal the post-update variable set, even if only one of the two fields was supplied.

Parameters:
Returns:

self, with record refreshed to the persisted state.

Raises:
Return type:

Agent

Examples

>>> agent.update(description="Triages a dataset against the v2 label vocab.")

Clear the description:

>>> agent.update(description=None)
property variables: list[roboto.ai.agent.record.TemplateVariable]#

TemplateVariable declarations. The set of names here equals the set of placeholders referenced in request_template.

Return type:

list[roboto.ai.agent.record.TemplateVariable]

class roboto.ai.agent.AgentRecord(/, **data)#

Bases: pydantic.BaseModel

A pre-filled StartAgentThreadRequest plus declared variables.

An AgentRecord captures most of what an AgentThread needs to start - system prompt, model profile, seeded messages, declared goals - and leaves a small set of named holes that callers must fill at invoke time. The canonical example is a triage agent whose goal carries a fixed label_vocabulary but a {{dataset.id}} placeholder, so the same agent can be re-run against many datasets without re-declaring the vocabulary.

The record’s main invariant - enforced by _validate_variables_match_placeholders() - is that the set of {{...}} placeholders found anywhere in request_template equals exactly {v.name for v in variables}. A mismatch fails at save time so stale invoke forms can’t paper over typos.

Parameters:

data (Any)

agent_id: str#

Canonical handle. Generated server-side (agt_<short>); stable across renames.

created: datetime.datetime#

Wall-clock time the record was first persisted.

created_by: str#

User id of the agent’s original author.

description: str | None = None#

Free-form description rendered on the agents list and detail pages.

modified: datetime.datetime#

Wall-clock time of the most recent successful update.

modified_by: str#

User id who applied the most recent update; equals created_by on records that have not been edited since creation.

name: str#

Human-readable display name. Mutable. Unique per (org_id, name) - enforced at the persistence layer, not on the record.

org_id: str#

Organization that owns the agent. All CRUD and invoke calls must come from a member of this org.

request_template: roboto.ai.agent_thread.record.StartAgentThreadRequest#

The body that will be cloned and resolved into a real StartAgentThreadRequest at invoke time. Any string leaf may contain {{name}} placeholders; non-string fields cannot be templated in v1 (the substitution engine never visits them).

variables: list[TemplateVariable] = None#

Declared variables. The set of names here must equal the set of placeholder names parsed from request_template.

exception roboto.ai.agent.AgentResolutionError#

Bases: ValueError

Base class for resolver-raised errors. Routes map subclasses to 4xx.

class roboto.ai.agent.CreateAgentRequest(/, **data)#

Bases: pydantic.BaseModel

Wire payload for POST /v1/ai/agents.

The server fills AgentRecord.agent_id, org_id, created_by, created, modified, and modified_by from the caller’s identity; everything else comes from this request.

Parameters:

data (Any)

description: str | None = None#
name: str#
request_template: roboto.ai.agent_thread.record.StartAgentThreadRequest#
variables: list[TemplateVariable] = None#
class roboto.ai.agent.LaunchAgentRequest(/, **data)#

Bases: pydantic.BaseModel

Wire payload for POST /v1/ai/agents/<agent_id>/launch.

The route resolves the named agent, substitutes values into its body, and starts a new AgentThread whose visibility comes from visibility and whose created_from_agent_id points back at the source agent.

Parameters:

data (Any)

values: dict[str, str] = None#

Variable name to caller-supplied value. Keys must match declared TemplateVariable names; values are coerced to str before being spliced into placeholders.

visibility: roboto.ai.agent_thread.record.ThreadVisibility#

Visibility of the resulting AgentThread. Invoke-time wins: this value overrides whatever the agent’s request_template.visibility field carries. Agents cannot pin visibility — authors who need to convey recommended visibility do so in the agent’s description. Defaults to ORG because agents exist to share workflows across teammates; PRIVATE is an explicit opt-out per invocation.

class roboto.ai.agent.TemplateVariable(/, **data)#

Bases: pydantic.BaseModel

A named slot in an AgentRecord that must be filled before the agent can be invoked.

Variables are declared up-front rather than inferred from the body so that the invoke UI can render typed inputs (dataset pickers, etc.) and so that a typo in the body ({{dataste.id}}) surfaces as a save-time validation error rather than a silent extra prompt at invoke time. The declared set and the set of placeholders parsed from the body must match exactly - see AgentRecord._validate_variables_match_placeholders().

Parameters:

data (Any)

collection_content_type: roboto.domain.collections.record.CollectionResourceType | None = None#

constrains the invoke-page collection picker to collections holding this resource type (e.g. event collections for a create-events goal). None leaves the picker unconstrained. Must be None for every other variable type — see _validate_resolvable().

Type:

Only meaningful when type is TemplateVariableType.COLLECTION_ID

default: str | None = None#

Default value substituted when the caller omits the variable. Coerced to str; types are a UI concern.

description: str | None = None#

Human-readable explanation rendered next to the input on the invoke page.

name: str#

Lookup key for substitution. Must match _VARIABLE_NAME_RE; see that pattern for the dotted-name namespace convention. Two variables sharing a prefix (dataset.id + dataset.name) without an upstream expander produce two unlinked inputs at invoke time — that’s a UI-authoring footgun, not an SDK contract.

required: bool = True#

If True the resolver raises UnresolvedAgentVariablesError when no value is supplied and no default is set.

type: TemplateVariableType#

UI hint for the invoke page. See TemplateVariableType.

class roboto.ai.agent.TemplateVariableType#

Bases: roboto.compat.StrEnum

How a variable is interpreted by the invoke-page UI and the invoke-time existence validator.

The substitution engine itself is type-agnostic: every value gets spliced in as a string. The type drives two separate consumers: the UI picks a richer input control (dataset picker, device picker), and the service-side invoke handler runs an existence check on typed values so a bad id fails at the invoke boundary instead of inside the agent’s first tool call. See roboto_service.routes.ai_router._VALUE_VALIDATORS.

COLLECTION_ID = 'collection_id'#

A Roboto collection identifier. UI renders a collection picker; invoke-time validator asserts the collection exists in the caller’s org.

DATASET_ID = 'dataset_id'#

A Roboto dataset identifier. UI renders a dataset picker; invoke-time validator asserts the dataset exists in the caller’s org.

DEVICE_ID = 'device_id'#

A Roboto device identifier. UI renders a device picker; invoke-time validator asserts the device is registered in the caller’s org.

STRING = 'string'#

Free-form text. The default; renders as a plain text input. No existence check (no entity to check against).

exception roboto.ai.agent.UnknownAgentVariablesError(names)#

Bases: AgentResolutionError

Caller supplied values for variables the agent no longer declares — almost always a stale invoke form. Mapped to 400 so the UI can refetch.

Parameters:

names (list[str])

names#
exception roboto.ai.agent.UnresolvedAgentVariablesError(names)#

Bases: AgentResolutionError

Required variables had no supplied value and no default. Mapped to 400 with the names so the invoke page can highlight the empty inputs.

Parameters:

names (list[str])

names#
class roboto.ai.agent.UpdateAgentRequest(/, **data)#

Bases: pydantic.BaseModel

Wire payload for PUT /v1/ai/agents/<agent_id>.

Uses the NotSetType sentinel to distinguish “field omitted” from “field explicitly set to None” so partial updates don’t accidentally null out unrelated fields.

Parameters:

data (Any)

description: str | None | roboto.sentinels.NotSetType#
model_config#

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

name: str | roboto.sentinels.NotSetType#
request_template: roboto.ai.agent_thread.record.StartAgentThreadRequest | roboto.sentinels.NotSetType#
variables: list[TemplateVariable] | roboto.sentinels.NotSetType#
roboto.ai.agent.extract_placeholders(body)#

Return every {{name}} placeholder referenced anywhere in body.

Walks the JSON form of the request rather than the Pydantic model so the traversal is type-agnostic - placeholders inside label_vocabulary values, message text, system prompt, etc. all get picked up without needing per-field logic. Only string values are scanned; dict keys are intentionally ignored so key-collision footguns never arise.

Parameters:

body (roboto.ai.agent_thread.record.StartAgentThreadRequest)

Return type:

set[str]

roboto.ai.agent.resolve_agent(agent, values)#

Substitute values into agent.request_template and return a fully-validated StartAgentThreadRequest.

Walks the body’s JSON form swapping every {{name}} occurrence in a string leaf. Embedded substitution is supported. Dict keys are never touched — placeholder syntax in keys is rejected at save time.

Raises:

  • UnknownAgentVariablesError – values contains keys not declared on the agent (typically a stale invoke form).

  • UnresolvedAgentVariablesError – a required variable has neither a supplied value nor a default; carries the offending names.

  • pydantic.ValidationError – the substituted body failed StartAgentThreadRequest validation — e.g. a resolved value doesn’t match a field-level regex or enum.

Parameters:
Return type:

roboto.ai.agent_thread.record.StartAgentThreadRequest

On this page