logo

Execute and manage workflows easily using our official Python SDK. We are actively working on TypeScript SDK, will launch it soon.

Python SDK

Installation

bash
pip install agentifyme

Initialization

python
from agentifyme import Client

# Initialize client
client = Client(
    api_key="your-api-key",
    endpoint="your-workflow-endpoint"
)

Use local mode to connect to you local development environment. More details here.

python
client = agentifyme.Client(local_mode=True)

Usage Examples

Execute Workflow (Synchronous)

python
# Run workflow and wait for result
result = await client.execute_workflow(
    name="search-candidates",
    parameters={
        "user_query": "Full stack engineer",
        "organization_id": "org-123",
        "filters": {"experience": "5+"} 
    }
)

print(f"Execution completed: {result.data}")

Execute Workflow (Asynchronous)

python
# Start workflow execution
job = await client.start_workflow(
    name="data-processing",
    parameters={
        "file_url": "https://example.com/data.csv",
        "process_type": "analyze"
    }
)

# Check job status later
status = await client.get_job_status(job.id)
if status.is_completed:
    result = status.result

Real-time Events (WebSocket)

python
async def handle_event(event):
    if event.type == "task_completed":
        print(f"Task completed: {event.data}")

# Connect and listen to events
async with client.connect_websocket() as websocket:
    await websocket.subscribe([
        "execution_started",
        "task_completed",
        "execution_completed"
    ])
    
    # Start workflow
    await websocket.execute_workflow(
        name="realtime-processing",
        parameters={"data": "example"},
        event_handler=handle_event
    )

Error Handling

Both SDKs provide consistent error handling:

python
# Python
from agentifyme.exceptions import WorkflowError

try:
    result = await client.execute_workflow(name="example")
except WorkflowError as e:
    print(f"Workflow failed: {e.message}")
except ConnectionError as e:
    print(f"Connection failed: {e}")
typescript
// TypeScript
try {
    const result = await client.executeWorkflow({ name: 'example' });
} catch (error) {
    if (error instanceof WorkflowError) {
        console.error('Workflow failed:', error.message);
    } else {
        console.error('Connection failed:', error);
    }
}

Type Safety

Python Type Hints

python
from typing import TypedDict, Optional

class SearchParams(TypedDict):
    user_query: str
    organization_id: str
    filters: Optional[dict]

# Use with type checking
result = await client.execute_workflow[SearchParams](
    name="search",
    parameters={
        "user_query": "engineer",
        "organization_id": "org-123",
        "filters": None
    }
)

TypeScript Types

typescript
interface WorkflowResult<T> {
    data: T;
    executionTime: number;
}

interface SearchResult {
    candidates: Array<{
        id: string;
        name: string;
        score: number;
    }>;
}

// Use with type checking
const result = await client.executeWorkflow<SearchResult>({
    name: 'search-candidates',
    parameters: searchParams
});

Best Practices

  1. Error Handling

    • Implement proper error handling for all operations
    • Use specific error types for different scenarios
    • Log errors with appropriate context

  2. Resource Management

    • Close WebSocket connections when done
    • Implement timeout handling
    • Clean up resources properly

  3. Performance

    • Reuse client instances
    • Use connection pooling for multiple operations
    • Implement proper retry strategies

  4. Type Safety

    • Use type hints/interfaces for parameters
    • Define response types
    • Enable strict type checking

For complete API reference and advanced usage, visit our SDK documentation.

REST API