> ## Documentation Index
> Fetch the complete documentation index at: https://docs.getnetra.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Simulation

> Run multi-turn conversation simulations using the Netra Simulation API

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.

```python theme={null}
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.

<CodeGroup>
  ```python Usage theme={null}
  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'])}")
  ```

  ```python Signature theme={null}
  run_simulation(
      name: str,
      dataset_id: str,
      task: BaseTask,
      context: Optional[dict[str, Any]] = None,
      max_concurrency: int = 5,
  ) -> Optional[dict[str, Any]]
  ```
</CodeGroup>

### 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)         |

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

<AccordionGroup>
  <Accordion title="Completed Item" icon="circle-check">
    | 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 |
  </Accordion>

  <Accordion title="Failed Item" icon="circle-xmark">
    | 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 |
  </Accordion>
</AccordionGroup>

***

## BaseTask

Create a custom task by inheriting from the `BaseTask` abstract base class. Your implementation wraps your AI agent and handles the conversation flow.

<CodeGroup>
  ```python Definition theme={null}
  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
  ```

  ```python TaskResult Dataclass theme={null}
  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
  ```
</CodeGroup>

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

***

## 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:

| 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

```python theme={null}
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

* [Dashboard Query](/sdk-reference/dashboard-query/python) - Query dashboard metrics
* [Usage Utilities](/usage/usage-utilities) - Query traces and spans
* [Simulation Overview](/Simulation/Simulation-overview) - Learn about simulation testing
* [Evaluation](/sdk-reference/evaluation/python) - Evaluate AI outputs