MyBotBox Python SDK

The official Python SDK for MyBotBox, allowing you to execute workflows programmatically from your Python applications.

Installation

pip install yarlis-studio-sdk

Quick Start

import os
from mybotbox import MyBotBoxClient

# Initialize the client
client = MyBotBoxClient(
    api_key=os.getenv("MBB_API_KEY", "your-api-key-here"),
    base_url="https://mybotbox.com"
)

# 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

MyBotBoxClient

Constructor

MyBotBoxClient(api_key: str, base_url: str = "https://mybotbox.com")

• api_key (str): Your MyBotBox API key
• base_url (str, optional): Base URL for the MyBotBox API (defaults to https://mybotbox.com)

Methods

execute_workflow(workflow_id, input_data=None, timeout=30.0)

Execute a workflow with optional input data.

result = client.execute_workflow(
    "workflow-id",
    input_data={"message": "Hello, world!"},
    timeout=30.0  # 30 seconds
)

Parameters:

• workflow_id (str): The ID of the workflow to execute
• input_data (dict, optional): Input data to pass to the workflow. File objects are automatically converted to base64.
• timeout (float): Timeout in seconds (default: 30.0)

Returns: WorkflowExecutionResult

get_workflow_status(workflow_id)

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

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.

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_data=None, timeout=30.0)

Execute a workflow and poll for completion (useful for long-running workflows).

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

Parameters:

• workflow_id (str): The ID of the workflow to execute
• input_data (dict, optional): Input data to pass to the workflow
• timeout (float): Timeout for the initial request in seconds

Returns: WorkflowExecutionResult

set_api_key(api_key)

Update the API key.

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

set_base_url(base_url)

Update the base URL.

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

close()

Close the underlying HTTP session.

client.close()

Management API

Beyond executing workflows, MyBotBoxClient can manage your workflows, projects/folders, and workspaces — the same operations the dashboard performs. Management calls go to /api/... on your instance host (e.g. https://app.mybotbox.com) and are subject to your API key's token scopes.

from mybotbox import MyBotBoxClient
import os

client = MyBotBoxClient(api_key=os.getenv("MBB_API_KEY"), base_url="https://app.mybotbox.com")

# Workflows
client.list_workflows(workspace_id=None)
client.get_workflow("workflow-id")
client.create_workflow(name="Lead enricher", workspace_id="ws-id", description=None, color=None, folder_id=None)
client.update_workflow("workflow-id", name="Renamed", folderId="folder-id")
client.delete_workflow("workflow-id")        # soft-delete (recoverable)
client.restore_workflow("workflow-id")
client.duplicate_workflow("workflow-id")
client.deploy_workflow("workflow-id")
client.move_workflow("workflow-id", "folder-id")  # pass None to move to the root

# Projects & folders (a Project is a top-level folder)
client.list_projects("workspace-id", include_archived=False)
client.list_folders("workspace-id")
client.create_folder(name="Q3 campaigns", workspace_id="ws-id", parent_id=None)
client.update_folder("folder-id", name="Renamed")
client.delete_folder("folder-id")

# Workspaces
client.list_workspaces()
client.get_workspace("workspace-id")
client.create_workspace("New workspace")
client.update_workspace("workspace-id", name="Renamed")
client.delete_workspace("workspace-id", delete_templates=False)

Each method returns the decoded JSON response as a dict. A call that exceeds the API key's scopes raises MyBotBoxError with code="INSUFFICIENT_SCOPE".

Data Classes

WorkflowExecutionResult

@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

WorkflowStatus

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

MyBotBoxError

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

Examples

Basic Workflow Execution

import os
from mybotbox import MyBotBoxClient

client = MyBotBoxClient(api_key=os.getenv("MBB_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",
            input_data={
                "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

from mybotbox import MyBotBoxClient, MyBotBoxError
import os

client = MyBotBoxClient(api_key=os.getenv("MBB_API_KEY"))

def execute_with_error_handling():
    try:
        result = client.execute_workflow("workflow-id")
        return result
    except MyBotBoxError 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

from mybotbox import MyBotBoxClient
import os

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

Environment Configuration

import os
from mybotbox import MyBotBoxClient

# Using environment variables
client = MyBotBoxClient(
    api_key=os.getenv("MBB_API_KEY"),
    base_url=os.getenv("MBB_BASE_URL", "https://mybotbox.com")
)

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:

{
  'type': 'file',
  'data': 'data:mime/type;base64,base64data',
  'name': 'filename',
  'mime': 'mime/type'
}

Alternatively, you can manually provide files using the URL format:

{
  'type': 'url',
  'data': 'https://example.com/file.pdf',
  'name': 'file.pdf',
  'mime': 'application/pdf'
}

from mybotbox import MyBotBoxClient
import os

client = MyBotBoxClient(api_key=os.getenv("MBB_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',
        input_data={
            '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',
        input_data={
            'attachments': [f1, f2],  # Must match your workflow's "files" field name
            'query': 'Compare these documents'
        }
    )

Batch Workflow Execution

from mybotbox import MyBotBoxClient
import os

client = MyBotBoxClient(api_key=os.getenv("MBB_API_KEY"))

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

    for workflow_id, input_data 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, input_data)
            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 MyBotBox 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:

   cd packages/python-sdk
   

2. Create and activate a virtual environment:

   python3 -m venv venv
   source venv/bin/activate  # On Windows: venv\Scripts\activate
   

3. Install the package in development mode with test dependencies:

   pip install -e ".[dev]"
   

4. Run the tests:

   pytest tests/ -v
   

Code Quality

Run code quality checks:

# Code formatting
black ystudio/

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

# Type checking
mypy ystudio/

# Import sorting
isort ystudio/

Requirements

• Python 3.8+
• requests >= 2.25.0

License

Apache-2.0