simplai-sdk 0.3.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

Install Dependencies

pip install -r requirements.txt

Quick Start

Using the SimplAI Class

from simplai 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 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 (get trace tree)

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

from simplai 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)

Runs (list runs)

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

from simplai 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 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 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 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 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")

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

Using Core Functions Directly

from simplai 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

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
│   ├── 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]