aixplain.v2.agent

Agent module for aiXplain v2 SDK.

ConversationMessage Objects

class ConversationMessage(TypedDict)

[view_source]

Type definition for a conversation message in agent history.

Attributes:

• role - The role of the message sender, either 'user' or 'assistant'
• content - The text content of the message
• attachments - Optional attachments — hosted-URL/local-path strings or dicts with url or path (plus optional type/name/mimeType).
• files - Deprecated. Local file paths to upload — pass through attachments.

validate_history

def validate_history(history: List[Dict[str, Any]]) -> bool

[view_source]

Validates conversation history for agent sessions.

This function ensures that the history is properly formatted for agent conversations, with each message containing the required 'role' and 'content' fields and proper types.

Arguments:

• history - List of message dictionaries to validate

Returns:

• bool - True if validation passes

Raises:

• ValueError - If validation fails with detailed error messages

Example:

>>> history = [
...     {"role": "user", "content": "Hello"},
...     {"role": "assistant", "content": "Hi there!"}
... ]
>>> validate_history(history)  # Returns True

OutputFormat Objects

class OutputFormat(str, Enum)

[view_source]

Output format options for agent responses.

ContextOverflowStrategy Objects

class ContextOverflowStrategy(str, Enum)

[view_source]

Strategy for condensing the working context when a run exceeds the model's context window.

Context condensation shapes only the working context sent to the model on a given run. It does not modify Shared Memory or the stored session history — the complete session history is always retained.

Attributes:

• TRUNCATE - Default. Remove the oldest unprotected turns until the context fits.
• SUMMARIZE - Summarize older context into a shorter form the model can still use. Retains more of the conversation's meaning than truncation, but adds latency and model cost.

Notes:

Available in SDK 0.2.46+ (use the current 0.2.47). Set it as the agent's saved default (agent.context_overflow_strategy) or override it per run via execution_params={"context_overflow_strategy": ...}. Precedence, highest first: per-run override -> saved agent setting -> Agent Engine default (truncate).

AgentRunParams Objects

class AgentRunParams(BaseRunParams)

[view_source]

Parameters for running an agent.

Attributes:

• session - Conversation thread to run within. A Session instance or a session id string. Omit for a one-shot, stateless run. Replaces the removed via_session flag and id-only session_id.
• query - The query to run
• variables - Variables to replace {{variable}} placeholders in instructions and description. The backend performs the actual substitution.
• tasks - List of tasks for the agent
• prompt - Custom prompt override
• history - Conversation history
• execution_params - Execution parameters (maxTokens, etc.). Passing max_iterations here is deprecated; set agent.budget.max_iterations instead. A deprecated value is folded into budget.max_iterations (the agent's budget wins on conflict) and the standalone key is not emitted.
• criteria - Criteria for evaluation
• evolve - Evolution parameters
• inspectors - Inspector configurations
• run_response_generation - Whether to run response generation. Defaults to False.
• attachments - Multimodal attachments for the turn. Each entry is a hosted-URL/local-path string or a dict with url or path (plus optional type/name/mimeType). Local paths are uploaded to aiXplain storage automatically.
• files - Deprecated. Local file paths to upload — pass through attachments instead.
• progress_format - Display format - "status" (single line) or "logs" (timeline). If None (default), progress tracking is disabled.
• progress_verbosity - Detail level - 1 (minimal), 2 (thoughts), 3 (full I/O)
• progress_truncate - Whether to truncate long text in progress display
• _progress_tracker - Internal. The tracker owned by the run() / sync_poll() call in progress, handed down to on_poll. Never sent to the backend.

Budget Objects

@dataclass_json

@dataclass
class Budget()

[view_source]

Budget caps governing an agent run (cost / duration / iterations).

Every Agent owns a budget (defaulting to an empty Budget()), mutated in place via attribute access — mirroring model.inputs:

agent.budget.max_cost = 0.5
agent.budget.max_iterations = 10

The same object serves two roles: agent.save() persists it as the agent's default budget, and agent.run(...) sends its current state as the run-time budget (the backend merges the run-time budget field-by-field over the persisted default). The Python API is snake_case; serialization produces the agreed camelCase wire keys (maxCost / maxDurationSeconds / maxIterations). All fields are optional and None fields are dropped from to_dict().

Artifact Objects

@dataclass_json

@dataclass
class Artifact()

[view_source]

A user-facing deliverable produced during an agent run.

Artifacts are captured by the agent engine and come from two sources:

• source="tool_output" — media a tool generated (image/audio/video/page). Carries a url, usually a presigned URL.
• source="workspace" — a file the agent wrote into its workspace. Carries inline UTF-8 text in content (binary workspace files are skipped by the engine; there is no uploader yet).

Exactly one of url / content is populated.

Warning: url_expires_at is when the presigned URL dies, not the artifact. Observed in the wild: a 24h window on a generated image URL. If you persist artifact URLs (database, cache, sent email), re-host the bytes before url_expires_at or the links will rot.

category and source are plain strings, not enums: the engine may add new media categories before this SDK knows about them, and an unknown value must pass through rather than raise.

Both wire casings deserialize: the poll/checkRequest path emits snake_case (mime_type) while the webhook body is camelCased (mimeType). If a payload somehow carries both spellings of a field, the one appearing last in the payload wins.

AgentResponseData Objects

@dataclass_json

@dataclass
class AgentResponseData()

[view_source]

Data structure for agent response.

__post_init__

def __post_init__() -> None

[view_source]

Normalize artifacts and assemble governance from flat wire fields.

AgentRunResult Objects

@dataclass_json

@dataclass
class AgentRunResult(Result)

[view_source]

Result from running an agent.

data

Override type from base class

__post_init__

def __post_init__() -> None

[view_source]

Promote diagnostic codes the backend nests under data.

The poll body carries them at data.diagnosticErrorCodes (or only inside executionStats on older builds), never top-level.

artifacts

@property
def artifacts() -> List[Artifact]

[view_source]

Deliverables produced during the run (see Artifact).

Always a list — empty when the run produced nothing, when artifact capture is disabled, or when the backend predates artifact support.

execution_id

@property
def execution_id() -> Optional[str]

[view_source]

Extract the execution ID from the poll URL or request_id.

The execution ID can be used with Agent.poll() and Agent.sync_poll() to resume polling a previously started run without persisting the full URL.

Returns:

The execution ID if available, None otherwise.

debug

def debug(prompt: Optional[str] = None,
          execution_id: Optional[str] = None,
          **kwargs: Any) -> "DebugResult"

[view_source]

Debug this agent response using the Debugger meta-agent.

This is a convenience method for quickly analyzing agent responses to identify issues, errors, or areas for improvement.

Note: This method requires the AgentRunResult to have been created through an Aixplain client context. If you have a standalone result, use the Debugger directly: aix.Debugger().debug_response(result)

Arguments:

• prompt - Optional custom prompt to guide the debugging analysis.
• Examples - "Why did it take so long?", "Focus on error handling"
• execution_id - Optional execution ID (poll ID) for the run. If not provided, it will be extracted from the response's request_id or poll URL. This allows the debugger to fetch additional logs and information.
• **kwargs - Additional parameters to pass to the debugger.

Returns:

• DebugResult - The debugging analysis result.

Raises:

• ValueError - If no client context is available for debugging.

Example:

agent = aix.Agent.get("my_agent_id")
response = agent.run("Hello!")
debug_result = response.debug()  # Uses default prompt
debug_result = response.debug("Why did it take so long?")  # Custom prompt
debug_result = response.debug(execution_id="abc-123")  # With explicit ID
print(debug_result.analysis)

Task Objects

@dataclass_json

@dataclass
class Task()

[view_source]

A task definition for agent workflows.

__post_init__

def __post_init__() -> None

[view_source]

Initialize task dependencies after dataclass creation.

Agent Objects

@dataclass_json

@dataclass(repr=False)
class Agent(BaseResource, SearchResourceMixin[BaseSearchParams, "Agent"],
            GetResourceMixin[BaseGetParams,
                             "Agent"], DeleteResourceMixin[BaseDeleteParams,
                                                           "Agent"],
            RunnableResourceMixin[AgentRunParams, AgentRunResult])

[view_source]

Agent resource class.

__post_init__

def __post_init__() -> None

[view_source]

Initialize agent after dataclass creation.

__setattr__

def __setattr__(name: str, value: Any) -> None

[view_source]

Keep self.budget a (never-None) Budget instance, and note role assignments.

Assigning agent.budget a dict / Budget / None is coerced into a Budget so attribute access (agent.budget.max_cost = ...) always works and the invariant "budget is never None" holds (mirrors how Model.__setattr__ coerces bulk inputs assignment). This runs for the generated __init__ assignment too, so the field is a Budget by the time __post_init__ executes.

A role ref assigned after hydration is the caller's intent and must always be sent on save, even when it happens to equal the class default (BUG-1093). Assignments made by the generated __init__ are not recorded — _explicit_roles does not exist yet at that point — but such an object also has no recorded server fields, so suppression is off for it anyway. Hydration resets the set (see _record_server_fields).

mark_as_deleted

def mark_as_deleted() -> None

[view_source]

Mark the agent as deleted by setting status to DELETED and calling parent method.

before_run

def before_run(*args: Any,
               **kwargs: Unpack[AgentRunParams]) -> Optional[AgentRunResult]

[view_source]

Hook called before running the agent to validate and prepare state.

on_poll

def on_poll(response: AgentRunResult,
            **kwargs: Unpack[AgentRunParams]) -> None

[view_source]

Hook called after each poll to update progress display.

Arguments:

• response - The poll response containing progress information
• **kwargs - Run parameters

after_run

def after_run(result: Union[AgentRunResult, BaseException], *args: Any,
              **kwargs: Unpack[AgentRunParams]) -> Optional[AgentRunResult]

[view_source]

Hook called after running the agent for result transformation.

Also reached on the failure path, where result is the raised exception (any BaseException, KeyboardInterrupt included). The progress display is not torn down here: run() owns it, so an override that skips super() cannot leak the display thread.

run

def run(*args: Any, **kwargs: Unpack[AgentRunParams]) -> AgentRunResult

[view_source]

Run the agent with optional progress display.

Arguments:

• *args - Positional arguments (first arg is treated as query)
• query - The query to run
• session - Run within a conversation thread. Accepts a Session instance or a session id string. When supplied, the run routes through the session path: the user message is posted to POST /v1/sessions/{id}/messages (carrying the session's executionConfig plus any per-run execution overrides) and the triggered agent run is awaited. Omit session for a one-shot, stateless run over POST /v2/agents/{id}/run. There is no via_session flag and no id-only session_id — manage threads through aix.Session and pass them here.
• progress_format - Display format - "status" or "logs". If None (default), progress tracking is disabled.
• progress_verbosity - Detail level 1-3 (default: 1)
• progress_truncate - Truncate long text (default: True)
• **kwargs - Additional run parameters

Returns:

• AgentRunResult - The result of the agent execution

run_async

def run_async(*args: Any, **kwargs: Unpack[AgentRunParams]) -> AgentRunResult

[view_source]

Run the agent asynchronously.

Arguments:

• *args - Positional arguments (first arg is treated as query)
• query - The query to run
• **kwargs - Additional run parameters

Returns:

• AgentRunResult - The result of the agent execution. Use result.url to poll for completion via sync_poll(result.url) or client.get(result.url). Do not construct /sdk/runs/{execution_id} — that endpoint is not supported for agent runs.

Notes:

progress_format is ignored here and logged as a warning: this call returns as soon as the run is submitted, so there is nothing to display. To watch a run started this way, pass the progress kwargs to the poll instead:

r = agent.run_async("hi")
agent.sync_poll(r.url, progress_format="status")

poll

def poll(poll_url: str, timeout: Optional[float] = None) -> AgentRunResult

[view_source]

Poll for the result of an asynchronous agent execution.

Unlike the base implementation, poll_url may be either a full URL (as returned in AgentRunResult.url) or a bare execution ID. When an execution ID is provided the correct /sdk/agents/{id}/result endpoint is used automatically, avoiding the common mistake of calling the unsupported /sdk/runs/{id} endpoint.

Arguments:

• poll_url - Full poll URL or execution ID.
• timeout - Optional upper bound, in seconds, on this single request's read phase. See RunnableResourceMixin.poll.

Returns:

AgentRunResult with current execution status.

sync_poll

def sync_poll(poll_url: str,
              **kwargs: Unpack[AgentRunParams]) -> AgentRunResult

[view_source]

Poll until an asynchronous agent execution completes.

Accepts either a full URL or a bare execution ID (see poll for details).

Arguments:

• poll_url - Full poll URL or execution ID.
• **kwargs - Run parameters including timeout and wait_time. progress_format / progress_verbosity / progress_truncate render a live progress display for the duration of the poll, exactly as on run; this is how a run started with run_async is watched.

Returns:

AgentRunResult with final execution status.

save

def save(*args: Any, **kwargs: Any) -> "Agent"

[view_source]

Save the agent with dependency management.

This method extends the base save functionality to handle saving of dependent child components before the agent itself is saved.

Arguments:

• *args - Positional arguments passed to parent save method.
• save_subcomponents - bool - If True, recursively save all unsaved child components (default: False)
• as_draft - bool - If True, save agent as draft status (default: False)
• **kwargs - Other attributes to set before saving

Returns:

• Agent - The saved agent instance

Raises:

• ResourceError - If the agent has been deleted.
• ValueError - If child components are not saved and save_subcomponents is False

before_save

def before_save(*args: Any, **kwargs: Any) -> Optional[dict]

[view_source]

Callback to be called before the resource is saved.

Handles status transitions based on save type.

after_duplicate

def after_duplicate(result: Union["Agent", Exception],
                    **kwargs: Any) -> Optional["Agent"]

[view_source]

Callback called after the agent is duplicated.

Sets the duplicated agent's status to DRAFT.

duplicate

@with_hooks
def duplicate(duplicate_subagents: bool = False,
              name: Optional[str] = None) -> "Agent"

[view_source]

Duplicate this agent on the aiXplain platform (server-side).

Creates a server-side copy of this agent with a clean usage baseline. The duplicate inherits the original's ownership, team, and permissions but resets all usage and cost metrics.

Arguments:

• duplicate_subagents - If True, recursively duplicates referenced subagents so the duplicate has independent copies. If False, the duplicate keeps references to the original subagents. Defaults to False.
• name - Custom name for the duplicate. If None, a unique name is auto-generated by the platform. Defaults to None.

Returns:

• Agent - The newly created duplicate agent.

Raises:

• ResourceError - If the duplication request fails.

search

@classmethod
def search(cls: type["Agent"],
           query: Optional[str] = None,
           **kwargs: Unpack[BaseSearchParams]) -> "Page[Agent]"

[view_source]

Search agents with optional query and filtering.

Arguments:

• query - Optional search query string
• **kwargs - Additional search parameters (ownership, status, etc.)

Returns:

Page of agents matching the search criteria

llm_id

@property
def llm_id() -> str

[view_source]

Return main LLM id whether llm is a string or Model.

build_save_payload

def build_save_payload(**kwargs: Any) -> dict

[view_source]

Build the payload for the save action.

build_run_payload

def build_run_payload(**kwargs: Unpack[AgentRunParams]) -> dict

[view_source]

Build the payload for the run action.

SDK-control kwargs (_RUN_CONTROL_KEYS: retries/timeouts, the progress_* display trio, api_key / resource_path, and the header-only run metadata) are dropped up front. The run path already filters them via _payload_kwargs_for_run; repeating it here means the catch-all snake_case→camelCase forwarder below cannot put them on the wire even when this builder is called directly (BUG-1091).