sim/packages/python-sdk/README.md

# Sim Python SDK

The official Python SDK for [Sim](https://sim.ai), allowing you to execute workflows programmatically from your Python applications.

## Server compatibility

`0.2.x` talks to the v2 API and has no fallback to the older endpoints, so it requires a Sim deployment that serves `POST /api/v2/workflows/{id}/execute`. That surface is newer than the endpoints `0.1.x` used, and a deployment can also have it switched off — a self-hosted build serves `/api/v2` only when the operator enables `V2_API`. Where it is unavailable every v2 route answers 404, so `execute_workflow` raises `SimStudioError('HTTP 404: Not Found')` — enable or upgrade the v2 API on the server, or pin `simstudio-sdk<0.2`, which keeps using `/api/workflows/{id}/execute` and `/api/jobs/{id}`.

## Upgrading from 0.1.x to 0.2.0

`0.2.0` is a breaking release.

- **Requests move to `/api/v2`.** `execute_workflow` posts to `/api/v2/workflows/{workflow_id}/execute`, sends the workflow input nested under `input`, and carries `async` / `executionTimeoutSeconds` in the body instead of the `X-Execution-Mode` and `X-Execution-Timeout-Seconds` headers.
- **`AsyncExecutionResult.job_id` is now `run_id`,** and `execution_id` has been removed from that dataclass. Replace `result.job_id` with `result.run_id`.
- **`get_job_status(job_id)` is legacy.** It still calls `/api/jobs/{job_id}` and only resolves IDs from a `0.1.x` async execution. For runs started by `0.2.x`, use `get_workflow_run(workflow_id, run_id)`, which reads `/api/v2/workflows/{workflow_id}/runs/{run_id}`.
- **`WorkflowExecutionResult.success` is derived from the run status** rather than read from the response body, and is `True` only for `completed` and `paused` runs — so a run cancelled while it was in flight now reports `success=False`, as it did before the v2 migration. The new `WorkflowExecutionResult.status` field carries the server's terminal status (`'completed'`, `'failed'`, `'paused'` or `'cancelled'`), which is how you tell a cancelled run from a failed one.
- **`metadata` is now built by the SDK,** with the keys `duration`, `runId`, `startTime` and `endTime`. The v2 response carries no execution logs or trace spans, so `logs` and `trace_spans` are always `None`; the pre-v2 `metadata['executionId']` is now `metadata['runId']`.

Note one deliberate difference from the TypeScript SDK: a failed synchronous run *throws* there, but here it returns normally with `error` set and `status='failed'`.

## Installation

```bash
pip install simstudio-sdk
```

## Quick Start

```python
import os
from simstudio import SimStudioClient

# Initialize the client
client = SimStudioClient(
    api_key=os.getenv("SIM_API_KEY", "your-api-key-here"),
    base_url="https://sim.ai"  # optional, defaults to https://sim.ai
)

# Execute a workflow
try:
    result = client.execute_workflow("workflow-id")
    print("Workflow executed successfully:", result)
except Exception as error:
    print("Workflow execution failed:", error)
```

## API Reference

### SimStudioClient

#### Constructor

```python
SimStudioClient(api_key: str, base_url: str = "https://sim.ai")
```

- `api_key` (str): Your Sim API key
- `base_url` (str, optional): Base URL for the Sim API (defaults to `https://sim.ai`)

#### Methods

##### execute_workflow(workflow_id, input=None, *, timeout=30.0, stream=None, selected_outputs=None, async_execution=None, execution_timeout_seconds=None)

Execute a workflow with optional input data.

```python
# With dict input (sent as the v2 input object)
result = client.execute_workflow("workflow-id", {"message": "Hello, world!"})

# With primitive input (sent as { input: { input: value } })
result = client.execute_workflow("workflow-id", "NVDA")

# With options (keyword-only arguments)
result = client.execute_workflow(
    "workflow-id",
    {"message": "Hello"},
    timeout=60.0,
    async_execution=True,
    execution_timeout_seconds=3600,
)
```

**Parameters:**
- `workflow_id` (str): The ID of the workflow to execute
- `input` (any, optional): Input data to pass to the workflow. Dicts become the v2 `input` object; primitives and lists become `{ input: value }` inside it. File objects are automatically converted to base64.
- `timeout` (float, keyword-only): Timeout in seconds (default: 30.0)
- `stream` (bool, keyword-only): Enable streaming responses
- `selected_outputs` (list, keyword-only): Block outputs to stream (e.g., `["agent1.content"]`)
- `async_execution` (bool, keyword-only): Execute asynchronously and return a run ID
- `execution_timeout_seconds` (int, keyword-only): Server-side async execution cap from 1 to 604800 seconds. Requires `async_execution=True` and cannot extend the account policy.

**Returns:** `WorkflowExecutionResult` or `AsyncExecutionResult`

##### get_workflow_status(workflow_id)

Get the status of a workflow (deployment status, etc.).

```python
status = client.get_workflow_status("workflow-id")
print("Is deployed:", status.is_deployed)
```

**Parameters:**
- `workflow_id` (str): The ID of the workflow

**Returns:** `WorkflowStatus`

##### validate_workflow(workflow_id)

Validate that a workflow is ready for execution.

```python
is_ready = client.validate_workflow("workflow-id")
if is_ready:
    # Workflow is deployed and ready
    pass
```

**Parameters:**
- `workflow_id` (str): The ID of the workflow

**Returns:** `bool`

##### execute_workflow_sync(workflow_id, input=None, *, timeout=30.0, stream=None, selected_outputs=None)

Execute a workflow synchronously (ensures non-async mode).

```python
result = client.execute_workflow_sync("workflow-id", {"data": "some input"}, timeout=60.0)
```

**Parameters:**
- `workflow_id` (str): The ID of the workflow to execute
- `input` (any, optional): Input data to pass to the workflow
- `timeout` (float, keyword-only): Timeout in seconds (default: 30.0)
- `stream` (bool, keyword-only): Enable streaming responses
- `selected_outputs` (list, keyword-only): Block outputs to stream (e.g., `["agent1.content"]`)

**Returns:** `WorkflowExecutionResult`

##### get_workflow_run(workflow_id, run_id, *, include_output=None, selected_outputs=None)

Get the status and optional outputs of a workflow run. Use the run ID returned by async execution.

```python
status = client.get_workflow_run(
    "workflow-id",
    "run-id",
    include_output=True,
    selected_outputs=["agent.content"]
)
print("Run status:", status["status"])
```

**Parameters:**
- `workflow_id` (str): The workflow ID
- `run_id` (str): The run ID returned from async execution
- `include_output` (bool, keyword-only): Include the final output for completed executions
- `selected_outputs` (list, keyword-only): Block output selectors to include

**Returns:** `dict`

##### get_job_status(job_id)

Get the status of a job created through the legacy async execution endpoint. New integrations should use `get_workflow_run()` with a run ID.

```python
status = client.get_job_status("legacy-job-id")
```

**Returns:** `dict`

##### execute_with_retry(workflow_id, input=None, *, timeout=30.0, stream=None, selected_outputs=None, async_execution=None, max_retries=3, initial_delay=1.0, max_delay=30.0, backoff_multiplier=2.0)

Execute a workflow with automatic retry on rate limit errors.

```python
result = client.execute_with_retry(
    "workflow-id",
    {"message": "Hello"},
    timeout=30.0,
    max_retries=3,
    initial_delay=1.0,
    max_delay=30.0,
    backoff_multiplier=2.0
)
```

**Parameters:**
- `workflow_id` (str): The ID of the workflow to execute
- `input` (any, optional): Input data to pass to the workflow
- `timeout` (float, keyword-only): Timeout in seconds (default: 30.0)
- `stream` (bool, keyword-only): Enable streaming responses
- `selected_outputs` (list, keyword-only): Block outputs to stream
- `async_execution` (bool, keyword-only): Execute asynchronously
- `max_retries` (int, keyword-only): Maximum retry attempts (default: 3)
- `initial_delay` (float, keyword-only): Initial delay in seconds (default: 1.0)
- `max_delay` (float, keyword-only): Maximum delay in seconds (default: 30.0)
- `backoff_multiplier` (float, keyword-only): Backoff multiplier (default: 2.0)

**Returns:** `WorkflowExecutionResult` or `AsyncExecutionResult`

##### get_rate_limit_info()

Get current rate limit information from the last API response.

```python
rate_info = client.get_rate_limit_info()
if rate_info:
    print("Remaining requests:", rate_info.remaining)
```

**Returns:** `RateLimitInfo` or `None`

##### get_usage_limits()

Get current usage limits and quota information.

```python
limits = client.get_usage_limits()
print("Current usage:", limits.usage)
```

**Returns:** `UsageLimits`

##### set_api_key(api_key)

Update the API key.

```python
client.set_api_key("new-api-key")
```

##### set_base_url(base_url)

Update the base URL.

```python
client.set_base_url("https://my-custom-domain.com")
```

##### close()

Close the underlying HTTP session.

```python
client.close()
```

## Data Classes

### WorkflowExecutionResult

```python
@dataclass
class WorkflowExecutionResult:
    success: bool
    output: Optional[Any] = None
    error: Optional[str] = None
    logs: Optional[list] = None
    metadata: Optional[Dict[str, Any]] = None
    trace_spans: Optional[list] = None
    total_duration: Optional[float] = None
    status: Optional[str] = None
```

`success` is `True` only for the `completed` and `paused` statuses. `status` carries the server's terminal status verbatim, so a cancelled run (`success=False`, `error=None`) is distinguishable from a failed one.

### WorkflowStatus

```python
@dataclass
class WorkflowStatus:
    is_deployed: bool
    deployed_at: Optional[str] = None
    needs_redeployment: bool = False
```

### SimStudioError

```python
class SimStudioError(Exception):
    def __init__(self, message: str, code: Optional[str] = None, status: Optional[int] = None):
        super().__init__(message)
        self.code = code
        self.status = status
```

### AsyncExecutionResult

```python
@dataclass
class AsyncExecutionResult:
    success: bool
    run_id: str
    status_url: str
    message: str = ""
    async_execution: bool = True
```

### RateLimitInfo

```python
@dataclass
class RateLimitInfo:
    limit: int
    remaining: int
    reset: int
    retry_after: Optional[int] = None
```

### UsageLimits

```python
@dataclass
class UsageLimits:
    success: bool
    rate_limit: Dict[str, Any]
    usage: Dict[str, Any]
```

## Examples

### Basic Workflow Execution

```python
import os
from simstudio import SimStudioClient

client = SimStudioClient(api_key=os.getenv("SIM_API_KEY"))

def run_workflow():
    try:
        # Check if workflow is ready
        is_ready = client.validate_workflow("my-workflow-id")
        if not is_ready:
            raise Exception("Workflow is not deployed or ready")

        # Execute the workflow
        result = client.execute_workflow(
            "my-workflow-id",
            {
                "message": "Process this data",
                "user_id": "12345"
            }
        )

        if result.success:
            print("Output:", result.output)
            print("Duration:", result.metadata.get("duration") if result.metadata else None)
        else:
            print("Workflow failed:", result.error)
            
    except Exception as error:
        print("Error:", error)

run_workflow()
```

### Error Handling

```python
from simstudio import SimStudioClient, SimStudioError
import os

client = SimStudioClient(api_key=os.getenv("SIM_API_KEY"))

def execute_with_error_handling():
    try:
        result = client.execute_workflow("workflow-id")
        return result
    except SimStudioError as error:
        if error.code == "UNAUTHORIZED":
            print("Invalid API key")
        elif error.code == "TIMEOUT":
            print("Workflow execution timed out")
        elif error.code == "USAGE_LIMIT_EXCEEDED":
            print("Usage limit exceeded")
        elif error.code == "INVALID_JSON":
            print("Invalid JSON in request body")
        else:
            print(f"Workflow error: {error}")
        raise
    except Exception as error:
        print(f"Unexpected error: {error}")
        raise
```

### Context Manager Usage

```python
from simstudio import SimStudioClient
import os

# Using context manager to automatically close the session
with SimStudioClient(api_key=os.getenv("SIM_API_KEY")) as client:
    result = client.execute_workflow("workflow-id")
    print("Result:", result)
# Session is automatically closed here
```

### Environment Configuration

```python
import os
from simstudio import SimStudioClient

# Using environment variables
client = SimStudioClient(
    api_key=os.getenv("SIM_API_KEY"),
    base_url=os.getenv("SIM_BASE_URL", "https://sim.ai")
)
```

### File Upload

File objects are automatically detected and converted to base64 format. Include them in your input under the field name matching your workflow's API trigger input format:

The SDK converts file objects to this format:
```python
{
  'type': 'file',
  'data': 'data:mime/type;base64,base64data',
  'name': 'filename',
  'mime': 'mime/type'
}
```

Alternatively, you can manually provide files using the URL format:
```python
{
  'type': 'url',
  'data': 'https://example.com/file.pdf',
  'name': 'file.pdf',
  'mime': 'application/pdf'
}
```

```python
from simstudio import SimStudioClient
import os

client = SimStudioClient(api_key=os.getenv("SIM_API_KEY"))

# Upload a single file - include it under the field name from your API trigger
with open('document.pdf', 'rb') as f:
    result = client.execute_workflow(
        'workflow-id',
        {
            'documents': [f],  # Must match your workflow's "files" field name
            'instructions': 'Analyze this document'
        }
    )

# Upload multiple files
with open('doc1.pdf', 'rb') as f1, open('doc2.pdf', 'rb') as f2:
    result = client.execute_workflow(
        'workflow-id',
        {
            'attachments': [f1, f2],  # Must match your workflow's "files" field name
            'query': 'Compare these documents'
        }
    )
```

### Batch Workflow Execution

```python
from simstudio import SimStudioClient
import os

client = SimStudioClient(api_key=os.getenv("SIM_API_KEY"))

def execute_workflows_batch(workflow_data_pairs):
    """Execute multiple workflows with different input data."""
    results = []

    for workflow_id, workflow_input in workflow_data_pairs:
        try:
            # Validate workflow before execution
            if not client.validate_workflow(workflow_id):
                print(f"Skipping {workflow_id}: not deployed")
                continue

            result = client.execute_workflow(workflow_id, workflow_input)
            results.append({
                "workflow_id": workflow_id,
                "success": result.success,
                "output": result.output,
                "error": result.error
            })

        except Exception as error:
            results.append({
                "workflow_id": workflow_id,
                "success": False,
                "error": str(error)
            })

    return results

# Example usage
workflows = [
    ("workflow-1", {"type": "analysis", "data": "sample1"}),
    ("workflow-2", {"type": "processing", "data": "sample2"}),
]

results = execute_workflows_batch(workflows)
for result in results:
    print(f"Workflow {result['workflow_id']}: {'Success' if result['success'] else 'Failed'}")
```

## Getting Your API Key

1. Log in to your [Sim](https://sim.ai) account
2. Navigate to your workflow
3. Click on "Deploy" to deploy your workflow
4. Select or create an API key during the deployment process
5. Copy the API key to use in your application

## Development

### Running Tests

To run the tests locally:

1. Clone the repository and navigate to the Python SDK directory:
   ```bash
   cd packages/python-sdk
   ```

2. Create and activate a virtual environment:
   ```bash
   python3 -m venv venv
   source venv/bin/activate  # On Windows: venv\Scripts\activate
   ```

3. Install the package in development mode with test dependencies:
   ```bash
   pip install -e ".[dev]"
   ```

4. Run the tests:
   ```bash
   pytest tests/ -v
   ```

### Code Quality

Run code quality checks:

```bash
# Code formatting
black simstudio/

# Linting
flake8 simstudio/ --max-line-length=100

# Type checking
mypy simstudio/

# Import sorting
isort simstudio/
```

## Requirements

- Python 3.8+
- requests >= 2.25.0

## License

Apache-2.0