Workflows

What Are Workflows?

Workflows are Python functions decorators to describe functionality to the platform. They:

• Define and accept parameters that can be exposed forms or API calls
• Execute business logic
• Return results
• Are discoverable without manual registration

How They Work

from bifrost import workflow, ExecutionContext

@workflow(name="create_user", description="Create new user")
async def create_user(context: ExecutionContext, email: str):
    # Business logic here
    return {"user_id": "123"}

Workflow lifecycle:

1. Developer writes Python function with @workflow decorator
2. Platform discovers it on startup and during execution
3. Workflow appears in UI automatically
4. Admins will usually create forms and tie them to workflows for execution
5. Users execute via forms
6. Results logged and displayed in realtime

Key Concepts

Discovery

Workflows auto-register - no manual configuration needed:

The workspace is configurable, but the Azure deployment will automatically set the location so it can be mounted using Azure Files for persistence.

/workspace
├── user_management.py
├── license_automation.py
└── reporting.py

Platform scans these files, finds @workflow decorators, registers them.

Parameters

Define inputs with @param:

@param("email", type="email", required=True)
@param("department", type="string", data_provider="get_departments")

Parameters automatically generate fields availble to forms with validation.

Execution Context

Every workflow receives ExecutionContext:

• Current user info
• Organization context

Execution Modes

The choice between Async or Sync is something you can avoid thinking about for the most part.

If a workflow is enabled as an endpoint, it will be synchronous by default. This is because you almost always are using this to get an immediate result.

If the workflow is not an endpoint, it will be asynchronous by default. This is because most other interactions will be with forms which will update in realtime as the workflow runs.

Synchronous:

• Runs immediately
• Returns result directly
• Best for quick operations (< 10s)

@workflow(
    name="export_ninjaone_devices_csv",
    execution_mode="sync"
)

Asynchronous (default):

• Queued for background execution
• Returns execution ID immediately
• Best for long-running tasks (> 30s)

@workflow(
    name="export_ninjaone_devices_csv",
    execution_mode="async"
)

Scheduled:

• Runs on cron schedule
• Best for recurring tasks

@workflow(
    name="export_ninjaone_devices_csv",
    schedule="0 9 * * *"
)

Workflow Categories

Organize workflows by purpose, defined completely by the developer.

Multi-Tenancy

Workflows are organization-aware:

• Access organization-specific data via context.org_id or context.organization
• Use organization-scoped secrets and config

@workflow(name="get_org_data")
async def get_org_data(context):
    # Automatically scoped to current org
    data = await db.query(
        "SELECT * FROM data WHERE org_id = ?",
        context.org_id
    )
    return {"data": data}

Workflow vs Form

Workflow: Backend Python function (business logic) Form: Frontend UI (user input collection)

Workflows can be:

• Executed directly via API
• Triggered by forms
• Scheduled to run automatically
• Called from other workflows

Forms always execute workflows, but workflows don’t require forms.

State & Logging

Workflows are stateless but:

• Log progress with Python logging
• Return results
• Use external storage for state
• Capture runtime varialbes for troubleshooting purposes

import logging

logger = logging.getLogger(__name__)

@workflow(name="process_items")
async def process_items(context, items: list):
    logger.info(f"Processing {len(items)} items")

    for i, item in enumerate(items):
        logger.info(f"Processing item {i+1}/{len(items)}")
        # Process item

    return {"processed": len(items)}

Logs appear in execution detail view.

Error Handling

Workflows return error states (don’t raise exceptions):

@workflow(name="example")
async def example(context, param: str):
    try:
        result = await do_work(param)
        return {"success": True, "result": result}
    except Exception as e:
        return {"success": False, "error": str(e)}

This allows:

• Partial success tracking
• User-friendly error messages
• Execution history

Security

Workflows inherit user permissions:

• Run as executing user
• Access org-scoped resources only
• Can require specific permissions via required_permission

@workflow(
    name="admin_task",
    required_permission="canManageOrganization"
)
async def admin_task(context):
    # Only runs if user has permission
    pass

Next Steps

• Build Your First Workflow - Hands-on tutorial
• Writing Workflows Guide - Complete guide
• Platform Overview - Architecture