Agent

Custom AI agents for specialized tasks within your apps.

Agent Interface

DispatcherAgentInterface

DispatcherAgentInterface(*, default_prompts: Optional[list[str]] = None, options: Optional[LLMOptions] = None, **kwargs)

Abstract Base Class for a dispatcher agent.

This interface defines the contract for agents that handle the lifecycle of a chat interaction. This includes pre-processing user prompts, streaming responses from a language model, and post-processing the final response.

It provides hooks (on_llm_request, on_llm_response) that can be used to customize the agent's behavior.

Initializes the DispatcherAgentInterface.

Parameters:

• default_prompts (Optional[list[str]], default: None ) –

  An optional list of default prompts to be used by the agent.

• options (Optional[LLMOptions], default: None ) –

  LLMOptions to pass to every LLM call (e.g. reasoning_effort for reasoning models).

init async

init(redis_client: Redis, chat_stream_channel: str, chat_context: RawChatContext)

Initializes the agent with a Redis client and a channel for streaming.

This will be called by the platform prior to calling the run function

Parameters:

• redis_client (Redis) –

  An asynchronous Redis client instance.

• chat_stream_channel (str) –

  The name of the Redis channel to publish stream events to.

publish_message async

publish_message(token: str)

Publish a message token.

Automatically sends analysis_trace_end on first call if analysis trace was active.

Draft replacement behavior - First call: Replaces any draft message in the analysis trace (draft disappears, message appears) - Subsequent calls: Append to the message (no replacement)

If you called push_analysis_trace_draft() before this, the first publish_message() will replace the entire draft content with the message content. Additional publish_message() calls add to the message incrementally.

Example: await agent.push_analysis_trace_draft("Draft: The data suggests...") await agent.publish_message("The analysis shows...") # Replaces draft await agent.publish_message(" with 95% confidence.") # Appends to message

publish_sources async

publish_sources(sources: list[ChatSourceCard])

Publish source_data (clickable chips) from post-processing LLM response

publish_element_data async

publish_element_data(element, text: str | None = None)

Publish an element inline at the current message position.

Automatically inserts element marker and publishes metadata. The element will be rendered inline where the marker appears in the message stream.

Supported elements: Table, PlotlyPlot

Parameters:

• element –

  SDK Element to publish (must have ID)

• text (str | None, default: None ) –

  Optional text to publish before the element marker

Raises:

• TypeError –

  If element is not a supported type

• ValueError –

  If element has no ID (not added to page) Example: await self.publish_element_data(table, "Here's the data:")

push_response_metadata async

push_response_metadata(meta: dict) -> None

Store data in the final response produced in the current conversation turn. If this method is called multiple times, the newly provided meta dict will be merged with the old one and already present keys will be overridden.

    self.push_response_metadata({"counter": 1}) // counter: 1
    self.push_response_metadata({"counter": 5}) // counter: 5

The last stored metadata can be retrieved in the next conversation turn by using the get_message_meta method of this class.

Parameters:

• meta (dict) –

  Metadata dictionary to store

trigger_page_refresh async

trigger_page_refresh()

Publish a request to refresh the current page

publish_chat_end async

publish_chat_end()

Publish chat end event.

Automatically closes analysis trace phase if active.

publish_action_page_update async

publish_action_page_update(page_update_data: ActionPageUpdate)

Publish a message that triggers a page update action (or filter update action)

publish_thinking async

publish_thinking(msg: str)

Update the IRIS "Thinking" progress text while waiting for the first token.

This ephemeral status message is only sent if analysis trace has not started yet. Once analysis trace begins, it replaces the thinking indicator.

Parameters:

• msg (str) –

  New text that will replace the original text

start_analysis_trace async

start_analysis_trace(title: str = 'Thinking')

Start Analysis Trace with a global title.

The title becomes the header for the entire analysis trace in the UI and database. This is a global property of the trace, separate from individual step titles.

Args: title: Global title for the entire analysis trace (e.g., "Analyzing data", "Searching database"). Shown as the header of the collapsible analysis trace section in the UI.

Note: - This is optional - auto-sent on first push_analysis_trace_text() or push_analysis_trace_step() - Can only be set once per trace - subsequent calls are ignored with a warning - If not called explicitly, defaults to "Thinking"

push_analysis_trace_text async

push_analysis_trace_text(content: str)

Push free-form analysis trace text (type: "text" in DB).

Use this for streaming unstructured analysis or thinking content. Multiple consecutive calls are automatically consolidated into a single text item in the database.

Automatically starts trace on first call if not already started (with default title "Thinking"). To customize the trace title, call start_analysis_trace(title=...) before this method.

Args: content: Text content chunk. Chunks are concatenated WITHOUT adding spaces - ensure your content includes necessary whitespace (spaces, newlines). Supports markdown formatting.

Consolidation behavior: - Consecutive push_analysis_trace_text() calls merge into ONE "text" item - Text chunks are joined with NO separator - include spaces in your content - A push_analysis_trace_step() call breaks the consolidation and starts a new item

Example: await agent.push_analysis_trace_text("Let me analyze this request. ") await agent.push_analysis_trace_text("First, I'll check the data source.") # Results in ONE item: "Let me analyze this request. First, I'll check..."

Example with custom title: await agent.start_analysis_trace(title="Custom Analysis Title") await agent.push_analysis_trace_text("Analyzing your request...")

Note: Cannot be called after publish_message() or push_analysis_trace_draft() (V1 constraint). Draft must be the final trace item.

push_analysis_trace_step async

push_analysis_trace_step(title: str, content: str | None = None, icon: str | None = None)

Push structured reasoning step (type: "step" in DB).

    Use this for discrete, titled sections of your analysis. Each call creates a
    separate step item in the database, unlike push_analysis_trace_text() which consolidates.

    Automatically starts trace on first call if not already started.

    Args:
        title: Step title (e.g., "Unpacking dashboard content", "Using Dynamic Search tool").
               This is the title for THIS SPECIFIC STEP, not the global trace title.
               Each step can have its own unique title.
        content: Optional markdown content for this step. Use markdown lists for nested items.
                 Unlike text chunks, this content is NOT concatenated with other steps.
        icon: Optional icon variant for the step indicator. Valid values: "filled", "checkmark".
              If None, frontend uses default icon ("filled").

    Differences from push_analysis_trace_text():
        - Each call creates a SEPARATE item (no consolidation)
        - Has a step-specific title (shown in the UI for this step only)
        - Breaks any ongoing text consolidation

    When to use:
        - Discrete phases of analysis: "Loading data", "Analyzing patterns", "Generating summary"
        - Tool usage: "Using Dynamic Search", "Calling weather API"
        - Checklists: title + markdown list in content
        - Any time you want a labeled subsection

    Example:
        await agent.push_analysis_trace_step("Unpacking dashboard content")
        await agent.push_analysis_trace_step(
            "Checking the following",
            content="1. historical context

1. dashboard analysis" ) await agent.push_analysis_trace_step( "Using Dynamic Search tool", content="Searching records in All Units table" ) await agent.push_analysis_trace_step("Complete", icon="checkmark") # Results in FOUR separate step items, each with its own title and icon

   Note:
       Cannot be called after publish_message() or push_analysis_trace_draft() (V1 constraint).
       Draft must be the final trace item.
   

stop_analysis_trace async

stop_analysis_trace()

Manually end the reasoning trace phase.

This is optional - the reasoning trace is automatically ended when you call publish_message() for the first time.

Use this method if you want explicit control over when the reasoning phase ends, for example if you want to add a visual separation before starting the message.

Note: - Calling this multiple times is safe (subsequent calls are ignored) - Automatically called on first publish_message() if not already called - Automatically called on publish_exception() or chat_end if trace is still open

start_analysis_trace_draft async

start_analysis_trace_draft(title: str = 'Drafting response', icon: str | None = 'filled')

Configure draft entry metadata before streaming draft content.

Call this before push_analysis_trace_draft() if you want to customize the title and icon shown for the draft entry. If not called, push_analysis_trace_draft() will auto-send draft start with defaults (title="Drafting response", icon="filled").

Args: title: Title text for the draft entry (e.g., "Generating answer", "Preparing response"). Default: "Drafting response" icon: Icon variant for the draft entry indicator. Valid values: "filled", "checkmark". Default: "filled" (neutral, consistent with step default).

Example: await agent.start_analysis_trace_draft(title="Generating answer", icon="checkmark") await agent.push_analysis_trace_draft("Based on my analysis...") await agent.push_analysis_trace_draft(" the trend is positive.") await agent.publish_message("Final answer here")

Note: - Must be called BEFORE push_analysis_trace_draft() - Automatically starts the analysis trace if not already started - Calling multiple times is safe (subsequent calls are ignored)

push_analysis_trace_draft async

push_analysis_trace_draft(content: str, **draft_config)

Push ephemeral draft message (24h TTL, Redis-backed).

Draft content is streamed to the frontend for display, but stored in Redis (not DB) with a 24-hour TTL. A draft_id reference is saved in the DB for recovery when fetching conversation history.

Must be called BEFORE publish_message() (V1 constraint - no interleaving). Automatically starts analysis trace if not already started.

Draft replacement behavior The first publish_message() call replaces the entire draft with the message content. Subsequent publish_message() calls append to the message (not to the draft).

Multiple consecutive push_analysis_trace_draft() calls are consolidated into a single draft item (similar to text consolidation).

Args: content: Draft message content chunk (preview of the final response). Consecutive draft chunks are concatenated with NO separator. **draft_config: Optional configuration passed to start_analysis_trace_draft() on first call. Supports 'title' (str) and 'icon' (str). Ignored on subsequent calls.

Example (inline config): await agent.push_analysis_trace_draft("Based on the data...", icon="checkmark") await agent.push_analysis_trace_draft(" the trend is positive.") # Config ignored

Example (explicit config): await agent.start_analysis_trace_draft(title="Generating answer", icon="checkmark") await agent.push_analysis_trace_draft("Based on the data...") await agent.push_analysis_trace_draft(" the trend is positive.")

Example (default config): await agent.push_analysis_trace_draft("Content...") # Auto-starts with defaults

Note: - Draft is shown in the analysis trace UI as the last item before trace ends - Backend generates draft_id internally (not exposed to frontend) - Frontend receives only content text, backend handles Redis storage/recovery - Draft content expires after 24 hours (Redis TTL) - draft_config only used on first call (when draft_start event is sent)

publish_exception async

publish_exception(message: str)

Publish exception event.

Automatically closes reasoning phase if active.

get_attachment_contents async

get_attachment_contents(attachment_id: str | None = None) -> list[AttachmentContent] | AttachmentContent | None

Read attachment files from S3.

Without arguments, returns all attachments. With attachment_id, returns a single AttachmentContent or None if not found.

get_message_meta async staticmethod

get_message_meta(chat_context: RawChatContext, message_id: int | None = None) -> dict | None

Retrieve the metadata defined in a message. There are two ways of using this method: 1. Do not set the message_id argument and retrieve the metadata associated with the latest generated response, so, the last message saved with the text pushed with the publish_message method. 2. Set message_id and retrieve the metadata associated with a specific message.

In case no metadata is definied, this will return None.

get_page_context async staticmethod

get_page_context(chat_context: RawChatContext)

Retrieve the stored page context from when the current page was executed

get_page async staticmethod

get_page(chat_context: RawChatContext) -> Page

A more performant way to get the page object, which can be useful for retrieving element contents and a page representation

summarize_page staticmethod

summarize_page(page_context: dict)

A rudimentary way to express the page which focuses on giving a small amount of context on the page in order to best minimize the size of the content

run abstractmethod async

run(chat_context: RawChatContext)

The main entry point for the agent's execution logic.

Subclasses must implement this method to define how they process a chat request.

Overview

Agents allow you to create specialized AI assistants that can:

• Handle specific types of queries
• Execute custom logic based on user input
• Integrate with external systems
• Maintain conversation context

Creating a Custom Agent

from virtualitics_sdk.llm.agent import DispatcherAgentInterface

class DataAnalysisAgent(DispatcherAgentInterface):
    """Agent specialized in data analysis tasks."""

    async def handle_query(self, query: str, context: dict) -> str:
        """
        Process a user query and return a response.

        Args:
            query: The user's question or request
            context: App context (dataset, previous results, etc.)

        Returns:
            Response string to display to user
        """
        # Access app context
        dataset = context.get('dataset')
        if dataset:
            df = dataset.data

            # Analyze query intent
            if 'summary' in query.lower():
                return self._generate_summary(df)
            elif 'correlation' in query.lower():
                return self._generate_correlation(df)
            else:
                return "I can help you summarize data or find correlations. What would you like to know?"

        return "No dataset available for analysis."

    def _generate_summary(self, df):
        """Generate dataset summary."""
        summary = df.describe().to_string()
        return f"Here's a statistical summary:\n\n{summary}"

    def _generate_correlation(self, df):
        """Generate correlation analysis."""
        numeric_cols = df.select_dtypes(include='number').columns
        if len(numeric_cols) > 1:
            corr = df[numeric_cols].corr()
            return f"Correlation matrix:\n\n{corr.to_string()}"
        return "Not enough numeric columns for correlation analysis."

Using Agents in Apps

from virtualitics_sdk import App, Step

# Create agent instance
analysis_agent = DataAnalysisAgent()

# Attach to app
app = App(
    name="AI Analysis App",
    description="App with custom AI agent",
    agent=analysis_agent
)

# Or attach to specific step
step = MyStep(
    title="Analysis",
    description="Analyze data with AI",
    parent="Main",
    type=StepType.RESULTS,
    page=Page(...),
    agent=analysis_agent
)

Agent with External APIs

import httpx
from virtualitics_sdk.llm.agent import DispatcherAgentInterface

class WeatherAgent(DispatcherAgentInterface):
    """Agent that fetches weather data."""

    async def handle_query(self, query: str, context: dict) -> str:
        # Extract location from query (simplified)
        location = self._extract_location(query)

        if not location:
            return "Please specify a location (e.g., 'weather in San Francisco')"

        # Fetch weather data
        weather = await self._get_weather(location)

        return f"Weather in {location}: {weather['description']}, {weather['temp']}°F"

    async def _get_weather(self, location: str) -> dict:
        """Fetch weather from external API."""
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"https://api.weather.example.com/v1/weather",
                params={"location": location}
            )
            return response.json()

    def _extract_location(self, query: str) -> str:
        """Extract location from query (simplified)."""
        # In practice, use NLP or regex
        words = query.lower().split()
        if 'in' in words:
            idx = words.index('in')
            if idx + 1 < len(words):
                return words[idx + 1]
        return None

Stateful Agents

Maintain conversation state:

class StatefulAgent(DispatcherAgentInterface):
    """Agent that remembers conversation context."""

    def __init__(self):
        super().__init__()
        self.conversation_history = []
        self.user_preferences = {}

    async def handle_query(self, query: str, context: dict) -> str:
        # Add to history
        self.conversation_history.append({
            "query": query,
            "timestamp": datetime.now()
        })

        # Use conversation context
        if self._is_followup_question(query):
            previous_query = self.conversation_history[-2]["query"]
            return f"Following up on '{previous_query}': {self._answer(query)}"

        return self._answer(query)

    def _is_followup_question(self, query: str) -> bool:
        """Check if query is a follow-up."""
        followup_indicators = ["and that", "also", "what about", "how about"]
        return any(indicator in query.lower() for indicator in followup_indicators)

Agent Registry

Register multiple agents for different tasks:

class AgentDispatcher:
    """Dispatch queries to specialized agents."""

    def __init__(self):
        self.agents = {
            "data": DataAnalysisAgent(),
            "weather": WeatherAgent(),
            "code": CodeHelperAgent()
        }

    async def handle_query(self, query: str, context: dict) -> str:
        # Classify query intent
        intent = self._classify_intent(query)

        # Route to appropriate agent
        agent = self.agents.get(intent)
        if agent:
            return await agent.handle_query(query, context)

        return "I'm not sure how to help with that. Try asking about data, weather, or code."

    def _classify_intent(self, query: str) -> str:
        """Classify query to route to correct agent."""
        query_lower = query.lower()

        if any(word in query_lower for word in ["data", "analyze", "statistics"]):
            return "data"
        elif any(word in query_lower for word in ["weather", "forecast", "temperature"]):
            return "weather"
        elif any(word in query_lower for word in ["code", "function", "error"]):
            return "code"

        return None

Best Practices

• Single Responsibility: Each agent should handle one domain well
• Error Handling: Handle API failures and edge cases gracefully
• Context Usage: Leverage app context for better responses
• Performance: Consider caching for expensive operations
• Testing: Test agents with various query types

See Also

• LLM Overview
• Chat
• App - Attaching agents to apps