| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
| Name | Name | Last commit date | ||
|---|---|---|---|---|
parent directory.. | ||||
The official Python SDK for Sim, allowing you to execute workflows programmatically from your Python applications.
0.2.x talks to the v2 API and has no fallback to the older endpoints, so it requires a Sim deployment that serves POST /api/v2/workflows/{id}/execute. That surface is newer than the endpoints 0.1.x used. If it is unavailable, execute_workflow raises SimStudioError('HTTP 404: Not Found') — upgrade the server or pin simstudio-sdk<0.2, which keeps using /api/workflows/{id}/execute and /api/jobs/{id}.
0.2.0 is a breaking release.
Note one deliberate difference from the TypeScript SDK: a failed synchronous run throws there, but here it returns normally with error set and status='failed'.
pip install simstudio-sdkimport os
from simstudio import SimStudioClient
# Initialize the client
client = SimStudioClient(
api_key=os.getenv("SIM_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)SimStudioClient(api_key: str, base_url: str = "https://sim.ai")Execute a workflow with optional input data.
# With dict input (sent as the v2 input object)
result = client.execute_workflow("workflow-id", {"message": "Hello, world!"})
# With primitive input (sent as { input: { input: value } })
result = client.execute_workflow("workflow-id", "NVDA")
# With options (keyword-only arguments)
result = client.execute_workflow(
"workflow-id",
{"message": "Hello"},
timeout=60.0,
async_execution=True,
execution_timeout_seconds=3600,
)Parameters:
Returns: WorkflowExecutionResult or AsyncExecutionResult
Get the status of a workflow (deployment status, etc.).
status = client.get_workflow_status("workflow-id")
print("Is deployed:", status.is_deployed)Parameters:
Returns: WorkflowStatus
Validate that a workflow is ready for execution.
is_ready = client.validate_workflow("workflow-id")
if is_ready:
# Workflow is deployed and ready
passParameters:
Returns: bool
Execute a workflow synchronously (ensures non-async mode).
result = client.execute_workflow_sync("workflow-id", {"data": "some input"}, timeout=60.0)Parameters:
Returns: WorkflowExecutionResult
Get the status and optional outputs of a workflow run. Use the run ID returned by async execution.
status = client.get_workflow_run(
"workflow-id",
"run-id",
include_output=True,
selected_outputs=["agent.content"]
)
print("Run status:", status["status"])Parameters:
Returns: dict
Get the status of a job created through the legacy async execution endpoint. New integrations should use get_workflow_run() with a run ID.
status = client.get_job_status("legacy-job-id")Returns: dict
Execute a workflow with automatic retry on rate limit errors.
result = client.execute_with_retry(
"workflow-id",
{"message": "Hello"},
timeout=30.0,
max_retries=3,
initial_delay=1.0,
max_delay=30.0,
backoff_multiplier=2.0
)Parameters:
Returns: WorkflowExecutionResult or AsyncExecutionResult
Get current rate limit information from the last API response.
rate_info = client.get_rate_limit_info()
if rate_info:
print("Remaining requests:", rate_info.remaining)Returns: RateLimitInfo or None
Get current usage limits and quota information.
limits = client.get_usage_limits()
print("Current usage:", limits.usage)Returns: UsageLimits
Update the API key.
client.set_api_key("new-api-key")Update the base URL.
client.set_base_url("https://my-custom-domain.com")Close the underlying HTTP session.
client.close()@dataclass
class WorkflowExecutionResult:
success: bool
output: Optional[Any] = None
error: Optional[str] = None
logs: Optional[list] = None
metadata: Optional[Dict[str, Any]] = None
trace_spans: Optional[list] = None
total_duration: Optional[float] = None
status: Optional[str] = Nonesuccess is True only for the completed and paused statuses. status carries the server's terminal status verbatim, so a cancelled run (success=False, error=None) is distinguishable from a failed one.
@dataclass
class WorkflowStatus:
is_deployed: bool
deployed_at: Optional[str] = None
needs_redeployment: bool = Falseclass SimStudioError(Exception):
def __init__(self, message: str, code: Optional[str] = None, status: Optional[int] = None):
super().__init__(message)
self.code = code
self.status = status@dataclass
class AsyncExecutionResult:
success: bool
run_id: str
status_url: str
message: str = ""
async_execution: bool = True@dataclass
class RateLimitInfo:
limit: int
remaining: int
reset: int
retry_after: Optional[int] = None@dataclass
class UsageLimits:
success: bool
rate_limit: Dict[str, Any]
usage: Dict[str, Any]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",
{
"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("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}")
raisefrom 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 hereimport os
from simstudio import SimStudioClient
# Using environment variables
client = SimStudioClient(
api_key=os.getenv("SIM_API_KEY"),
base_url=os.getenv("SIM_BASE_URL", "https://sim.ai")
)File objects are automatically detected and converted to base64 format. Include them in your input under the field name matching your workflow's API trigger input format:
The SDK converts file objects to this format:
{
'type': 'file',
'data': 'data:mime/type;base64,base64data',
'name': 'filename',
'mime': 'mime/type'
}Alternatively, you can manually provide files using the URL format:
{
'type': 'url',
'data': 'https://example.com/file.pdf',
'name': 'file.pdf',
'mime': 'application/pdf'
}from simstudio import SimStudioClient
import os
client = SimStudioClient(api_key=os.getenv("SIM_API_KEY"))
# Upload a single file - include it under the field name from your API trigger
with open('document.pdf', 'rb') as f:
result = client.execute_workflow(
'workflow-id',
{
'documents': [f], # Must match your workflow's "files" field name
'instructions': 'Analyze this document'
}
)
# Upload multiple files
with open('doc1.pdf', 'rb') as f1, open('doc2.pdf', 'rb') as f2:
result = client.execute_workflow(
'workflow-id',
{
'attachments': [f1, f2], # Must match your workflow's "files" field name
'query': 'Compare these documents'
}
)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, workflow_input 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, workflow_input)
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'}")To run the tests locally:
Clone the repository and navigate to the Python SDK directory:
cd packages/python-sdkCreate and activate a virtual environment:
python3 -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activateInstall the package in development mode with test dependencies:
pip install -e ".[dev]"Run the tests:
pytest tests/ -vRun code quality checks:
# Code formatting
black simstudio/
# Linting
flake8 simstudio/ --max-line-length=100
# Type checking
mypy simstudio/
# Import sorting
isort simstudio/Apache-2.0
| Back | FazBrowse Home | New Git URL |