Sim
Sdks

Python SDK

Le SDK Python officiel pour Sim vous permet d'exécuter des workflows de maniÚre programmatique à partir de vos applications Python en utilisant le SDK Python officiel.

Le SDK Python prend en charge Python 3.8+ et fournit une exécution synchrone des workflows. Toutes les exécutions de workflow sont actuellement synchrones.

Installation

Installez le SDK en utilisant pip :

pip install simstudio-sdk

Démarrage rapide

Voici un exemple simple pour commencer :

from simstudio import SimStudioClient

# Initialize the client
client = SimStudioClient(
    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)

Référence de l'API

SimStudioClient

Constructeur

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

ParamĂštres :

  • api_key (str) : Votre clĂ© API Sim
  • base_url (str, facultatif) : URL de base pour l'API Sim

Méthodes

execute_workflow()

Exécuter un workflow avec des données d'entrée facultatives.

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

ParamĂštres :

  • workflow_id (str) : L'ID du workflow Ă  exĂ©cuter
  • input_data (dict, facultatif) : DonnĂ©es d'entrĂ©e Ă  transmettre au workflow
  • timeout (float, facultatif) : DĂ©lai d'attente en secondes (par dĂ©faut : 30.0)

Retourne : WorkflowExecutionResult

get_workflow_status()

Obtenir le statut d'un workflow (statut de déploiement, etc.).

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

ParamĂštres :

  • workflow_id (str) : L'ID du workflow

Retourne : WorkflowStatus

validate_workflow()

Valide qu'un workflow est prĂȘt pour l'exĂ©cution.

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

ParamĂštres :

  • workflow_id (str) : L'identifiant du workflow

Retourne : bool

execute_workflow_sync()

Actuellement, cette méthode est identique à execute_workflow() puisque toutes les exécutions sont synchrones. Cette méthode est fournie pour une compatibilité future lorsque l'exécution asynchrone sera ajoutée.

Exécute un workflow (actuellement synchrone, identique à execute_workflow()).

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

ParamĂštres :

  • workflow_id (str) : L'identifiant du workflow Ă  exĂ©cuter
  • input_data (dict, facultatif) : DonnĂ©es d'entrĂ©e Ă  transmettre au workflow
  • timeout (float) : DĂ©lai d'attente pour la requĂȘte initiale en secondes

Retourne : WorkflowExecutionResult

set_api_key()

Met à jour la clé API.

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

Met Ă  jour l'URL de base.

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

Ferme la session HTTP sous-jacente.

client.close()

Classes de données

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

WorkflowStatus

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

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

Exemples

Exécution de flux de travail basique

Configurez le SimStudioClient avec votre clé API.

VĂ©rifiez si le flux de travail est dĂ©ployĂ© et prĂȘt pour l'exĂ©cution.

Lancez le flux de travail avec vos données d'entrée.

Traitez le résultat de l'exécution et gérez les éventuelles erreurs.

import os
from simstudio import SimStudioClient

client = SimStudioClient(api_key=os.getenv("SIMSTUDIO_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()

Gestion des erreurs

Gérez différents types d'erreurs qui peuvent survenir pendant l'exécution du flux de travail :

from simstudio import SimStudioClient, SimStudioError
import os

client = SimStudioClient(api_key=os.getenv("SIMSTUDIO_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

Utilisation du gestionnaire de contexte

Utilisez le client comme gestionnaire de contexte pour gérer automatiquement le nettoyage des ressources :

from simstudio import SimStudioClient
import os

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

Exécution de flux de travail par lots

Exécutez plusieurs flux de travail efficacement :

from simstudio import SimStudioClient
import os

client = SimStudioClient(api_key=os.getenv("SIMSTUDIO_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'}")

Configuration de l'environnement

Configurez le client en utilisant des variables d'environnement :

import os
from simstudio import SimStudioClient

# Development configuration
client = SimStudioClient(
    api_key=os.getenv("SIMSTUDIO_API_KEY"),
    base_url=os.getenv("SIMSTUDIO_BASE_URL", "https://sim.ai")
)
import os
from simstudio import SimStudioClient

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

client = SimStudioClient(
    api_key=api_key,
    base_url=os.getenv("SIMSTUDIO_BASE_URL", "https://sim.ai")
)

Obtenir votre clé API

Accédez à Sim et connectez-vous à votre compte.

Naviguez vers le flux de travail que vous souhaitez exécuter par programmation.

Cliquez sur "Déployer" pour déployer votre flux de travail s'il n'a pas encore été déployé.

Pendant le processus de déploiement, sélectionnez ou créez une clé API.

Copiez la clé API pour l'utiliser dans votre application Python.

Gardez votre clé API sécurisée et ne la soumettez jamais au contrÎle de version. Utilisez des variables d'environnement ou une gestion de configuration sécurisée.

Prérequis

  • Python 3.8+
  • requests >= 2.25.0

Licence

Apache-2.0

Python SDK