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

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) -> TaskResult:
        response = self.agent.chat(message, session_id=session_id)
        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

​
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 February 10, 2026