Skip to main content
The Netra SDK exposes a simulation client that lets you:
  • Run simulations - Execute multi-turn conversations against your AI agent
  • Define tasks - Create custom task implementations to wrap your agent
  • Handle file attachments - Receive base64-encoded files from dataset items in your task
  • Control concurrency - Manage parallel execution for throughput
This page shows how to use Netra.simulation to run multi-turn simulations and test your AI agents programmatically.

​
Getting Started

The simulation client is available on the main Netra entry point after initialization. 
from netra import Netra

Netra.init(app_name="sample-app")

# Access the simulation client
Netra.simulation.run_simulation(...)

​
run_simulation

Execute a multi-turn conversation simulation against a dataset. Your task function is called repeatedly for each turn until the conversation completes. 
from netra import Netra
from netra.simulation.task import BaseTask
from netra.simulation.models import TaskResult, ProcessedFile

Netra.init(app_name="sample-app")

class MyAgentTask(BaseTask):
    def __init__(self, agent):
        self.agent = agent
    
    def run(
        self,
        message: str,
        session_id: str | None = None,
        files: list[ProcessedFile] | None = None,
    ) -> TaskResult:
        response = self.agent.chat(message, session_id=session_id, files=files)
        return TaskResult(
            message=response.text,
            session_id=session_id or "default",
        )

result = Netra.simulation.run_simulation(
    name="Customer Support Simulation",
    dataset_id="dataset-123",
    task=MyAgentTask(my_agent),
    context={"environment": "staging"},
    max_concurrency=5,
)

print(f"Completed: {len(result['completed'])}")
print(f"Failed: {len(result['failed'])}")

​
Parameters

ParameterTypeDescription
namestrName of the simulation run (required)
dataset_idstrID of the dataset containing conversation scenarios
taskBaseTaskTask instance that wraps your AI agent
contextdict?Optional context data passed to the simulation
max_concurrencyintMaximum parallel conversations (default: 5)

​
Response

FieldTypeDescription
successboolOverall success status
total_itemsintNumber of dataset items
completedlist[dict]Successfully completed items
failedlist[dict]Failed items with error details
FieldTypeDescription
run_item_idstrUnique item identifier
successboolAlways True for completed items
final_turn_idstrID of the last turn in the conversation
FieldTypeDescription
run_item_idstrUnique item identifier
successboolAlways False for failed items
errorstrError message describing the failure
turn_idstrID of the turn where failure occurred

​
BaseTask

Create a custom task by inheriting from the BaseTask abstract base class. Your implementation wraps your AI agent and handles the conversation flow. 
from abc import ABC, abstractmethod
from typing import Optional, Awaitable
from netra.simulation.models import ProcessedFile, TaskResult

class BaseTask(ABC):
    @abstractmethod
    def run(
        self,
        message: str,
        session_id: Optional[str] = None,
        files: Optional[list[ProcessedFile]] = None,
    ) -> TaskResult | Awaitable[TaskResult]:
        """
        Execute a single turn in the conversation.
        
        Args:
            message: The input message from the simulation.
            session_id: The session identifier (optional).
                        Will be None for the first turn of a conversation.
            files: Optional list of base64-encoded file attachments from the
                   dataset item.  Will be None when no files are attached.
            
        Returns:
            TaskResult: The task result containing the response message and session ID.
        """
        pass

​
Implementation Requirements

RequirementDescription
Inherit BaseTaskYour class must inherit from the BaseTask class
Implement run()The run() method must return a TaskResult (can be async)
Handle sessionsMaintain conversation context using the session_id
Return messageAlways return a response message in the TaskResult
Accept filesThe run() method receives an optional files parameter containing base64-encoded attachments

​
File Handling

Dataset items can include file attachments. When present, the simulation framework automatically detectes the required files for a turn and delivers it to your task’s run() method as a list of ProcessedFile objects.

​
ProcessedFile

Each file delivered to your task is a ProcessedFile instance with the following fields:
FieldTypeDescription
file_namestrOriginal file name
content_typestrMIME type of the file content
descriptionstr?Optional description of the file
datastrBase64-encoded file content

​
Complete Example

from netra import Netra
from netra.simulation.task import BaseTask
from netra.simulation.models import TaskResult
from openai import OpenAI

# Initialize
Netra.init(
    app_name="simulation-demo",
    headers="x-api-key=your-api-key",
)
client = OpenAI()

# Store conversation history per session
conversations: dict[str, list] = {}

class OpenAITask(BaseTask):
    """Task that uses OpenAI for multi-turn conversations."""
    
    def run(self, message: str, session_id: str | None = None) -> TaskResult:
        import uuid
        session = session_id or str(uuid.uuid4())
        
        # Initialize conversation
        if session not in conversations:
            conversations[session] = [
                {"role": "system", "content": "You are a helpful customer support agent."}
            ]
        
        # Add user message
        conversations[session].append({"role": "user", "content": message})
        
        # Call OpenAI
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=conversations[session]
        )
        
        # Store and return response
        content = response.choices[0].message.content
        conversations[session].append({"role": "assistant", "content": content})
        
        return TaskResult(message=content, session_id=session)

# Run simulation
result = Netra.simulation.run_simulation(
    name="Customer Support Test",
    dataset_id="your-dataset-id",
    task=OpenAITask(),
    context={"model": "gpt-4o-mini", "temperature": 0},
    max_concurrency=5,
)

# Analyze results
print(f"Total: {result['total_items']}")
print(f"Completed: {len(result['completed'])}")
print(f"Failed: {len(result['failed'])}")

for failure in result["failed"]:
    print(f"  Failed {failure['run_item_id']}: {failure['error']}")

Netra.shutdown()

​
Next Steps

Last modified on June 17, 2026