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 Simbase_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écuterinput_data
(dict, facultatif) : Données d'entrée à transmettre au workflowtimeout
(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écuterinput_data
(dict, facultatif) : Données d'entrée à transmettre au workflowtimeout
(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