Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.praison.ai/llms.txt

Use this file to discover all available pages before exploring further.

Wire your own backend services into PraisonAIUI using the dependency injection API for custom data sources and business logic.

​
Quick Start

1

Auto-wire Default Backends

Let bridges auto-configure available services:
from praisonai.integration import configure_host, setup_bridges

configure_host(title="My App")
setup_bridges()  # Auto-wires usage, workflows, hooks, approvals
2

Custom Backend Override

Replace specific backends with your implementations:
import praisonaiui.backends as backends

# Override usage tracking
backends.set_backend("usage_sink", my_database_sink)
backends.set_backend("usage_query", my_analytics_query)

# Override workflow execution
backends.set_backend("workflows", my_workflow_runner)

​
Available Backend Keys

The injection API supports these standard backend keys:
KeyExpected TypePurposeAuto-wired by
hooksCallable[[], List[Dict]]List UI-friendly hook definitionshooks_query.list_hooks_for_api
workflowsCallable[str, dict, dict] -> dictExecute YAML workflowsworkflows_service.run_workflow
usage_sinkTokenUsageSinkProtocolPersist token usage datausage_bridge.register_usage_sink
usage_queryUsageQueryProtocolQuery usage analyticsusage_bridge.get_usage_query
approvals_pendingCallable[[], List[Dict]]List pending approvalsapprovals_bridge.list_pending_approvals
approvals_policiesCallable[[], List[Dict]]List approval policiesapprovals_bridge.get_approval_policies

​
Backend Interfaces

​
Hooks Backend

List registered hooks for UI consumption: 
def list_hooks_for_api() -> List[Dict[str, Any]]:
    """Return hooks in UI-friendly format."""
    return [
        {
            "name": "pre_task_hook",
            "event": "task.start",
            "description": "Validates task inputs",
            "enabled": True
        }
    ]

backends.set_backend("hooks", list_hooks_for_api)

​
Workflows Backend

Execute YAML-driven workflows: 
def execute_workflow(workflow_id: str, workflow: dict, input_data: dict) -> dict:
    """Execute workflow and return run record."""
    text = input_data.get("text", input_data.get("message", ""))
    
    # Your workflow execution logic
    result = my_workflow_engine.run(workflow_id, text, workflow)
    
    return {
        "id": result.run_id,
        "workflow_id": workflow_id,
        "status": result.status,
        "input": {"text": text},
        "output": result.output,
        "error": result.error,
        "created_at": result.timestamp
    }

backends.set_backend("workflows", execute_workflow)

​
Usage Tracking Backends

Wire custom analytics services: 
from praisonaiagents.telemetry.protocols import TokenUsageSinkProtocol, UsageQueryProtocol

class MyAnalyticsSink:
    def persist(self, task_id: str, agent_name: str, model: str, 
               metrics: Any, metadata: dict = None) -> None:
        # Send to analytics service
        analytics.track_usage(task_id, agent_name, model, 
                             input_tokens=metrics.input_tokens,
                             output_tokens=metrics.output_tokens)

class MyAnalyticsQuery:
    def get_summary(self) -> dict:
        return analytics.get_usage_summary()
    
    def list_recent(self, limit: int = 50) -> list:
        return analytics.get_recent_usage(limit)

backends.set_backend("usage_sink", MyAnalyticsSink())
backends.set_backend("usage_query", MyAnalyticsQuery())

​
Approvals Backends

Integrate with approval workflow systems: 
def list_pending_approvals() -> List[Dict[str, Any]]:
    """List approvals awaiting human review."""
    pending = approval_system.get_pending()
    return [
        {
            "id": item.id,
            "type": item.type,
            "description": item.description,
            "requester": item.agent_name,
            "created_at": item.timestamp,
            "metadata": item.data
        }
        for item in pending
    ]

def get_approval_policies() -> List[Dict[str, Any]]:
    """List configured approval policies."""
    return [
        {
            "name": "high_cost_operations",
            "description": "Require approval for operations >$10",
            "conditions": ["cost > 10.0"],
            "approvers": ["admin", "finance"]
        }
    ]

backends.set_backend("approvals_pending", list_pending_approvals)
backends.set_backend("approvals_policies", get_approval_policies)

​
Auto-wired Bridges

The setup_bridges() function automatically wires available services: 
from praisonai.integration.bridges import (
    usage_bridge,
    hooks_query, 
    workflows_service,
    approvals_bridge,
    schedules_runner
)

# Auto-wired when available:
sink = usage_bridge.register_usage_sink()        # -> "usage_sink"
query = usage_bridge.get_usage_query()           # -> "usage_query"  
workflows_service.run_workflow                   # -> "workflows"
hooks_query.list_hooks_for_api                  # -> "hooks"
approvals_bridge.list_pending_approvals         # -> "approvals_pending"
approvals_bridge.get_approval_policies          # -> "approvals_policies"

# Plus schedule runner (separate system)
schedules_runner.ensure_schedule_runner()        # Starts daemon if needed

​
Custom Integration Example

Complete example integrating with external systems: 
import praisonaiui.backends as backends
from praisonai.integration import configure_host, create_host_app

# External service clients
class WorkflowEngine:
    def execute(self, workflow_id: str, input_text: str) -> dict:
        # Call external workflow service
        response = requests.post(f"{WORKFLOW_API}/execute", json={
            "workflow_id": workflow_id,
            "input": input_text
        })
        return response.json()

class AuditService:
    def log_usage(self, metrics: dict) -> None:
        # Send to audit/compliance system
        requests.post(f"{AUDIT_API}/usage", json=metrics)

# Custom backends
class ExternalWorkflowBackend:
    def __init__(self):
        self.engine = WorkflowEngine()
    
    def __call__(self, workflow_id: str, workflow: dict, input_data: dict) -> dict:
        text = input_data.get("text", "")
        result = self.engine.execute(workflow_id, text)
        
        return {
            "id": result["run_id"],
            "workflow_id": workflow_id,
            "status": result["status"],
            "output": result.get("output"),
            "error": result.get("error")
        }

class AuditUsageSink:
    def __init__(self):
        self.audit = AuditService()
    
    def persist(self, task_id: str, agent_name: str, model: str, 
               metrics: Any, metadata: dict = None) -> None:
        self.audit.log_usage({
            "task_id": task_id,
            "agent": agent_name,
            "model": model,
            "tokens": getattr(metrics, 'total_tokens', 0),
            "cost": getattr(metrics, 'cost', 0.0),
            "timestamp": datetime.now().isoformat()
        })

# Wire custom backends
configure_host(title="External Integration Demo")

backends.set_backend("workflows", ExternalWorkflowBackend())
backends.set_backend("usage_sink", AuditUsageSink())

# Other backends auto-wired by setup_bridges()
from praisonai.integration import setup_bridges
setup_bridges()

app = create_host_app()

​
Backend Discovery

Check what backends are currently registered: 
import praisonaiui.backends as backends

# List all registered backends
current_backends = backends.get_all_backends()
print(current_backends.keys())
# ['hooks', 'workflows', 'usage_sink', 'usage_query']

# Check specific backend
if backends.has_backend("usage_sink"):
    sink = backends.get_backend("usage_sink")
    print(f"Usage sink: {type(sink).__name__}")

# Clear a backend
backends.clear_backend("workflows")

# Clear all backends
backends.clear_all_backends()

​
Error Handling

Backends should handle errors gracefully: 
class RobustWorkflowBackend:
    def __call__(self, workflow_id: str, workflow: dict, input_data: dict) -> dict:
        try:
            result = self.execute_workflow(workflow_id, workflow, input_data)
            return result
        except ExternalServiceError as e:
            # Return error format expected by UI
            return {
                "id": f"failed-{uuid.uuid4()}",
                "workflow_id": workflow_id,
                "status": "failed",
                "error": f"External service error: {str(e)}",
                "created_at": datetime.now().isoformat()
            }
        except Exception as e:
            # Log unexpected errors but don't crash UI
            logger.exception("Workflow backend error")
            return {
                "id": f"error-{uuid.uuid4()}",
                "workflow_id": workflow_id,
                "status": "error", 
                "error": "Internal error occurred",
                "created_at": datetime.now().isoformat()
            }

​
Best Practices

Let auto-wiring handle standard cases, override specific backends:
from praisonai.integration import setup_bridges
import praisonaiui.backends as backends

setup_bridges()  # Auto-wire available services
backends.set_backend("usage_sink", MyCustomSink())  # Override specific
Handle missing external services gracefully:
class FallbackWorkflowBackend:
    def __call__(self, workflow_id, workflow, input_data):
        try:
            return self.external_service.execute(workflow_id, input_data)
        except ConnectionError:
            # Fallback to local execution
            return self.local_executor.run(workflow, input_data)
Backends are called frequently - keep them fast and stateless:
# Good - lightweight, cached connection
class CachedDatabaseSink:
    def __init__(self):
        self._connection = None
    
    @property
    def connection(self):
        if self._connection is None:
            self._connection = create_db_connection()
        return self._connection

# Avoid - heavy initialization on every call
class BadDatabaseSink:
    def persist(self, ...):
        connection = create_new_db_connection()  # Expensive!
        # ...

Host Integration

Main integration module

Token Usage Protocol

Usage tracking backends