> ## 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.
# AIUI Backend Injection
> Dependency injection API for wiring custom services into PraisonAIUI
Wire your own backend services into PraisonAIUI using the dependency injection API for custom data sources and business logic.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
subgraph "Backend Injection"
A[🎯 AIUI Pages] --> B[⚙️ set_backend]
B --> C[🔗 Your Services]
B --> D[📊 Usage Sink]
B --> E[🔄 Workflows]
B --> F[🎣 Hooks]
end
classDef ui fill:#8B0000,stroke:#7C90A0,color:#fff
classDef api fill:#F59E0B,stroke:#7C90A0,color:#fff
classDef service fill:#189AB4,stroke:#7C90A0,color:#fff
class A ui
class B api
class C,D,E,F service
```
## Quick Start
Let bridges auto-configure available services:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonai.integration import configure_host, setup_bridges
configure_host(title="My App")
setup_bridges() # Auto-wires usage, workflows, hooks, approvals
```
Replace specific backends with your implementations:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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:
| Key | Expected Type | Purpose | Auto-wired by |
| -------------------- | ----------------------------------- | --------------------------------- | ----------------------------------------- |
| `hooks` | `Callable[[], List[Dict]]` | List UI-friendly hook definitions | `hooks_query.list_hooks_for_api` |
| `workflows` | `Callable[str, dict, dict] -> dict` | Execute YAML workflows | `workflows_service.run_workflow` |
| `usage_sink` | `TokenUsageSinkProtocol` | Persist token usage data | `usage_bridge.register_usage_sink` |
| `usage_query` | `UsageQueryProtocol` | Query usage analytics | `usage_bridge.get_usage_query` |
| `approvals_pending` | `Callable[[], List[Dict]]` | List pending approvals | `approvals_bridge.list_pending_approvals` |
| `approvals_policies` | `Callable[[], List[Dict]]` | List approval policies | `approvals_bridge.get_approval_policies` |
***
## Backend Interfaces
### Hooks Backend
List registered hooks for UI consumption:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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:
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 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!
# ...
```
***
## Related
Main integration module
Auto-registered pages backed by custom backends
L3 dashboard pages can be backed by custom backends via `set_backend()`. For example, set a custom `"workflows"` backend to power the Workflow Runs page with your own data.