Context API

Overview

The context proxy provides access to organization, user, and execution information in workflows and data providers. No need to pass it as a parameter—just import and use it.

from bifrost import workflow, context

@workflow(name="example")
async def example(name: str):
    # Access context via proxy - no parameter needed
    org_id = context.org_id
    user_id = context.user_id
    return {"org": org_id, "user": user_id}

Note

The context proxy is read-only and provides identity, organization, and execution metadata. It’s automatically available during workflow execution.

Properties

User Information

Property	Type	Description	Example
user_id	str	Current user’s ID	"user-123"
email	str	User’s email address	"alice@example.com"
name	str	User’s display name	"Alice Smith"

Organization Information

Property	Type	Description	Example
org_id	str | None	Organization ID	"org-456"
org_name	str | None	Organization name	"Acme Corp"
organization	Organization | None	Full organization object	See Organization type
scope	str	Scope identifier	"org-456" or "GLOBAL"

Execution Information

Property	Type	Description	Example
execution_id	str	Unique execution ID	"exec-789"
parameters	dict[str, Any]	Extra parameters passed to workflow	{"custom_key": "value"}
startup	dict[str, Any] | None	Results from launch workflow	{"user_data": {...}}

Authorization Flags

Property	Type	Description
is_platform_admin	bool	User is platform administrator
is_function_key	bool	Called via API key (not user)
is_global_scope	bool	Executing in global scope (no org)

ROI Tracking

Property	Type	Description
roi	ROIContext	ROI tracking context
roi.time_saved	int	Minutes saved (from workflow definition)
roi.value	float	Value metric (from workflow definition)

Methods

Method	Parameters	Returns	Description
get_config(key, default=None)	key: str, default: Any	Any	Async method to get config value

# Using get_config method
api_url = await context.get_config("api_url", default="https://api.example.com")

Computed Properties

org_id

Returns organization ID or None for global scope.

org_id: str | None = context.org_id

if context.org_id:
    # Organization-scoped operation
    users = await get_org_users(context.org_id)
else:
    # Global operation
    all_orgs = await get_all_organizations()

org_name

Returns organization display name.

org_name: str | None = context.org_name

logger.info(f"Executing for {context.org_name}")

is_global_scope

Check if executing in global (platform-wide) scope.

if context.is_global_scope:
    # Platform-level operation
    pass
else:
    # Organization-level operation
    pass

Organization Object

When context.organization is not None:

Property	Type	Description
id	str	Organization ID
name	str	Display name
is_active	bool	Organization is active

if context.organization:
    org_name = context.organization.name
    is_active = context.organization.is_active

Common Patterns

Check Permissions

from bifrost import workflow, context

@workflow(name="admin_task")
async def admin_task():
    if not context.is_platform_admin:
        return {
            "success": False,
            "error": "Admin privileges required"
        }

    # Perform admin operation
    return {"success": True}

Org-Scoped Query

from bifrost import workflow, context

@workflow(name="get_org_data")
async def get_org_data():
    if not context.org_id:
        return {"error": "Organization context required"}

    data = await db.query(
        "SELECT * FROM data WHERE org_id = ?",
        context.org_id
    )

    return {"data": data}

Logging

from bifrost import workflow, context
import logging

logger = logging.getLogger(__name__)

@workflow(name="example")
async def example():
    logger.info("Workflow started", extra={
        "org_id": context.org_id,
        "user_id": context.user_id,
        "execution_id": context.execution_id
    })

Global vs Org

from bifrost import workflow, context

@workflow(name="flexible_query")
async def flexible_query():
    if context.is_global_scope:
        # Platform admin viewing all data
        results = await db.query("SELECT * FROM resources")
    else:
        # Organization user viewing their data
        results = await db.query(
            "SELECT * FROM resources WHERE org_id = ?",
            context.org_id
        )

    return {"results": results}

Auth Type

from bifrost import workflow, context
import logging

logger = logging.getLogger(__name__)

@workflow(name="api_endpoint")
async def api_endpoint():
    if context.is_function_key:
        # Called via API key - automated system
        logger.info("API key authentication")
    else:
        # Called by user - interactive
        logger.info(f"User authentication: {context.email}")

Context in Data Providers

Data providers can access context the same way:

from bifrost import data_provider, context

@data_provider(name="get_org_users")
async def get_org_users():
    # Filter users by organization
    users = await db.query(
        "SELECT id, name FROM users WHERE org_id = ?",
        context.org_id
    )

    return [
        {"label": u["name"], "value": u["id"]}
        for u in users
    ]

Type Definition

from dataclasses import dataclass
from datetime import datetime
from typing import Any

@dataclass
class ROIContext:
    time_saved: int  # Minutes saved per execution
    value: float     # Value metric per execution

@dataclass
class ExecutionContext:
    user_id: str
    email: str
    name: str
    scope: str
    organization: Organization | None
    is_platform_admin: bool
    is_function_key: bool
    execution_id: str
    parameters: dict[str, Any]      # Extra parameters passed to workflow
    startup: dict[str, Any] | None  # Results from launch workflow
    roi: ROIContext                 # ROI tracking context

    @property
    def org_id(self) -> str | None:
        return self.organization.id if self.organization else None

    @property
    def org_name(self) -> str | None:
        return self.organization.name if self.organization else None

    @property
    def is_global_scope(self) -> bool:
        return self.scope == "GLOBAL"

    async def get_config(self, key: str, default: Any = None) -> Any:
        """Get a config value for the current scope."""
        ...

Best Practices

Tip

Follow these patterns for clean, maintainable workflows:

1. Use the Proxy: Import from bifrost import context - no parameter needed
2. Always Check org_id: Verify org context before org-scoped operations
3. Log Context: Include context in log messages for debugging
4. Don’t Mutate: Context is read-only, don’t try to modify it
5. Use Properties: Use context.org_id not context.organization.id
6. Never Log Secrets: Don’t include sensitive data in log messages

Next Steps

• bifrost Module Reference - Full SDK reference
• Decorators Reference - Decorator parameters
• Writing Workflows - Practical guide