Decorators Reference

Decorators are the foundation of Bifrost’s discovery system. They register workflows and data providers automatically when your modules are imported.

@workflow Decorator

Registers an async function as a discoverable workflow. All parameters are optional - the decorator infers name and description from your function.

Minimal Usage

from bifrost import workflow

@workflow
async def send_greeting(name: str, count: int = 1):
    """Send a greeting message to the user."""
    return {"message": f"Hello {name}!" * count}

This automatically:

• Sets name to "send_greeting" (from function name)
• Sets description to "Send a greeting message to the user." (from docstring)
• Extracts parameters from the function signature
• Generates a persistent id on first discovery

Syntax

@workflow(
    name: str | None = None,           # Auto-derived from function name
    description: str | None = None,    # Auto-derived from docstring
    category: str = "General",
    tags: list[str] | None = None,
    execution_mode: Literal["sync", "async"] | None = None,  # Auto-selected
    timeout_seconds: int = 1800,
    retry_policy: dict[str, Any] | None = None,
    schedule: str | None = None,
    endpoint_enabled: bool = False,
    allowed_methods: list[str] | None = None,
    disable_global_key: bool = False,
    public_endpoint: bool = False,
    is_tool: bool = False,
    tool_description: str | None = None,
    time_saved: int = 0,
    value: float = 0.0,
)

Parameters

Parameter	Type	Default	Description
name	str	None	Workflow identifier. Auto-derived from function name if not provided
description	str	None	User-facing description. Auto-derived from docstring if not provided
category	str	”General”	Category for grouping (e.g., “User Management”)
tags	list	None	Tags for filtering and search
execution_mode	str	None	”sync” or “async”. Auto-selects based on endpoint_enabled
timeout_seconds	int	1800	Maximum execution time (30 min default, 2 hour max)
retry_policy	dict	None	Retry configuration
schedule	str	None	Cron schedule for scheduled execution
endpoint_enabled	bool	False	Expose as HTTP endpoint at /api/endpoints/{name}
allowed_methods	list	None	HTTP methods allowed (GET, POST, etc.)
disable_global_key	bool	False	Require workflow-specific API key
public_endpoint	bool	False	Skip authentication for webhooks
is_tool	bool	False	Make callable by AI agents
tool_description	str	None	Description for AI agent tool use
time_saved	int	0	Estimated minutes saved per execution (for ROI tracking)
value	float	0.0	Custom value metric per execution

Auto-Generated ID

Each workflow gets a persistent UUID generated by the discovery system. This ID is written back to your source file on first discovery:

@workflow(
    id="550e8400-e29b-41d4-a716-446655440000",  # Auto-generated
    # ... other options
)

The ID ensures workflows maintain identity across renames and deployments.

Execution Mode Auto-Selection

If execution_mode is not specified:

• endpoint_enabled=True → defaults to "sync" (webhooks need immediate response)
• endpoint_enabled=False → defaults to "async" (background tasks scale better)

Examples

Minimal

@workflow
async def create_user(email: str, name: str, department: str = "General"):
    """Create a new user in the system."""
    user = await create_in_db(email, name, department)
    return {"user_id": user.id}

With Options

@workflow(
    category="User Management",
    tags=["onboarding"],
    timeout_seconds=60
)
async def create_user(email: str, name: str):
    """Create a new user in the system."""
    return {"success": True}

Scheduled

@workflow(
    schedule="0 9 * * *"  # Daily at 9 AM UTC
)
async def daily_report():
    """Generate daily report."""
    return {"status": "sent"}

HTTP Endpoint

@workflow(
    endpoint_enabled=True,
    allowed_methods=["POST"],
    public_endpoint=True  # No auth required
)
async def handle_webhook(payload: dict):
    """Handle incoming webhook."""
    # Available at: POST /api/endpoints/handle_webhook
    return {"status": "processed"}

AI Tool

@workflow(
    is_tool=True,
    tool_description="Look up a customer by email address and return their account details"
)
async def lookup_customer(email: str):
    """Look up customer information."""
    customer = await get_customer(email)
    return {"name": customer.name, "account": customer.id}

Parameter Inference

Parameters are automatically extracted from your function signature. No @param decorators needed for basic use cases.

How It Works

1. Type hints determine field types:

   • str → "string"
   • int → "int"
   • float → "float"
   • bool → "bool"
   • list → "list"
   • dict → "json"
   • str | None → "string" (optional)

2. Default values determine optionality:

   • No default → required
   • Has default → optional with that default
   • | None union → optional

3. Labels auto-generated from names:

   • user_name → “User Name”
   • firstName → “First Name”

Example

@workflow
async def onboard_user(
    first_name: str,           # Required string, label: "First Name"
    last_name: str,            # Required string, label: "Last Name"
    email: str,                # Required string, label: "Email"
    license: str = "E3",       # Optional string, default: "E3"
    active: bool = True,       # Optional bool, default: True
    tags: list | None = None   # Optional list
):
    """Onboard a new user."""
    return {"success": True}

Tip

Use from bifrost import context to access execution context - no parameter needed.

@data_provider Decorator

Registers a function that provides dynamic options for form fields.

Syntax

@data_provider(
    name: str | None = None,
    description: str | None = None
)

Parameters

Parameter	Type	Default	Description
name	str	None	Unique provider identifier (auto-derived from function name if not provided)
description	str	None	Description of what data it provides (auto-derived from docstring if not provided)

Examples

Basic

from bifrost import data_provider

@data_provider(
    name="get_departments",
    description="List available departments"
)
async def get_departments():
    """List available departments for selection."""
    return [
        {"label": "Engineering", "value": "eng"},
        {"label": "Sales", "value": "sales"},
        {"label": "Support", "value": "support"}
    ]

With Parameters

@data_provider(
    name="get_users_by_department",
    description="Get users in a specific department"
)
async def get_users_by_department(department: str):
    users = await fetch_users(department)
    return [
        {"label": user["name"], "value": user["id"]}
        for user in users
    ]

With Metadata

@data_provider(name="get_licenses", description="Available licenses")
async def get_licenses():
    return [
        {
            "label": "Microsoft 365 E3",
            "value": "SPE_E3",
            "metadata": {"group": "Microsoft", "seats": 100}
        },
        {
            "label": "Microsoft 365 E5",
            "value": "SPE_E5",
            "metadata": {"group": "Microsoft", "seats": 50}
        }
    ]

Tip

Data provider results are cached automatically. Results are reused until the cache expires, reducing API calls and improving form load times.

Return Values

Workflow

Workflows return any JSON-serializable value:

@workflow
async def example():
    """Example workflow."""
    return {
        "success": True,
        "message": "Operation completed",
        "data": [1, 2, 3]
    }

Data Provider

Data providers return a list of objects with label and value:

@data_provider(name="example", description="Example provider")
async def example():
    return [
        {"label": "Option 1", "value": "opt1"},
        {"label": "Option 2", "value": "opt2"}
    ]

Best Practices

Tip

Follow these guidelines for production-ready workflows:

✅ DO:

• Use descriptive function names (they become the workflow name)
• Write clear docstrings (they become the description)
• Use type hints for all parameters
• Set appropriate timeouts for long operations
• Cache data provider results appropriately

❌ DON’T:

• Use generic names like “workflow1”
• Forget docstrings (empty description in UI)
• Ignore type hints (parameters won’t be extracted)
• Use sync mode for long-running tasks
• Cache dynamic data too long

Complete Example

from bifrost import workflow, data_provider, context
import logging

logger = logging.getLogger(__name__)

# Data provider for departments
@data_provider(
    name="get_departments",
    description="Get available departments"
)
async def get_departments():
    """Fetch departments from system."""
    org_id = context.org_id  # Access context via proxy
    return [
        {"label": "Engineering", "value": "eng"},
        {"label": "Sales", "value": "sales"},
        {"label": "Support", "value": "support"}
    ]

# Workflow using inferred parameters
@workflow(category="User Management", tags=["onboarding"])
async def create_user(
    email: str,
    name: str,
    department: str = "General",
    send_welcome: bool = True
):
    """Create a new user in the system."""
    logger.info(f"Creating user: {email}", extra={
        "org_id": context.org_id,
        "user_id": context.user_id
    })

    user_id = await create_user_in_system(email, name, department)

    if send_welcome:
        await send_welcome_email(email, name)

    logger.info(f"User created: {user_id}")
    return {"user_id": user_id, "email": email}

See Also

• Workflow Development Guide
• Context API Reference
• Discovery System