simplai-sdk 0.4.0

SimplAI Python SDK for agents, workflows, and traces

SimplAI Python SDK

Python SDK for interacting with SimplAI's agents, workflows, and traces.

Installation

Install as a Package (Recommended)

Install the SDK in editable mode for development:

pip install simplai-sdk

Install Dependencies

pip install -r requirements.txt

Quick Start

Using the SimplAI Class

from simplai_sdk import SimplAI
import asyncio


async def main():
    # Initialize the SDK
    # - API key can be provided directly or loaded from .env (API_KEY)
    # - Base URL is read from SIMPAI_BASE_URL in your environment/.env
    simplai = SimplAI()  # or SimplAI(api_key="your-api-key")

    # Agent Chat (non-streaming)
    result = await simplai.agent_chat(
        agent_id="agent-123",
        message="Hello!",
    )

    print(f"Response: {result.response}")
    print(f"Trace ID: {result.trace_id}")
    print(f"Node ID: {result.node_id}")


asyncio.run(main())

Agent Chat (streaming)

Stream responses and optionally use on_chunk to handle each chunk. The first chunk may contain trace_id and node_id (in a role: "trace" message); the final AgentResult also has trace_id and node_id.

from simplai_sdk import SimplAI
import asyncio
import json


async def main():
    simplai = SimplAI()

    def on_chunk(chunk):
        # chunk.content may be JSON; chunk.trace_id, chunk.node_id set from first trace chunk
        raw = chunk.content
        for part in raw.split("<-------->"):
            part = part.strip()
            if not part:
                continue
            try:
                data = json.loads(part)
                if isinstance(data, dict) and "content" in data:
                    print(data["content"], end="", flush=True)
                else:
                    print(part, end="", flush=True)
            except (json.JSONDecodeError, TypeError):
                print(part, end="", flush=True)

    result = await simplai.agent_chat_stream(
        agent_id="agent-123",
        message="Hello",
        on_chunk=on_chunk,
    )
    print()
    print("Done. trace_id:", result.trace_id, "node_id:", result.node_id)


asyncio.run(main())

Traces

1. Get Trace Tree

Fetch a trace sub-tree by tree_id (same as trace_id) and node_id. This is synchronous.

from simplai_sdk import SimplAI

simplai = SimplAI()

tree = simplai.get_trace_tree(
    tree_id="your-tree-id",   # same as trace_id from agent_chat / agent_chat_stream
    node_id="your-node-id",
)
print(tree)

# Raw API response instead of parsed TraceNode
raw = simplai.get_trace_tree(tree_id="...", node_id="...", raw_response=True)

2. List Runs

Use list_runs() to discover runs for a workspace/project, then inspect a selected run with the trace tree API.

from simplai_sdk import SimplAI
from dataclasses import asdict
import json

simplai = SimplAI()

runs = simplai.list_runs(
    start_date="2026-05-06T00:00:00",  # required: ISO 8601 date/datetime
    end_date="2026-05-18T23:59:59",    # required: ISO 8601 date/datetime
    run_type="all",                    # optional: "runs", "voice_runs", or "all"
    # application_type="tools",        # optional: "agent" or "tools"
    # status="failed",                 # optional: "completed", "in_progress", or "failed"
    # source="sdk",                    # optional: "dashboard", "api", "sdk", or "webhook"
    # user="4",                        # optional: workspace user ID
    # end_user_name="Demo User",       # optional: end-user name
    # end_user_id="demo-user-1",       # optional: end-user ID
    # bulk_run_id="bulk-run-id",       # optional: bulk run ID
    # name="agent-or-tool-name",       # optional: partial agent/tool name search
    page=1,                            # optional: page number
    page_size=2,                       # optional: results per page, max 100
    # raw_response=True,               # optional: return backend response unchanged
)

if runs.data:
    first_run = runs.data[0]
    tree = simplai.get_trace_tree(
        tree_id=first_run.trace_id,
        node_id=first_run.node_id,
    )

    print(f"First Run: \n{json.dumps(asdict(first_run), indent=4, default=str)}\n\n")
    print(f"Tree:\n{json.dumps(asdict(tree), indent=4, default=str)}")
else:
    print("No runs found")

list_runs() returns a normalized RunListResponse:

Field	Type	Notes
data	list[RunSummary]	Normalized run summaries
total_count	int	Total matching runs
page	int	1-based page number
page_size	int	Results per page, max 100
total_pages	int	Total pages

Each RunSummary includes run_id, trace_id, node_id, run_type, name, version, application_type, status, source, user, end_user_name, end_user_id, bulk_run_id, input, output, latency, started_at, completed_at, and payload. latency is normalized to seconds. started_at and completed_at are ISO 8601 strings. Voice runs may also include call_duration_seconds and voice_metadata.

Common queries:

# Failed runs in a date range
failed_runs = simplai.list_runs(
    start_date="2026-04-21",
    end_date="2026-04-28",
    status="failed",
)

# Voice runs only
voice_runs = simplai.list_runs(
    start_date="2026-04-21",
    end_date="2026-04-28",
    run_type="voice_runs",
)

# Paginate through a large result set
page = 1
while True:
    runs = simplai.list_runs(
        start_date="2026-04-21",
        end_date="2026-04-28",
        page=page,
        page_size=100,
    )
    for run in runs.data:
        print(run.run_id, run.status, run.latency)

    if page >= runs.total_pages:
        break
    page += 1

Pass raw_response=True to receive the backend response unchanged instead of the normalized SDK response.

Tool / Workflow Execution

The SDK provides several functions for executing workflows and managing their execution lifecycle.

1. Execute Workflow (Async - Returns Immediately)

Execute a workflow and get the execution ID immediately without waiting for completion.

from simplai_sdk import SimplAI
import asyncio


async def main():
    simplai = SimplAI()  # API key + SIMPAI_BASE_URL loaded from .env

    # Execute a workflow (returns immediately with execution ID)
    execution_info = await simplai.execute_workflow(
        workflow_id="your-workflow-id",
        inputs={"number_1": 1, "number_2": 2},  # Key is app input variable name, value is the input value
    )
    print(f"Execution ID: {execution_info.get('execution_id')}")
    print(f"Status: {execution_info.get('status')}")


asyncio.run(main())

Returns:

• execution_id (str): The execution ID to track this workflow run
• status (str): Initial status of the execution

2. Get Tool Result

Get the execution result and status by execution ID. Use this to check the status of a previously started execution.

from simplai_sdk import SimplAI
import asyncio


async def main():
    simplai = SimplAI()

    # Get execution result by execution ID
    result = await simplai.get_tool_result(
        execution_id="your-execution-id",
    )
    print(f"Execution ID: {result.get('executionId')}")
    print(f"Status: {result.get('status')}")
    print(f"Trace ID: {result.get('traceId')}")  # Optional
    print(f"API Response: {result.get('api_response')}")  # Optional, contains parsed result


asyncio.run(main())

Returns:

• executionId (str): The execution ID
• status (str): Current status (e.g., "PENDING", "COMPLETED", "FAILED")
• traceId (str, optional): Trace ID for this execution
• api_response (Any, optional): Parsed JSON result from the workflow execution

3. Execute and Wait for Workflow

Execute a workflow and automatically poll until completion. This is a convenience function that combines execute_workflow and get_tool_result with polling logic.

from simplai_sdk import SimplAI
import asyncio


async def main():
    simplai = SimplAI()

    # Execute a workflow and wait for completion
    result = await simplai.execute_and_wait_workflow(
        workflow_id="your-workflow-id",
        inputs={"number_1": 1, "number_2": 2},  # Key is app input variable name, value is the input value
        timeout=60.0,  # Maximum time to wait in seconds (default: 60.0)
        poll_interval=2.0,  # Time between status checks in seconds (default: 2.0)
    )
    print(f"Execution ID: {result.get('executionId')}")
    print(f"Status: {result.get('status')}")
    print(f"Trace ID: {result.get('traceId')}")  # Optional
    print(f"API Response: {result.get('api_response')}")  # Optional, contains parsed result


asyncio.run(main())

Parameters:

• workflow_id (str): The workflow ID to execute
• inputs (dict): Input parameters for the workflow (key = app input variable name, value = input value)
• timeout (float, optional): Maximum time to wait in seconds (default: 60.0)
• poll_interval (float, optional): Time between status checks in seconds (default: 2.0)

Returns:

• executionId (str): The execution ID
• status (str): Final status (e.g., "COMPLETED", "FAILED", "CANCELLED")
• traceId (str, optional): Trace ID for this execution
• api_response (Any, optional): Parsed JSON result from the workflow execution

Raises:

• TimeoutException: If the execution doesn't complete within the specified timeout

4. Cancel Execution

Cancel a running or pending workflow execution.

from simplai_sdk import SimplAI
import asyncio


async def main():
    simplai = SimplAI()

    # Cancel an execution
    cancel_result = await simplai.cancel_execution(
        execution_id="your-execution-id",
    )
    print(f"Execution ID: {cancel_result.get('execution_id')}")
    print(f"Status: {cancel_result.get('status')}")


asyncio.run(main())

Parameters:

• execution_id (str): The execution ID to cancel

Returns:

• execution_id (str): The execution ID that was cancelled
• status (str): Status after cancellation (typically "CANCELLED")

Workflow Authoring (Advanced Tools)

Build a workflow DAG in code and create it via the API. Workflows authored this way are of type ADVANCED_TOOL. Steps are added using shipped, typed step classes (simplai.steps): each class's constructor encodes that step's catalog inputs, so your IDE autocompletes exactly what to configure.

import asyncio
from simplai_sdk import SimplAI
from simplai.steps import PythonCode  # generated, shipped per step type

async def main():
    simplai = SimplAI()  # api_key from arg/env/.env

    # Discover what a step needs (prints inputs/outputs from the live catalog):
    await simplai.describe_step("api_request")

    # `new_workflow` injects the cached step catalog so inputs/outputs are validated as you build.
    wf = await simplai.new_workflow(title="Square", description="Squares the input")
    wf.set_inputs({"x": "number"})
    wf.add_step("square", PythonCode(code="return {'result': inputs['x'] ** 2}"))
    wf.add_edge(wf.START, "square")                       # START marks the initial step
    wf.set_outputs({"result": wf.ref_step("square", "result")})

    # Create then run in one call (reuses the existing workflow executor):
    result = await simplai.create_and_run_workflow(wf, inputs={"x": 5}, wait=True)
    print(result)

asyncio.run(main())

DAG building blocks

• WorkflowBuilder / simplai.new_workflow(title, description) — start a DAG (type ADVANCED_TOOL).
• add_step(step_id, StepClass(...)) — add a node from a typed step class (simplai.steps).
• add_edge(source, target) — connect steps; use START as source for initial steps.
• set_inputs({...}) / set_outputs({...}) — define workflow inputs and the output mapping.
• wf.ref_input("name") → {{workflow.input.name}}; wf.ref_step("id", "field") → {{steps.id.output.field}} — wire data between inputs/steps/outputs.
• build() validates the DAG (orphan steps, missing edges, START present, output refs) and returns the create payload; create_workflow / create_and_run_workflow send it.

Discovery helpers (for picking step types and resource ids)

await simplai.list_step_types(search="api")          # available step types
await simplai.describe_step("api_request")            # a step's input/output spec
await simplai.list_tools()                            # tools usable as tool_id
await simplai.list_agents(agent_type="VOICE")         # agents usable as agent_id/voice_agent_id

CRUD

created = await simplai.create_workflow(wf)
fetched = await simplai.get_workflow(created.id)
page    = await simplai.list_workflows(type="ADVANCED_TOOL")
await simplai.update_workflow(created.id, wf)
await simplai.delete_workflow(created.id)

Regenerating the typed step classes (maintainers). The classes under simplai.steps are generated from the step catalog and committed. Refresh them when the catalog changes:

simplai-gen-steps --from-api          # uses SIMPAI_BASE_URL + SIMPAI_API_KEY
simplai-gen-steps --seed path/to/simplai_steps.json

Bulk Runs

The SDK provides functions for executing workflows in bulk using CSV files. Bulk runs process multiple records from a file in batches.

1. Trigger Bulk Run

Trigger a bulk run to process multiple records from a CSV file. The file should be accessible via a URL (e.g., S3, HTTP, etc.).

from simplai_sdk import SimplAI
import asyncio


async def main():
    simplai = SimplAI()

    # Trigger a bulk run
    bulk_run = await simplai.trigger_bulk_run(
        workflow_id="your-workflow-id",
        file_link="s3://your-bucket/your-file.csv",  # URL to the CSV file
        batch_size=100,  # Number of records to process per batch
        input_mapping=[
            # Map workflow input fields to CSV column names
            # app_input_field: The workflow input variable name from the platform
            # file_input_field: The CSV column name (e.g., "A", "B", or column header)
            {"app_input_field": "number_1", "file_input_field": "A"},
            {"app_input_field": "number_2", "file_input_field": "B"},
        ],
    )
    bulk_run_id = bulk_run["bulk_run_id"]
    print(f"Bulk Run ID: {bulk_run_id}")
    print(f"Status: {bulk_run['status']}")


asyncio.run(main())

Parameters:

• workflow_id (str): The workflow ID to execute
• file_link (str): URL/link to the input CSV file (e.g., S3 URL, HTTP URL)
• batch_size (int): Number of records to process per batch
• input_mapping (list): List of dictionaries mapping workflow input fields to CSV columns
  • app_input_field (str): The workflow input variable name from the platform
  • file_input_field (str): The CSV column name (e.g., "A", "B", or column header name)

Returns:

• bulk_run_id (str): The bulk run ID to track this bulk execution
• status (str): Initial status of the bulk run

2. Get Bulk Run Status

Get the current status and progress of a bulk run by bulk run ID.

from simplai_sdk import SimplAI
import asyncio


async def main():
    simplai = SimplAI()

    # Get bulk run status
    status = await simplai.get_bulk_run_status(
        bulk_run_id="your-bulk-run-id",
    )
    print(f"Bulk Run ID: {status.get('bulk_run_id')}")
    print(f"Status: {status.get('status')}")
    print(f"Batches Completed: {status.get('batch_completed')}")
    print(f"Total Batches: {status.get('total_batches')}")
    print(f"Trace ID: {status.get('trace_id')}")  # Optional


asyncio.run(main())

Parameters:

• bulk_run_id (str): The bulk run ID to check

Returns:

• bulk_run_id (str): The bulk run ID
• status (str): Current status (e.g., "PENDING", "RUNNING", "COMPLETED", "FAILED")
• batch_completed (int): Number of batches that have been completed
• total_batches (int): Total number of batches to process
• trace_id (str, optional): Trace ID for this bulk run

3. Cancel Bulk Run

Cancel a running or pending bulk run.

from simplai_sdk import SimplAI
import asyncio


async def main():
    simplai = SimplAI()

    # Cancel a bulk run
    cancel_result = await simplai.cancel_bulk_run(
        bulk_run_id="your-bulk-run-id",
    )
    print(f"Bulk Run ID: {cancel_result.get('bulk_run_id')}")
    print(f"Status: {cancel_result.get('status')}")
    print(f"Trace ID: {cancel_result.get('trace_id')}")  # Optional


asyncio.run(main())

Parameters:

• bulk_run_id (str): The bulk run ID to cancel

Returns:

• bulk_run_id (str): The bulk run ID that was cancelled
• status (str): Status after cancellation (typically "CANCELLED")
• trace_id (str, optional): Trace ID for this bulk run

4. Download Bulk Run Result

Download the results of a completed bulk run as CSV text.

from simplai_sdk import SimplAI
import asyncio


async def main():
    simplai = SimplAI()

    # Download bulk run result as CSV
    csv_text = await simplai.download_bulk_run_result(
        bulk_run_id="your-bulk-run-id",
    )
    print(f"CSV Result (first 500 chars): {csv_text[:500]}")
    
    # Optionally save to file
    with open("bulk_result.csv", "w") as f:
        f.write(csv_text)


asyncio.run(main())

Parameters:

• bulk_run_id (str): The bulk run ID to download results for

Returns:

• str: CSV text containing the bulk run results

Note: The bulk run must be completed before downloading results. Use get_bulk_run_status() to check if the bulk run has finished.

Scheduled Runs

The SDK provides functions for scheduling workflows to run automatically at specified times using cron expressions. Scheduled runs support two types: BULK (processes data from a file) and MANUAL (uses static input values).

1. Schedule Run

Schedule a workflow to run automatically based on a cron expression. Supports both BULK and MANUAL types.

BULK Type Scheduled Run:

Processes data from a CSV file on a schedule. Requires a file URL and batch size.

from simplai_sdk import SimplAI
import asyncio


async def main():
    simplai = SimplAI()

    # Schedule a BULK type run
    schedule = await simplai.schedule_run(
        workflow_id="your-workflow-id",
        scheduled_run_type="BULK",
        cron_expression="0 */3 * ? * *",  # Every 3 hours
        time_zone="Asia/Kolkata",  # IANA Time Zone identifier
        input_mapping=[
            # Map workflow input fields to CSV column names
            # app_input_field: The workflow input variable name from the platform
            # file_input_field: The CSV column name (e.g., "A", "B", or column header)
            {"app_input_field": "number_1", "file_input_field": "A"},
            {"app_input_field": "number_2", "file_input_field": "B"},
        ],
        file_url="s3://your-bucket/your-file.csv",  # Required for BULK type
        batch_size=100,  # Required for BULK type
    )
    print(f"Unique ID: {schedule.get('unique_id')}")
    print(f"Job Master ID: {schedule.get('job_master_id')}")
    print(f"Status: {schedule.get('status')}")
    print(f"Trace ID: {schedule.get('trace_id')}")  # Optional


asyncio.run(main())

MANUAL Type Scheduled Run:

Runs a workflow with static input values on a schedule. No file URL required.

from simplai_sdk import SimplAI
import asyncio


async def main():
    simplai = SimplAI()

    # Schedule a MANUAL type run
    schedule = await simplai.schedule_run(
        workflow_id="your-workflow-id",
        scheduled_run_type="MANUAL",
        cron_expression="0 0 12 * * ?",  # Daily at noon
        time_zone="UTC",
        input_mapping=[
            # For MANUAL type, use "value" instead of "file_input_field"
            # app_input_field: The workflow input variable name from the platform
            # value: The static value to use for this input
            {"app_input_field": "interview_id", "value": "0148eee6-a867-4c7a-8d42-f039a5ef6adf"},
            {"app_input_field": "number_1", "value": 1},
            {"app_input_field": "number_2", "value": 2},
        ],
        # file_url is not required for MANUAL type (ignored if provided)
        batch_size=1,  # Optional for MANUAL type
    )
    print(f"Unique ID: {schedule.get('unique_id')}")
    print(f"Job Master ID: {schedule.get('job_master_id')}")
    print(f"Status: {schedule.get('status')}")


asyncio.run(main())

Parameters:

• workflow_id (str): The workflow ID to execute
• scheduled_run_type (str): Type of scheduled run - "BULK" or "MANUAL"
• cron_expression (str): Quartz-compatible cron expression for scheduling (e.g., "0 */3 * ? * *" for every 3 hours)
  • Format: "second minute hour day-of-month month day-of-week"
  • Uses Quartz cron syntax (supports ? for day-of-month or day-of-week)
  • Example: "0 0 12 * * ?" = Daily at 12:00 PM
  • Example: "0 */3 * ? * *" = Every 3 hours
• time_zone (str): IANA Time Zone identifier for the schedule (e.g., "Asia/Kolkata", "UTC", "America/New_York")
• input_mapping (list): List of dictionaries with input field mappings
  • For BULK type:
    • app_input_field (str): The workflow input variable name from the platform
    • file_input_field (str): The CSV column name (e.g., "A", "B", or column header)
  • For MANUAL type:
    • app_input_field (str): The workflow input variable name from the platform
    • value (str/int/float): The static value to use for this input
• file_url (str, optional): URL/link to the input CSV file (required for BULK, ignored for MANUAL)
• batch_size (int, optional): Number of records to process per batch (required for BULK, optional for MANUAL)

Returns:

• unique_id (str): Unique identifier for the scheduled run (used for cancellation)
• job_master_id (str): Job master identifier (used for cancellation)
• status (str): Status of the scheduled run creation
• trace_id (str, optional): Trace ID for this scheduled run

2. Cancel Scheduled Run

Cancel a scheduled run using the job_master_id and unique_id returned from schedule_run().

from simplai_sdk import SimplAI
import asyncio


async def main():
    simplai = SimplAI()

    # Cancel a scheduled run
    cancel_result = await simplai.cancel_schedule_run(
        job_master_id=1238,  # Can be int or str
        unique_id="4d5ec902-39d0-45fd-93e1-ed599b576256",
    )
    print(f"Status: {cancel_result.get('status')}")
    print(f"Trace ID: {cancel_result.get('trace_id')}")  # Optional


asyncio.run(main())

Parameters:

• job_master_id (int or str): Job master identifier from the scheduled run
• unique_id (str): Unique identifier from the scheduled run

Returns:

• status (str): Status after cancellation
• trace_id (str, optional): Trace ID for this cancellation operation

Note: Both job_master_id and unique_id are returned when you create a scheduled run using schedule_run(). Save these values if you need to cancel the schedule later.

Billing

The SDK provides functions for accessing billing and usage information for workflow and agent executions.

Get Billing Detail

Get billing details (consumption, app name, app type) for a specific trace ID. This is useful for tracking costs and usage associated with a particular execution.

from simplai_sdk import SimplAI
import asyncio


async def main():
    simplai = SimplAI()

    # Get billing details for a trace ID
    billing = await simplai.get_billing_detail(
        trace_id="your-trace-id",  # Trace ID from workflow execution or agent chat
    )
    print(f"Total Consumption: {billing.get('totalConsumption')}")
    print(f"App Name: {billing.get('appName')}")
    print(f"App Type: {billing.get('appType')}")


asyncio.run(main())

Parameters:

• trace_id (str): The trace ID from a workflow execution or agent chat response

Returns:

• totalConsumption (int/float): Total consumption/cost for this trace
• appName (str): Name of the application/workflow/agent
• appType (str): Type of the application (e.g., "WORKFLOW", "AGENT")

Note: The trace ID can be obtained from:

• Workflow execution results: result.get('traceId') from execute_and_wait_workflow() or get_tool_result()
• Agent chat results: result.trace_id from agent_chat() or agent_chat_stream()
• Bulk run status: status.get('trace_id') from get_bulk_run_status()

Raises:

• APIException: If the trace ID is invalid or billing information is not available (e.g., 204 No Content)

Evaluations

The evaluations SDK supports dataset management, AI-assisted evaluator generation, reusable evaluator templates, test runs, version comparisons, and row-level results. These methods are synchronous; create_test_run() returns after the run is created, and you can use get_test_run_status() to monitor server-side execution.

All examples below use the high-level SimplAI client. It loads API_KEY and SIMPAI_BASE_URL from the environment when they are not passed explicitly.

from simplai_sdk import SimplAI

simplai = SimplAI()  # or SimplAI(api_key="your-api-key")

1. Create a Dataset

Create a dataset from inline rows or from a .csv, .json, or .jsonl file. Provide rows or file_path, but not both. JSON files must contain a list of objects, and each non-empty JSONL line must contain one object.

# Inline rows
dataset = simplai.create_dataset(
    name="credit-analyst-regression",
    description="Core credit-analysis scenarios",
    rows=[
        {
            "input": {"question": "List the outstanding loans."},
            "expected_output": "A complete list of outstanding loans.",
        },
        {
            "input": {"question": "Summarize the auditor's opinion."},
            "expected_output": "A concise summary of the auditor's opinion.",
        },
    ],
)

print(dataset.id)
print(dataset.name)
print(dataset.row_count)
print(dataset.created_at)

# File upload; also supports .json and .jsonl
uploaded_dataset = simplai.create_dataset(
    name="credit-analyst-file-dataset",
    description="Regression cases maintained in source control",
    file_path="/path/to/test_data.csv",
)

Parameters:

• name (str): Dataset name.
• description (str, optional): Dataset description.
• rows (list[dict], optional): Inline dataset rows.
• file_path (str, optional): Path to a CSV, JSON, or JSONL file.

Returns: Dataset with id, name, row_count, created_at, updated_at, and the raw response in payload when available.

2. Generate a Dataset with AI

Generate rows using application context, then save them as a reusable dataset.

dataset = simplai.generate_dataset(
    name="credit-analyst-generated",
    application_id="your-application-id",
    description="Synthetic credit-analysis regression cases",
    version_id="latest",           # Optional
    row_count=20,                   # Default: 20
    instructions="Focus on edge cases involving loan defaults.",
)

print(dataset.id, dataset.row_count)

Parameters:

• name (str): Name used when saving the generated dataset.
• application_id (str): Application whose context is used for generation.
• description (str, optional): Description saved with the dataset; it is also used as the generation prompt when instructions is omitted.
• version_id (str, optional): Application version used for generation.
• row_count (int, optional): Requested row count; defaults to 20.
• instructions (str, optional): Guidance for the generated scenarios.

Returns: Dataset.

3. List Datasets

datasets = simplai.list_datasets(page=1, page_size=25)

print(f"Found {datasets.total} datasets")
for dataset in datasets.datasets:
    print(dataset.id, dataset.name, dataset.row_count)

Parameters: page defaults to 1; page_size defaults to 25.

Returns: DatasetList with datasets, total, page, and page_size.

4. List Evaluator Templates

List saved evaluator templates that can be referenced when creating a test run.

templates = simplai.list_evaluator_templates(page=1, page_size=25)

for template in templates.templates:
    print(template.id, template.name, template.evaluator_type)

Parameters: page defaults to 1; page_size defaults to 25.

Returns: EvaluatorTemplateList with templates, total, page, and page_size. Each template includes its evaluator type, scoring configuration, prompt or Python code, and timestamps when available.

5. Generate an Evaluator with AI

generate_evaluator() returns a complete inline configuration that can be passed directly to create_test_run(). Supported evaluator types are "llm_judge" and "python".

accuracy_evaluator = simplai.generate_evaluator(
    name="Factual Accuracy",
    evaluator_type="llm_judge",
    description="Score factual consistency against the expected output.",
    input_source="dataset",
    application_id="your-application-id",
    version_id="latest",
    dataset_id="your-dataset-id",
    trace_source="message",
)

python_evaluator = simplai.generate_evaluator(
    name="Response Format",
    evaluator_type="python",
    description="Verify that the response is valid JSON.",
    input_source="dataset",
    application_id="your-application-id",
    version_id="latest",
    dataset_id="your-dataset-id",
    trace_source="message",
)

Parameters:

• name (str): Unique evaluator name within a test run.
• evaluator_type (str): "llm_judge" or "python".
• description (str): Evaluation goal used by AI autofill.
• input_source (str): Source supplied to the evaluator, such as "dataset".
• application_id (str): Application to evaluate.
• version_id (str): Application version used for context.
• dataset_id (str): Dataset used for schema and sample context.
• trace_source (str, optional): Trace field used as additional evaluator context, such as "message".

Returns: A dictionary containing the generated evaluator configuration.

6. Create a Test Run

A test run requires at least one inline evaluator or saved evaluator template. It may compare the primary version with up to five additional versions (six versions total).

test_run = simplai.create_test_run(
    name="credit-analyst-release-gate",
    application_type="agent",
    application_id="your-application-id",
    version_id="v3",
    dataset_id="your-dataset-id",
    comparison_versions=["v2", "v1"],
    evaluators=[
        {
            "name": "Logic Check",
            "description": "Checks the response for logical consistency.",
            "type": "llm_judge",
            "model": "anthropic/claude-haiku-4-5",
            "prompt": "Score the response against the dataset input and expected output.",
            "scoring_format": "numeric",
            "scoring_config": {"min": 0, "max": 1},
        },
        {
            "name": "JSON Validator",
            "description": "Checks whether the response is valid JSON.",
            "type": "python_executor",
            "code": (
                "def __main__(dataset, run_output):\n"
                "    import json\n"
                "    try:\n"
                "        json.loads(run_output)\n"
                "        return True\n"
                "    except (TypeError, json.JSONDecodeError):\n"
                "        return False"
            ),
            "scoring_format": "binary",
        },
    ],
)

print(test_run.id, test_run.status)
print(test_run.compare_test_run_ids)

Parameters:

• name (str): Test-run name.
• application_type (str): Application type, such as "agent", "workflow", or the type supported by your SimplAI deployment.
• application_id (str): Application to evaluate.
• version_id (str): Primary application version.
• dataset_id (str): Dataset used by the run.
• evaluators (list[dict], optional): Inline evaluator configurations.
• comparison_versions (list[str | dict], optional): Up to five additional version IDs. Dictionary entries must contain version_id.
• evaluator_ids (list[str], optional): IDs of saved evaluator templates. Templates are resolved and combined with inline evaluators.

Returns: TestRun with fields including id, name, status, version_id, dataset information, progress, evaluator configuration, metrics, and compare_test_run_ids.

Evaluator names must be unique within a run. The SDK accepts both concise SDK keys and backend-style keys:

Purpose	SDK key	Backend-style key
Evaluator type	type	evaluator_type
Model	model	model_id
LLM prompt	prompt	evaluation_prompt
Python code	code	python_code

Supported evaluator types are llm_judge, python_executor, and python; python_executor is normalized to python. Supported scoring formats are:

• numeric: requires scoring_config={"min": ..., "max": ...}.
• categorical: requires at least two labels, for example scoring_config={"labels": ["pass", "fail"], "best_label": "pass"}. The first label is used as the best label when one is not specified.
• binary: does not require a scoring configuration.

To create a run entirely from saved templates, use evaluator_ids:

test_run = simplai.create_test_run(
    name="template-based-run",
    application_type="agent",
    application_id="your-application-id",
    version_id="latest",
    dataset_id="your-dataset-id",
    evaluator_ids=["your-evaluator-template-id"],
)

To apply per-run overrides to a saved template, reference it from evaluators:

test_run = simplai.create_test_run(
    name="template-with-overrides",
    application_type="agent",
    application_id="your-application-id",
    version_id="latest",
    dataset_id="your-dataset-id",
    evaluators=[
        {
            "template_id": "your-evaluator-template-id",
            "name": "Release Accuracy",
            "model": "anthropic/claude-haiku-4-5",
        }
    ],
)

7. Get Test Run Status

Poll the lightweight status endpoint instead of repeatedly downloading full results.

import time

while True:
    status = simplai.get_test_run_status(test_run_id=test_run.id)
    print(status.status, status.rows_processed, status.total_rows)

    if status.status.lower() in {"completed", "failed", "cancelled"}:
        break
    time.sleep(2)

Returns: TestRunStatus with test_run_id, status, rows_processed, and total_rows.

8. List Test Runs

Discover historical runs for an application and optionally filter them.

runs = simplai.list_test_runs(
    application_id="your-application-id",
    application_type="agent",
    status="completed",              # Optional
    dataset_id="your-dataset-id",    # Optional
    version_id="v3",                 # Optional
    test_run_name="release-gate",    # Optional
    page=1,
    page_size=25,
)

print(runs.total, runs.total_pages)
for run in runs.test_runs:
    print(run.id, run.name, run.version_id, run.status)

Returns: TestRunList with test_runs, total, page, page_size, and total_pages.

9. Clone a Test Run

Clone an existing run's application, dataset, and evaluator configuration, and trigger the new run immediately. Any omitted override is copied from the source run; the default name is <source_name>_copy.

cloned_run = simplai.clone_test_run(
    test_run_id="source-test-run-id",
    test_run_name="credit-analyst-v4",
    version_id="v4",
    comparison_versions=["v3", "v2"],
    dataset_id="replacement-dataset-id",
)

print(cloned_run.id, cloned_run.status)

Parameters:

• test_run_id (str): Source test-run ID.
• test_run_name (str, optional): New run name; defaults to <source_name>_copy.
• version_id (str, optional): Replacement primary version.
• comparison_versions (list[str | dict], optional): Replacement comparison versions. When omitted, the source run's comparison versions are copied.
• dataset_id (str, optional): Replacement dataset.

Returns: The newly created TestRun.

10. Add and Remove Comparisons

Add a compatible existing run to an anchor run's comparison set, or remove it. Compatibility and comparison limits are enforced by the service.

updated_run = simplai.add_comparison(
    test_run_id="anchor-test-run-id",
    comparison_test_run_id="comparison-test-run-id",
)
print(updated_run.compare_test_run_ids)

updated_run = simplai.remove_comparison(
    test_run_id="anchor-test-run-id",
    comparison_test_run_id="comparison-test-run-id",
)
print(updated_run.compare_test_run_ids)

Both methods require the anchor test_run_id and the existing comparison_test_run_id to add or remove.

Both methods return the updated TestRun.

11. Get Test Run Results

Fetch run metadata, aggregate evaluator scores, row-level outputs, and optional comparison aggregates.

results = simplai.get_test_run_results(
    test_run_id="your-test-run-id",
    include_comparison=True,  # Default
)

print(results.test_run_name, results.status)
for summary in results.evaluator_summaries:
    print(summary.evaluator_name, summary.aggregate_value)

for row in results.rows:
    print(row.input, row.output, row.status, row.latency)
    for evaluation in row.evaluator_results:
        print(
            evaluation.evaluator_name,
            evaluation.score_or_label,
            evaluation.explanation,
        )

if results.compared_versions:
    for version in results.compared_versions:
        print(version.version_id, version.version_name)

Parameters:

• test_run_id (str): Test run to inspect.
• include_comparison (bool, optional): Include stored comparison aggregates when available; defaults to True.
• comparison_versions (list[str], optional): Reserved for an explicit comparison override. The current results API does not support this override; leave it as None and use add_comparison() or remove_comparison() to change the stored comparison set.

Returns: TestRunResults with run and dataset metadata, evaluator_summaries, row-level rows, and optional compared_versions.

End-to-End Evaluation Workflow

The following example generates a dataset and evaluator, starts a comparison run, waits for completion, and fetches the results:

import time

from simplai_sdk import SimplAI

simplai = SimplAI()

dataset = simplai.generate_dataset(
    name="nightly-agent-regression",
    application_id="your-application-id",
    version_id="latest",
    row_count=20,
    instructions="Cover normal cases, boundary cases, and malformed inputs.",
)

evaluator = simplai.generate_evaluator(
    name="Answer Quality",
    evaluator_type="llm_judge",
    description="Score correctness and completeness against expected output.",
    input_source="dataset",
    application_id="your-application-id",
    version_id="latest",
    dataset_id=dataset.id,
    trace_source="message",
)

test_run = simplai.create_test_run(
    name="nightly-release-gate",
    application_type="agent",
    application_id="your-application-id",
    version_id="latest",
    dataset_id=dataset.id,
    evaluators=[evaluator],
    comparison_versions=["previous-version-id"],
)

while True:
    status = simplai.get_test_run_status(test_run_id=test_run.id)
    if status.status.lower() in {"completed", "failed", "cancelled"}:
        break
    time.sleep(2)

if status.status.lower() != "completed":
    raise RuntimeError(f"Evaluation finished with status: {status.status}")

results = simplai.get_test_run_results(
    test_run_id=test_run.id,
    include_comparison=True,
)

for summary in results.evaluator_summaries:
    print(summary.evaluator_name, summary.aggregate_value)

Evaluation validation errors, such as duplicate evaluator names or invalid scoring configurations, raise ValueError. API, authentication, file parsing, and response errors raise evaluations.EvaluationsError.

Using Core Functions Directly

from simplai_sdk import SimplAI
import asyncio


async def main():
    # API key is optional - will be loaded from .env if not provided
    simplai = SimplAI()

    result = await simplai.agent_chat(
        agent_id="agent-123",
        message="Hello!",
    )

    print(f"Trace ID: {result.trace_id}")
    print(f"Node ID: {result.node_id}")


asyncio.run(main())

Features

• Agent Chat: Non-streaming and streaming agent conversations
• Workflows: Execute and manage workflows
• Bulk Runs: Trigger and monitor bulk workflow executions
• Traces: Fetch and analyze execution traces
• Billing: Access usage and cost information
• Evaluations: Manage datasets, evaluators, test runs, results, and version comparisons

Environment Setup

See ENV_SETUP.md for detailed environment variable configuration.

At a minimum you should set the following in your .env or environment:

• API_KEY – your SimplAI API key (used by SimplAI and core functions)
• SIMPAI_BASE_URL – base URL for the SimplAI edge service

The SDK automatically loads .env and reads SIMPAI_BASE_URL in src/constants/__init__.py, and all HTTP clients derive their base URLs from this value.

Testing

Run the test scripts from the project root (with the SDK installed, e.g. pip install -e .):

# Non-streaming agent chat (agent_chat)
python tests/test_agent_chat.py

# Streaming agent chat (agent_chat_stream with on_chunk)
python tests/test_agent_chat_stream.py

# Trace tree (get_trace_tree)
python tests/test_traces.py

Test script overview

Script	Function	Description
tests/test_agent_chat.py	simplai.agent_chat(agent_id, message)	Non-streaming agent chat; returns AgentResult with response, trace_id, node_id, payload.
tests/test_agent_chat_stream.py	simplai.agent_chat_stream(agent_id, message, on_chunk=...)	Streaming agent chat; on_chunk receives each chunk; first chunk may contain trace_id/node_id; final result has trace_id, node_id.
tests/test_traces.py	simplai.get_trace_tree(tree_id, node_id)	Fetches trace sub-tree; use tree_id (same as trace_id from agent calls) and node_id. Use raw_response=True to get raw API JSON.

Package Structure

python-sdk/
├── src/            # Source code (installed as package)
│   ├── core/       # Core SDK functionality
│   │   ├── agents/     # Agent chat and execution
│   │   └── workflows/  # Workflow execution and management
│   ├── simplai/    # High-level SimplAI class
│   ├── traces/     # Trace fetching and analysis
│   ├── billing/    # Billing and usage
│   ├── evaluations/ # Datasets, evaluators, test runs, and comparisons
│   ├── utils/      # Utility functions
│   └── exceptions/ # Custom exceptions
├── tests/         # Test scripts
├── pyproject.toml   # Package configuration
└── README.md       # This file

Requirements

• Python >= 3.9
• httpx >= 0.25.0
• pydantic >= 2.0.0

Documentation

• Environment Setup Guide

License

This SDK is proprietary software. See LICENSE for details.

Copyright (c) 2024 SimplAI. All rights reserved.

For licensing inquiries, please contact: [e-mail]