OpenAgentAGI
Sdks

Python SDK

Das offizielle Python SDK für OpenAgentAGI ermöglicht es Ihnen, Workflows programmatisch aus Ihren Python-Anwendungen mithilfe des offiziellen Python SDKs auszuführen.

Das Python SDK unterstützt Python 3.8+ und bietet synchrone Workflow-Ausführung. Alle Workflow-Ausführungen sind derzeit synchron.

Installation

Installieren Sie das SDK mit pip:

pip install oaistudio-sdk

Schnellstart

Hier ist ein einfaches Beispiel für den Einstieg:

from oaistudio import SimStudioClient

# Initialize the client
client = SimStudioClient(
    api_key="your-api-key-here",
    base_url="https://www.openagentagi.com"  # optional, defaults to https://www.openagentagi.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-Referenz

SimStudioClient

Konstruktor

SimStudioClient(api_key: str, base_url: str = "https://www.openagentagi.com")

Parameter:

  • api_key (str): Ihr OpenAgentAGI API-Schlüssel
  • base_url (str, optional): Basis-URL für die OpenAgentAGI API

Methoden

execute_workflow()

Führt einen Workflow mit optionalen Eingabedaten aus.

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

Parameter:

  • workflow_id (str): Die ID des auszuführenden Workflows
  • input_data (dict, optional): Eingabedaten, die an den Workflow übergeben werden
  • timeout (float, optional): Timeout in Sekunden (Standard: 30.0)

Rückgabewert: WorkflowExecutionResult

get_workflow_status()

Ruft den Status eines Workflows ab (Deployment-Status usw.).

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

Parameter:

  • workflow_id (str): Die ID des Workflows

Rückgabe: WorkflowStatus

validate_workflow()

Überprüft, ob ein Workflow für die Ausführung bereit ist.

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

Parameter:

  • workflow_id (str): Die ID des Workflows

Rückgabe: bool

execute_workflow_sync()

Derzeit ist diese Methode identisch mit execute_workflow(), da alle Ausführungen synchron sind. Diese Methode wird für zukünftige Kompatibilität bereitgestellt, wenn asynchrone Ausführung hinzugefügt wird.

Führt einen Workflow aus (derzeit synchron, identisch mit execute_workflow()).

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

Parameter:

  • workflow_id (str): Die ID des auszuführenden Workflows
  • input_data (dict, optional): Eingabedaten, die an den Workflow übergeben werden
  • timeout (float): Timeout für die initiale Anfrage in Sekunden

Rückgabe: WorkflowExecutionResult

set_api_key()

Aktualisiert den API-Schlüssel.

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

Aktualisiert die Basis-URL.

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

Schließt die zugrunde liegende HTTP-Sitzung.

client.close()

Datenklassen

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

Beispiele

Grundlegende Workflow-Ausführung

Richten Sie den SimStudioClient mit Ihrem API-Schlüssel ein.

Prüfen Sie, ob der Workflow bereitgestellt und für die Ausführung bereit ist.

Führen Sie den Workflow mit Ihren Eingabedaten aus.

Verarbeiten Sie das Ausführungsergebnis und behandeln Sie eventuelle Fehler.

import os
from oaistudio 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()

Fehlerbehandlung

Behandeln Sie verschiedene Fehlertypen, die während der Workflow-Ausführung auftreten können:

from oaistudio 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

Verwendung des Kontextmanagers

Verwenden Sie den Client als Kontextmanager, um die Ressourcenbereinigung automatisch zu handhaben:

from oaistudio 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

Batch-Workflow-Ausführung

Führen Sie mehrere Workflows effizient aus:

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

Umgebungskonfiguration

Konfigurieren Sie den Client mit Umgebungsvariablen:

import os
from oaistudio import SimStudioClient

# Development configuration
client = SimStudioClient(
    api_key=os.getenv("SIMSTUDIO_API_KEY"),
    base_url=os.getenv("SIMSTUDIO_BASE_URL", "https://www.openagentagi.com")
)
import os
from oaistudio 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://www.openagentagi.com")
)

API-Schlüssel erhalten

Navigieren Sie zu OpenAgentAGI und melden Sie sich bei Ihrem Konto an.

Navigieren Sie zu dem Workflow, den Sie programmatisch ausführen möchten.

Klicken Sie auf "Deploy", um Ihren Workflow bereitzustellen, falls dies noch nicht geschehen ist.

Wählen Sie während des Bereitstellungsprozesses einen API-Schlüssel aus oder erstellen Sie einen neuen.

Kopieren Sie den API-Schlüssel zur Verwendung in Ihrer Python-Anwendung.

Halte deinen API-Schlüssel sicher und committe ihn niemals in die Versionskontrolle. Verwende Umgebungsvariablen oder sicheres Konfigurationsmanagement.

Anforderungen

  • Python 3.8+
  • requests >= 2.25.0

Lizenz

Apache-2.0

YAML-Schema für Workflow-BlöckeTypeScript/JavaScript SDK
On this page

On this page

Installation
Schnellstart
API-Referenz
SimStudioClient
Konstruktor
Methoden
execute_workflow()
get_workflow_status()
validate_workflow()
execute_workflow_sync()
set_api_key()
set_base_url()
close()
Datenklassen
WorkflowExecutionResult
WorkflowStatus
SimStudioError
Beispiele
Grundlegende Workflow-Ausführung
Fehlerbehandlung
Verwendung des Kontextmanagers
Batch-Workflow-Ausführung
Umgebungskonfiguration
API-Schlüssel erhalten
Anforderungen
Lizenz
Python SDK