Python SDK

Simの公式Python SDKを使用すると、公式Python SDKを使用してPythonアプリケーションからプログラムでワークフローを実行できます。

Python SDKはPython 3.8以上をサポートし、同期的なワークフロー実行を提供します。現在、すべてのワークフロー実行は同期的です。

インストール

pipを使用してSDKをインストールします：

pip install simstudio-sdk

クイックスタート

以下は、始めるための簡単な例です：

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)

APIリファレンス

SimStudioClient

コンストラクタ

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

パラメータ：

api_key (str): SimのAPIキー
base_url (str, オプション): Sim APIのベースURL

メソッド

execute_workflow()

オプションの入力データでワークフローを実行します。

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

パラメータ：

workflow_id (str): 実行するワークフローのID
input_data (dict, オプション): ワークフローに渡す入力データ
timeout (float, オプション): タイムアウト（秒）（デフォルト：30.0）

戻り値： WorkflowExecutionResult

get_workflow_status()

ワークフローのステータス（デプロイメントステータスなど）を取得します。

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

パラメータ：

workflow_id (str): ワークフローのID

戻り値: WorkflowStatus

validate_workflow()

ワークフローが実行準備ができているかを検証します。

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

パラメータ:

workflow_id (str): ワークフローのID

戻り値: bool

execute_workflow_sync()

現在、このメソッドは execute_workflow() と同一です。すべての実行は同期的に行われるためです。このメソッドは、将来的に非同期実行が追加された際の互換性のために提供されています。

ワークフローを実行します（現在は同期的、execute_workflow() と同じ）。

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

パラメータ:

workflow_id (str): 実行するワークフローのID
input_data (dict, optional): ワークフローに渡す入力データ
timeout (float): 初期リクエストのタイムアウト（秒）

戻り値: WorkflowExecutionResult

set_api_key()

APIキーを更新します。

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

set_base_url()

ベースURLを更新します。

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

close()

基盤となるHTTPセッションを閉じます。

client.close()

データクラス

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

例

基本的なワークフロー実行

APIキーを使用してSimStudioClientをセットアップします。

ワークフローがデプロイされ、実行準備ができているか確認します。

入力データでワークフローを実行します。

実行結果を処理し、エラーがあれば対処します。

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

エラー処理

ワークフロー実行中に発生する可能性のある様々なタイプのエラーを処理します：

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

コンテキストマネージャーの使用

リソースのクリーンアップを自動的に処理するためにコンテキストマネージャーとしてクライアントを使用します：

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

バッチワークフロー実行

複数のワークフローを効率的に実行します：

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

環境設定

環境変数を使用してクライアントを設定します：

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

APIキーの取得方法

Simに移動してアカウントにログインします。

プログラムで実行したいワークフローに移動します。

まだデプロイされていない場合は、「デプロイ」をクリックしてワークフローをデプロイします。

デプロイプロセス中に、APIキーを選択または作成します。

Pythonアプリケーションで使用するAPIキーをコピーします。

APIキーを安全に保管し、決してバージョン管理にコミットしないでください。環境変数や安全な設定管理を使用してください。

要件

Python 3.8+
requests >= 2.25.0

ライセンス

Apache-2.0

Python SDK

On this page