Simulation

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
Use lifecycle hooks - Run setup/teardown scripts around scenarios (before_all, before_each, before, after, after_each, after_all)
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 import BaseTask, SimulationHooks, TaskResult
from netra.simulation.models import 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,
        setup_context: dict | None = None,
    ) -> TaskResult:
        ctx = setup_context or {}
        response = self.agent.chat(
            message,
            session_id=session_id,
            files=files,
            auth_token=ctx.get("auth_token"),
        )
        return TaskResult(
            message=response.text,
            session_id=session_id or "default",
        )

hooks = SimulationHooks(
    before_all=lambda: {"employee_id": "emp-1"},
    before_each=lambda shared: {"auth_token": "tok-123"},
)

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

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

run_simulation(
    name: str,
    dataset_id: str,
    task: BaseTask,
    context: Optional[dict[str, Any]] = None,
    max_concurrency: int = 5,
    max_turns: int = 50,
    hooks: Optional[SimulationHooks] = None,
) -> Optional[dict[str, Any]]

Parameters

Parameter	Type	Description
`name`	`str`	Name of the simulation run (required)
`dataset_id`	`str`	ID of the dataset containing conversation scenarios
`task`	`BaseTask`	Task instance that wraps your AI agent
`context`	`dict?`	Optional context data passed to the simulation
`max_concurrency`	`int`	Maximum parallel conversations (default: 5)
`max_turns`	`int`	Maximum turns per conversation before aborting (default: 50)
`hooks`	`SimulationHooks?`	Optional lifecycle hooks for setup/teardown (see Hooks)

Response

Field	Type	Description
`success`	`bool`	Overall success status
`total_items`	`int`	Number of dataset items
`completed`	`list[dict]`	Successfully completed items
`failed`	`list[dict]`	Failed items with error details

Completed Item

Field	Type	Description
`run_item_id`	`str`	Unique item identifier
`success`	`bool`	Always `True` for completed items
`final_turn_id`	`str`	ID of the last turn in the conversation

Failed Item

Field	Type	Description
`run_item_id`	`str`	Unique item identifier
`success`	`bool`	Always `False` for failed items
`error`	`str`	Error message describing the failure
`turn_id`	`str`	ID 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 Any, Awaitable, Optional
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,
        setup_context: Optional[dict[str, Any]] = 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.
            setup_context: Optional merged context from lifecycle hooks
                           (`before_all` + `before_each` + item `before`).
                           None when no hooks are configured. Existing tasks
                           that omit this parameter remain backwards compatible.

        Returns:
            TaskResult: The task result containing the response message and session ID.
        """
        pass

from dataclasses import dataclass

@dataclass(slots=True, frozen=True)
class TaskResult:
    message: str      # The response message from your agent
    session_id: str   # The session identifier for conversation continuity

Implementation Requirements

Requirement	Description
Inherit `BaseTask`	Your class must inherit from the `BaseTask` class
Implement `run()`	The `run()` method must return a `TaskResult` (can be async)
Handle sessions	Maintain conversation context using the `session_id`
Return message	Always return a response message in the `TaskResult`
Accept files	The `run()` method receives an optional `files` parameter containing base64-encoded attachments
Accept setup context	Optionally declare `setup_context` to receive merged hook context (tokens, accounts, etc.)

Hooks (Pre/Post Scripts)

Multi-turn datasets often share state across scenarios. Rather than merging scenarios or forcing sequential execution, use hooks to set up and tear down state around scenario runs. Hooks run entirely on the SDK side — script code never leaves your environment. Lightweight metadata (function name and description) is sent to the backend so the Netra UI can show which hooks are configured. In the Netra dashboard (multi-turn Scenario Run Details → Conversation tab), configured hooks appear in collapsible Pre-script / Post-script panels with each hook’s name and description. Item run status badges include Prescript Failed and Postscript Failed.

Hook points

Hook	When it runs	Failure effect
`before_all`	Once, before any scenario starts	Entire run is marked failed; no scenarios execute
`before_each`	Before every scenario	That scenario is marked `prescript_failed` (terminal; eval suppressed); others continue
`before`	Before specific scenarios only (keyed by `dataset_item_id`)	That scenario is marked `prescript_failed` (terminal; eval suppressed); others continue
`after`	After specific scenarios only (keyed by `dataset_item_id`)	If the scenario otherwise succeeded → `postscript_failed` (terminal; eval not suppressed). If already failed/`prescript_failed`, status is preserved
`after_each`	After every scenario (after item-specific `after`)	Same as `after` — both always attempt to run; errors are combined when marking `postscript_failed`
`after_all`	Once, after all scenarios complete	Successfully completed scenarios are marked `postscript_failed`; already-failed/`prescript_failed` items keep their status

Context passing

before_all()                            → shared_context (dict)
before_each(shared_context)             → merged into setup_context
before[dataset_item_id](merged_context) → merged into setup_context (if registered)
BaseTask.run(..., setup_context=merged_context)
after[dataset_item_id](result, setup_context) (if registered)
after_each(result, setup_context)
after_all(results, shared_context)

setup_context is the merge of before_all + before_each + item before. It is passed to BaseTask.run, item after, and after_each. If a before hook fails mid-way, teardown still receives the furthest successfully built setup_context (so after_each can clean up what before_each created). after_all receives only the run-level shared_context, but its results include setup/first-turn failures as well as conversation failures. Prefer before_each / after_each when the same setup/teardown applies to every scenario. Use item-keyed before / after for scenario-specific logic.

SimulationHooks

from netra.simulation import SimulationHooks

hooks = SimulationHooks(
    before_all=setup_environment,          # () -> dict | None
    before_each=setup_each,                # (shared_context) -> dict | None
    before={                               # Dict[dataset_item_id, (shared_context) -> dict | None]
        "refund-scenario-001": setup_refund,
    },
    after={                                # Dict[dataset_item_id, (result, setup_context) -> None]
        "refund-scenario-001": teardown_refund,
    },
    after_each=teardown_each,              # (result, setup_context) -> None
    after_all=teardown_environment,        # (results, shared_context) -> None
)

All hooks can be sync or async. Set a .description property on each hook function so the Netra UI can display it (truncated to 200 characters):

def before_all():
    ...
before_all.description = "Create a test employee before any scenario runs."

Example

from netra import Netra
from netra.simulation import BaseTask, SimulationHooks, TaskResult

def before_all():
    employee = api.create_employee(name="Test User")
    return {"employee_id": employee.id}
before_all.description = "Create a test employee before any scenario runs."

def before_each(shared_context):
    token = api.login(employee_id=shared_context["employee_id"])
    return {"auth_token": token}
before_each.description = "Obtain a fresh auth token before every scenario."

def setup_refund(shared_context):
    account = api.create_refund_account()
    return {"refund_account_id": account.id}
setup_refund.description = "Create a refund account for the refund scenario only."

def teardown_refund(result, setup_context):
    api.delete_refund_account(setup_context.get("refund_account_id"))
teardown_refund.description = "Delete the refund account after the refund scenario."

def after_each(result, setup_context):
    api.logout(token=setup_context.get("auth_token"))
after_each.description = "Log out after every scenario."

def after_all(results, shared_context):
    api.delete_employee(shared_context["employee_id"])
after_all.description = "Delete the test employee once all scenarios finish."

hooks = SimulationHooks(
    before_all=before_all,
    before_each=before_each,
    before={"refund-scenario-001": setup_refund},
    after={"refund-scenario-001": teardown_refund},
    after_each=after_each,
    after_all=after_all,
)

class MyAgentTask(BaseTask):
    def run(self, message, session_id=None, files=None, setup_context=None):
        ctx = setup_context or {}
        response = my_agent.chat(message, session_id=session_id, auth_token=ctx.get("auth_token"))
        return TaskResult(message=response.text, session_id=session_id or "default")

result = Netra.simulation.run_simulation(
    name="Multi-Scenario Workflow Test",
    dataset_id="your-dataset-id",
    task=MyAgentTask(),
    hooks=hooks,
)

Keys in before / after must be real dataset_item_id values from your dataset. A prescript_failed item is terminal (eval suppressed); other scenarios continue. A postscript_failed item is also terminal, but eval is not suppressed — conversation evaluations remain valid. Wrap teardown in try/except so cleanup errors do not mark otherwise-successful items postscript_failed.

File Handling

Dataset items can include file attachments. When present, the simulation framework automatically detects 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:

Field	Type	Description
`file_name`	`str`	Original file name
`content_type`	`str`	MIME type of the file content
`description`	`str?`	Optional description of the file
`data`	`str`	Base64-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()

Complete Example with Hooks

from netra import Netra
from netra.simulation import BaseTask, SimulationHooks, TaskResult
from openai import OpenAI

Netra.init(app_name="simulation-with-hooks")
client = OpenAI()

conversations: dict[str, list] = {}

# --- Hook functions ---

def before_all():
    """Create a test employee for the simulation."""
    employee = create_test_employee(role="support-agent")
    return {"employee_id": employee.id, "api_key": employee.api_key}

def before_each(shared_context):
    """Obtain a fresh auth token before every scenario."""
    token = login_employee(shared_context["employee_id"])
    return {"auth_token": token}

def setup_refund_scenario(shared_context):
    """Provision a refund account for the refund test scenario."""
    account = create_refund_account(shared_context["auth_token"])
    return {"refund_account": account.id}

def teardown_refund_scenario(result, setup_context):
    """Clean up refund account regardless of test outcome."""
    if setup_context and setup_context.get("refund_account"):
        delete_refund_account(setup_context["refund_account"])

def after_each(result, setup_context):
    """Log out after every scenario."""
    if setup_context and setup_context.get("auth_token"):
        logout_employee(setup_context["auth_token"])

def after_all(results, shared_context):
    """Delete the test employee after all scenarios complete."""
    delete_test_employee(shared_context["employee_id"])

# --- Task ---

class SupportAgentTask(BaseTask):
    def run(
        self,
        message: str,
        session_id: str | None = None,
        files=None,
        setup_context: dict | None = None,
    ) -> TaskResult:
        import uuid
        session = session_id or str(uuid.uuid4())
        ctx = setup_context or {}

        # Use setup_context from hooks
        system_prompt = "You are a customer support agent."
        if ctx.get("employee_id"):
            system_prompt += f" Employee ID: {ctx['employee_id']}"

        if session not in conversations:
            conversations[session] = [{"role": "system", "content": system_prompt}]

        conversations[session].append({"role": "user", "content": message})

        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=conversations[session],
        )
        content = response.choices[0].message.content
        conversations[session].append({"role": "assistant", "content": content})

        return TaskResult(message=content, session_id=session)

# --- Run simulation with hooks ---

hooks = SimulationHooks(
    before_all=before_all,
    before_each=before_each,
    before={"refund-scenario-item-id": setup_refund_scenario},
    after={"refund-scenario-item-id": teardown_refund_scenario},
    after_each=after_each,
    after_all=after_all,
)

result = Netra.simulation.run_simulation(
    name="Support Agent with Hooks",
    dataset_id="your-dataset-id",
    task=SupportAgentTask(),
    hooks=hooks,
    max_concurrency=3,
)

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

Netra.shutdown()

Next Steps

Dashboard Query - Query dashboard metrics
Usage Utilities - Query traces and spans
Simulation Overview - Learn about simulation testing
Evaluation - Evaluate AI outputs

Python SDK

TypeScript SDK

Getting Started

run_simulation

Parameters

Response

BaseTask

Implementation Requirements

Hooks (Pre/Post Scripts)

Hook points

Context passing

SimulationHooks

Example

File Handling

ProcessedFile

Complete Example

Complete Example with Hooks

Next Steps

​Getting Started

​run_simulation

​Parameters

​Response

​BaseTask

​Implementation Requirements

​Hooks (Pre/Post Scripts)

​Hook points

​Context passing

​SimulationHooks

​Example

​File Handling

​ProcessedFile

​Complete Example

​Complete Example with Hooks

​Next Steps

Getting Started

run_simulation

Parameters

Response

BaseTask

Implementation Requirements

Hooks (Pre/Post Scripts)

Hook points

Context passing

SimulationHooks

Example

File Handling

ProcessedFile

Complete Example

Complete Example with Hooks

Next Steps