Python - Aurora AI

El SDK oficial de Python para Aurora Workflow.te permite ejecutar flujos de trabajo programáticamente desde tus aplicaciones Python utilizando el SDK oficial de Python.

El SDK de Python es compatible con Python 3.8+ con soporte para ejecución asíncrona, limitación automática de velocidad con retroceso exponencial y seguimiento de uso.

Instalación

Instala el SDK usando pip:

pip install simstudio-sdk

Inicio rápido

Aquí tienes un ejemplo sencillo para empezar:

from simstudio import SimStudioClient

# Initialize the client
client = SimStudioClient(
    api_key="your-api-key-here",
    base_url="https://Aurora Workflow.ai"  # optional, defaults to https://Aurora Workflow.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)

Referencia de la API

SimStudioClient

Constructor

SimStudioClient(api_key: str, base_url: str = "https://Aurora Workflow.ai")

Parámetros:

api_key (str): Tu clave API de Sim
base_url (str, opcional): URL base para la API de Sim

Métodos

execute_workflow()

Ejecuta un flujo de trabajo con datos de entrada opcionales.

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

Parámetros:

workflow_id (str): El ID del flujo de trabajo a ejecutar
input_data (dict, opcional): Datos de entrada para pasar al flujo de trabajo
timeout (float, opcional): Tiempo de espera en segundos (predeterminado: 30.0)
stream (bool, opcional): Habilitar respuestas en streaming (predeterminado: False)
selected_outputs (list[str], opcional): Salidas de bloque para transmitir en formato blockName.attribute (p. ej., ["agent1.content"])
async_execution (bool, opcional): Ejecutar de forma asíncrona (predeterminado: False)

Devuelve: WorkflowExecutionResult | AsyncExecutionResult Cuando async_execution=True, devuelve inmediatamente un ID de tarea para sondeo. De lo contrario, espera a que se complete.

get_workflow_status()

Obtener el estado de un flujo de trabajo (estado de implementación, etc.).

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

Parámetros:

workflow_id (str): El ID del flujo de trabajo

Devuelve: WorkflowStatus

validate_workflow()

Validar que un flujo de trabajo está listo para su ejecución.

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

Parámetros:

workflow_id (str): El ID del flujo de trabajo

Devuelve: bool

get_job_status()

Obtener el estado de una ejecución de trabajo asíncrono.

status = client.get_job_status("task-id-from-async-execution")
print("Status:", status["status"])  # 'queued', 'processing', 'completed', 'failed'
if status["status"] == "completed":
    print("Output:", status["output"])

Parámetros:

task_id (str): El ID de tarea devuelto de la ejecución asíncrona

Devuelve: Dict[str, Any] Campos de respuesta:

success (bool): Si la solicitud fue exitosa
taskId (str): El ID de la tarea
status (str): Uno de 'queued', 'processing', 'completed', 'failed', 'cancelled'
metadata (dict): Contiene startedAt, completedAt, y duration
output (any, opcional): La salida del flujo de trabajo (cuando se completa)
error (any, opcional): Detalles del error (cuando falla)
estimatedDuration (int, opcional): Duración estimada en milisegundos (cuando está procesando/en cola)

execute_with_retry()

Ejecutar un flujo de trabajo con reintento automático en errores de límite de velocidad usando retroceso exponencial.

result = client.execute_with_retry(
    "workflow-id",
    input_data={"message": "Hello"},
    timeout=30.0,
    max_retries=3,           # Maximum number of retries
    initial_delay=1.0,       # Initial delay in seconds
    max_delay=30.0,          # Maximum delay in seconds
    backoff_multiplier=2.0   # Exponential backoff multiplier
)

Parámetros:

workflow_id (str): El ID del flujo de trabajo a ejecutar
input_data (dict, opcional): Datos de entrada para pasar al flujo de trabajo
timeout (float, opcional): Tiempo de espera en segundos
stream (bool, opcional): Habilitar respuestas en streaming
selected_outputs (list, opcional): Salidas de bloque para transmitir
async_execution (bool, opcional): Ejecutar de forma asíncrona
max_retries (int, opcional): Número máximo de reintentos (predeterminado: 3)
initial_delay (float, opcional): Retraso inicial en segundos (predeterminado: 1.0)
max_delay (float, opcional): Retraso máximo en segundos (predeterminado: 30.0)
backoff_multiplier (float, opcional): Multiplicador de retroceso (predeterminado: 2.0)

Devuelve: WorkflowExecutionResult | AsyncExecutionResult La lógica de reintento utiliza retroceso exponencial (1s → 2s → 4s → 8s…) con fluctuación de ±25% para evitar el efecto de manada. Si la API proporciona un encabezado retry-after, se utilizará en su lugar.

get_rate_limit_info()

Obtiene la información actual del límite de tasa de la última respuesta de la API.

rate_limit_info = client.get_rate_limit_info()
if rate_limit_info:
    print("Limit:", rate_limit_info.limit)
    print("Remaining:", rate_limit_info.remaining)
    print("Reset:", datetime.fromtimestamp(rate_limit_info.reset))

Devuelve: RateLimitInfo | None

get_usage_limits()

Obtiene los límites de uso actuales y la información de cuota para tu cuenta.

limits = client.get_usage_limits()
print("Sync requests remaining:", limits.rate_limit["sync"]["remaining"])
print("Async requests remaining:", limits.rate_limit["async"]["remaining"])
print("Current period cost:", limits.usage["currentPeriodCost"])
print("Plan:", limits.usage["plan"])

Devuelve: UsageLimits Estructura de respuesta:

{
    "success": bool,
    "rateLimit": {
        "sync": {
            "isLimited": bool,
            "limit": int,
            "remaining": int,
            "resetAt": str
        },
        "async": {
            "isLimited": bool,
            "limit": int,
            "remaining": int,
            "resetAt": str
        },
        "authType": str  # 'api' or 'manual'
    },
    "usage": {
        "currentPeriodCost": float,
        "limit": float,
        "plan": str  # e.g., 'free', 'pro'
    }
}

set_api_key()

Actualiza la clave API.

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

set_base_url()

Actualiza la URL base.

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

close()

Cierra la sesión HTTP subyacente.

client.close()

Clases de datos

WorkflowExecutionResult

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

AsyncExecutionResult

@dataclass
class AsyncExecutionResult:
    success: bool
    task_id: str
    status: str  # 'queued'
    created_at: str
    links: Dict[str, str]  # e.g., {"status": "/api/jobs/{taskId}"}

WorkflowStatus

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

RateLimitInfo

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

UsageLimits

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

SimStudioError

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

Códigos de error comunes:

UNAUTHORIZED: Clave API inválida
TIMEOUT: Tiempo de espera agotado
RATE_LIMIT_EXCEEDED: Límite de tasa excedido
USAGE_LIMIT_EXCEEDED: Límite de uso excedido
EXECUTION_ERROR: Ejecución del flujo de trabajo fallida

Ejemplos

Ejecución básica de flujo de trabajo

Inicializar el cliente

Configura el SimStudioClient con tu clave API.

Validar el flujo de trabajo

Comprueba si el flujo de trabajo está desplegado y listo para su ejecución.

Ejecutar el flujo de trabajo

Ejecuta el flujo de trabajo con tus datos de entrada.

Manejar el resultado

Procesa el resultado de la ejecución y gestiona cualquier error.

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",
            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()

Manejo de errores

Maneja diferentes tipos de errores que pueden ocurrir durante la ejecución del flujo de trabajo:

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

Uso del gestor de contexto

Usa el cliente como un gestor de contexto para manejar automáticamente la limpieza de recursos:

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

Ejecución de flujos de trabajo por lotes

Ejecuta múltiples flujos de trabajo de manera eficiente:

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, 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'}")

Ejecución asíncrona de flujos de trabajo

Ejecuta flujos de trabajo de forma asíncrona para tareas de larga duración:

import os
import time
from simstudio import SimStudioClient

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

def execute_async():
    try:
        # Start async execution
        result = client.execute_workflow(
            "workflow-id",
            input_data={"data": "large dataset"},
            async_execution=True  # Execute asynchronously
        )

        # Check if result is an async execution
        if hasattr(result, 'task_id'):
            print(f"Task ID: {result.task_id}")
            print(f"Status endpoint: {result.links['status']}")

            # Poll for completion
            status = client.get_job_status(result.task_id)

            while status["status"] in ["queued", "processing"]:
                print(f"Current status: {status['status']}")
                time.sleep(2)  # Wait 2 seconds
                status = client.get_job_status(result.task_id)

            if status["status"] == "completed":
                print("Workflow completed!")
                print(f"Output: {status['output']}")
                print(f"Duration: {status['metadata']['duration']}")
            else:
                print(f"Workflow failed: {status['error']}")

    except Exception as error:
        print(f"Error: {error}")

execute_async()

Límite de tasa y reintentos

Maneja los límites de tasa automáticamente con retroceso exponencial:

import os
from simstudio import SimStudioClient, SimStudioError

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

def execute_with_retry_handling():
    try:
        # Automatically retries on rate limit
        result = client.execute_with_retry(
            "workflow-id",
            input_data={"message": "Process this"},
            max_retries=5,
            initial_delay=1.0,
            max_delay=60.0,
            backoff_multiplier=2.0
        )

        print(f"Success: {result}")
    except SimStudioError as error:
        if error.code == "RATE_LIMIT_EXCEEDED":
            print("Rate limit exceeded after all retries")

            # Check rate limit info
            rate_limit_info = client.get_rate_limit_info()
            if rate_limit_info:
                from datetime import datetime
                reset_time = datetime.fromtimestamp(rate_limit_info.reset)
                print(f"Rate limit resets at: {reset_time}")

execute_with_retry_handling()

Monitoreo de uso

Monitorea el uso de tu cuenta y sus límites:

import os
from simstudio import SimStudioClient

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

def check_usage():
    try:
        limits = client.get_usage_limits()

        print("=== Rate Limits ===")
        print("Sync requests:")
        print(f"  Limit: {limits.rate_limit['sync']['limit']}")
        print(f"  Remaining: {limits.rate_limit['sync']['remaining']}")
        print(f"  Resets at: {limits.rate_limit['sync']['resetAt']}")
        print(f"  Is limited: {limits.rate_limit['sync']['isLimited']}")

        print("\nAsync requests:")
        print(f"  Limit: {limits.rate_limit['async']['limit']}")
        print(f"  Remaining: {limits.rate_limit['async']['remaining']}")
        print(f"  Resets at: {limits.rate_limit['async']['resetAt']}")
        print(f"  Is limited: {limits.rate_limit['async']['isLimited']}")

        print("\n=== Usage ===")
        print(f"Current period cost: ${limits.usage['currentPeriodCost']:.2f}")
        print(f"Limit: ${limits.usage['limit']:.2f}")
        print(f"Plan: {limits.usage['plan']}")

        percent_used = (limits.usage['currentPeriodCost'] / limits.usage['limit']) * 100
        print(f"Usage: {percent_used:.1f}%")

        if percent_used > 80:
            print("⚠️  Warning: You are approaching your usage limit!")

    except Exception as error:
        print(f"Error checking usage: {error}")

check_usage()

Ejecución de flujo de trabajo en streaming

Ejecuta flujos de trabajo con respuestas en tiempo real:

from simstudio import SimStudioClient
import os

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

def execute_with_streaming():
    """Execute workflow with streaming enabled."""
    try:
        # Enable streaming for specific block outputs
        result = client.execute_workflow(
            "workflow-id",
            input_data={"message": "Count to five"},
            stream=True,
            selected_outputs=["agent1.content"]  # Use blockName.attribute format
        )

        print("Workflow result:", result)
    except Exception as error:
        print("Error:", error)

execute_with_streaming()

La respuesta en streaming sigue el formato de Server-Sent Events (SSE):

data: {"blockId":"7b7735b9-19e5-4bd6-818b-46aae2596e9f","chunk":"One"}

data: {"blockId":"7b7735b9-19e5-4bd6-818b-46aae2596e9f","chunk":", two"}

data: {"event":"done","success":true,"output":{},"metadata":{"duration":610}}

data: [DONE]

Ejemplo de streaming con Flask:

from flask import Flask, Response, stream_with_context
import requests
import json
import os

app = Flask(__name__)

@app.route('/stream-workflow')
def stream_workflow():
    """Stream workflow execution to the client."""

    def generate():
        response = requests.post(
            'https://workflow.aurora-ai.co/api/workflows/WORKFLOW_ID/execute',
            headers={
                'Content-Type': 'application/json',
                'X-API-Key': os.getenv('SIM_API_KEY')
            },
            json={
                'message': 'Generate a story',
                'stream': True,
                'selectedOutputs': ['agent1.content']
            },
            stream=True
        )

        for line in response.iter_lines():
            if line:
                decoded_line = line.decode('utf-8')
                if decoded_line.startswith('data: '):
                    data = decoded_line[6:]  # Remove 'data: ' prefix

                    if data == '[DONE]':
                        break

                    try:
                        parsed = json.loads(data)
                        if 'chunk' in parsed:
                            yield f"data: {json.dumps(parsed)}\n\n"
                        elif parsed.get('event') == 'done':
                            yield f"data: {json.dumps(parsed)}\n\n"
                            print("Execution complete:", parsed.get('metadata'))
                    except json.JSONDecodeError:
                        pass

    return Response(
        stream_with_context(generate()),
        mimetype='text/event-stream'
    )

if __name__ == '__main__':
    app.run(debug=True)

Configuración del entorno

Configura el cliente usando variables de entorno:

Tab Title
Tab Title

import os
from simstudio import SimStudioClient

# Development configuration
client = SimStudioClient(
    api_key=os.getenv("SIM_API_KEY")
    base_url=os.getenv("SIM_BASE_URL", "https://Aurora Workflow.ai")
)

import os
from simstudio import SimStudioClient

# Production configuration with error handling
api_key = os.getenv("SIM_API_KEY")
if not api_key:
    raise ValueError("SIM_API_KEY environment variable is required")

client = SimStudioClient(
    api_key=api_key,
    base_url=os.getenv("SIM_BASE_URL", "https://Aurora Workflow.ai")
)

Obtener tu clave API

Inicia sesión en Sim

Navega a [Sim](https://Aurora Workflow.ai) e inicia sesión en tu cuenta.

Abre tu flujo de trabajo

Navega al flujo de trabajo que quieres ejecutar programáticamente.

Despliega tu flujo de trabajo

Haz clic en “Deploy” para desplegar tu flujo de trabajo si aún no ha sido desplegado.

Crea o selecciona una clave API

Durante el proceso de despliegue, selecciona o crea una clave API.

Copia la clave API

Copia la clave API para usarla en tu aplicación Python.

Requisitos

Python 3.8+
requests >= 2.25.0

Licencia

Apache-2.0

​Instalación

​Inicio rápido

​Referencia de la API

​SimStudioClient

​Constructor

​Métodos

execute_workflow()

get_workflow_status()

validate_workflow()

get_job_status()

execute_with_retry()

get_rate_limit_info()

get_usage_limits()

set_api_key()

set_base_url()

close()

​Clases de datos

​WorkflowExecutionResult

​AsyncExecutionResult

​WorkflowStatus

​RateLimitInfo

​UsageLimits

​SimStudioError

​Ejemplos

​Ejecución básica de flujo de trabajo

​Manejo de errores

​Uso del gestor de contexto

​Ejecución de flujos de trabajo por lotes

​Ejecución asíncrona de flujos de trabajo

​Límite de tasa y reintentos

​Monitoreo de uso

​Ejecución de flujo de trabajo en streaming

​Configuración del entorno

​Obtener tu clave API

​Requisitos

​Licencia

Instalación

Inicio rápido

Referencia de la API

SimStudioClient

Constructor

Métodos

Clases de datos

WorkflowExecutionResult

AsyncExecutionResult

WorkflowStatus

RateLimitInfo

UsageLimits

SimStudioError

Ejemplos

Ejecución básica de flujo de trabajo

Manejo de errores

Uso del gestor de contexto

Ejecución de flujos de trabajo por lotes

Ejecución asíncrona de flujos de trabajo

Límite de tasa y reintentos

Monitoreo de uso

Ejecución de flujo de trabajo en streaming

Configuración del entorno

Obtener tu clave API

Requisitos

Licencia