Write Workflows

Basic Structure

Every workflow follows this pattern:

from bifrost import workflow
import logging

logger = logging.getLogger(__name__)

@workflow(category="Category Name")
async def my_workflow(param1: str, param2: int = 10):
    """Docstring explaining workflow purpose - this becomes the description."""
    logger.info(f"Processing {param1}")
    return {"result": "success"}

The decorator automatically infers:

• name: from the function name (my_workflow)
• description: from the docstring
• parameters: from the function signature with type hints

Note

Writing the file is only step one. You must also register the function before it appears in the platform. See Workflow Registration for details.

Decorator Reference

@workflow Options

The decorator only accepts identity metadata. Anything that changes runtime behavior belongs on the workflow record in the UI or API.

Option	Purpose
name	Override the function name
description	Override the first docstring line
category	Group workflows in the UI
tags	Filter and search hints
is_tool	Mark the workflow as AI-callable

Unknown keyword arguments are still accepted for backwards compatibility, but Bifrost warns and ignores them.

Workflow timeout, endpoint exposure, allowed methods, public access, cache TTL, retry policy, and access level are configured after discovery through the UI or API.

Parameter Inference

Parameters are automatically extracted from your function signature:

@workflow
async def create_user(
    email: str,              # Required string, label: "Email"
    name: str,               # Required string, label: "Name"
    department: str = "IT",  # Optional with default
    active: bool = True      # Optional boolean
):
    """Create a new user."""
    pass

@param for Advanced Features

Use @param only when you need data providers, validation, or help text:

@workflow
@param("department", data_provider="get_departments")
@param("email", validation={"pattern": r".*@company\.com$"}, help_text="Must be company email")
async def create_user(email: str, department: str):
    """Create a new user."""
    pass

Validation Rules

# String validation
@param("username", validation={
    "min_length": 3,
    "max_length": 50,
    "pattern": r"^[a-zA-Z0-9_]+$"
})

# Number validation
@param("quantity", validation={
    "min": 1,
    "max": 1000
})

# Enum options
@param("status", validation={
    "enum": ["active", "inactive", "pending"]
})

Execution Context

Access organization, user, and execution metadata via the context proxy:

from bifrost import context

@workflow
async def my_workflow(name: str):
    # Organization
    org_id = context.org_id
    org_name = context.org_name

    # User
    user_id = context.user_id
    email = context.email
    user_name = context.name

    # Execution
    execution_id = context.execution_id
    is_admin = context.is_platform_admin
    is_global = context.is_global_scope

SDK Modules

Access config, OAuth, and files via SDK:

from bifrost import config, oauth, files

async def my_workflow():
    """Example workflow using SDK modules."""
    # Configuration (async)
    api_url = await config.get("api_url", default="https://api.example.com")
    await config.set("api_url", "https://api.example.com")

    # Secrets (stored encrypted in database)
    api_key = await config.get("api_key")
    await config.set("api_key", "secret_value", is_secret=True)

    # OAuth connection (async)
    conn = await oauth.get("microsoft")
    if conn:
        access_token = conn["access_token"]
        # Use access_token for API calls

    # File operations (synchronous)
    files.write("data/output.txt", "content", location="workspace")
    content = files.read("data/output.txt", location="workspace")

Tip

SDK Methods: config and oauth are async and require await. File operations are synchronous. Use is_secret=True for encrypted storage.

Invocation and scheduling

How a workflow runs is determined by the place you invoke it from, not by decorator flags:

• Forms and API calls execute the registered workflow directly.
• HTTP exposure is turned on in the workflow settings, where you also choose methods and auth behavior.
• Recurring runs are modeled as schedule event sources, which attach subscriptions to a workflow.

See HTTP Endpoints, Scheduled Workflows, and Events for the current control surfaces.

Data Providers

Provide dynamic options for dropdowns:

1. Create a data provider:

   from bifrost import data_provider
   
   @data_provider(
       name="get_departments",
       description="List departments"
   )
   async def get_departments():
       return [
           {"label": "Engineering", "value": "eng"},
           {"label": "Sales", "value": "sales"}
       ]

2. Use in workflow with @param:

   @workflow
   @param("department", data_provider="get_departments")
   async def assign_user(department: str, user_email: str):
       """Assign user to department."""
       # department contains selected value ("eng" or "sales")
       return {"assigned_to": department}

Return Values

Always return a dictionary:

Success

return {"success": True, "result": data}

Error

# Workflow still succeeds, but indicates failure
return {"success": False, "error": "User not found"}

Nested Data

return {
    "user": {"id": "123", "email": "user@example.com"},
    "items": [1, 2, 3]
}

Error Handling

Exceptions are automatically caught and handled by the execution engine. You can optionally return {"success": False} to indicate partial failures:

import logging
from bifrost import workflow

logger = logging.getLogger(__name__)

@workflow
async def create_user(email: str, name: str):
    """Create a new user in the system."""
    # Raised exceptions are automatically caught and logged
    user = await create_user_in_system(email, name)
    logger.info(f"Created user: {user.id}")
    return {"user_id": user.id}

    # Or indicate partial failure (execution status: COMPLETED_WITH_ERRORS)
    # return {"success": False, "error": "User created but email failed"}

Note

Exception Handling: Raised exceptions automatically mark the execution as FAILED. Returning {"success": False} marks it as COMPLETED_WITH_ERRORS.

Logging

Use Python’s logging module for visibility into execution:

import logging
from bifrost import workflow

logger = logging.getLogger(__name__)

@workflow
async def process_data(items: list):
    """Process a list of data items."""
    logger.debug("Starting detailed processing...")  # Hidden from users
    logger.info(f"Processing {len(items)} items...")  # Visible to users
    logger.warning("Item 5 failed, continuing...")   # Visible to users
    logger.error("Critical failure detected")        # Visible to users

Tip

Log Visibility: logger.debug() is only visible to developers. All other levels (info, warning, error) are visible to users who execute workflows from forms.

Best Practices

• Single Responsibility: One workflow, one task. Keep workflows focused and composable.
• Validate Early: Check inputs before processing to fail fast and provide clear feedback.
• Log for Users: Use logger.info() for user-facing progress updates. Use logger.debug() for developer debugging.
• Let Exceptions Bubble: Raised exceptions are automatically handled - no need to catch and return error dicts.
• Use Type Hints: Enable IDE autocomplete and error detection with proper typing.
• Avoid Secrets in Returns: Never return credentials or PII in workflow results.

Common Patterns

List Processing

async def process_items(context: ExecutionContext, items: list):
    results = []
    errors = []

    for item in items:
        try:
            result = await process_item(item)
            results.append(result)
        except Exception as e:
            errors.append({"item": item, "error": str(e)})

    return {
        "processed": len(results),
        "failed": len(errors),
        "results": results,
        "errors": errors
    }

Chained Operations

async def create_user_with_license(context: ExecutionContext, email: str, sku: str):
    # Step 1
    logger.info("Creating user...")
    user = await create_user(email)

    # Step 2
    logger.info("Assigning license...")
    await assign_license(user.id, sku)

    return {"user_id": user.id, "license": sku}

Next Steps

• Using Decorators - Advanced decorator features
• Error Handling - Comprehensive error patterns
• SDK Reference - Full API documentation