HTTP Endpoints

Expose workflows as HTTP endpoints for webhooks, integrations, and external API access.

Basic Endpoint

from bifrost import workflow

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

Endpoint Parameters

Parameter	Type	Default	Description
endpoint_enabled	bool	False	Expose as HTTP endpoint
allowed_methods	list	[“POST”]	HTTP methods to accept
public_endpoint	bool	False	Skip authentication
disable_global_key	bool	False	Require workflow-specific API key

URL Pattern

Endpoints are available at:

POST/GET/etc. /api/endpoints/{workflow_name}

Authentication

Authenticated Endpoint (Default)

Requires API key or session token:

@workflow(
    endpoint_enabled=True,
    allowed_methods=["POST"]
)
async def secure_endpoint(data: dict):
    """Requires authentication."""
    return {"processed": True}

Call with API key:

curl -X POST https://your-instance.com/api/endpoints/secure_endpoint \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"data": "value"}'

Public Endpoint (Webhooks)

No authentication required:

@workflow(
    endpoint_enabled=True,
    allowed_methods=["POST"],
    public_endpoint=True
)
async def github_webhook(payload: dict):
    """Public webhook for GitHub."""
    event = payload.get("action")
    return {"event": event}

Caution

Public endpoints have no authentication. Validate payloads using signatures or other verification.

HTTP Methods

Accept specific HTTP methods:

@workflow(
    endpoint_enabled=True,
    allowed_methods=["GET", "POST"]
)
async def flexible_endpoint(query: str = None, payload: dict = None):
    """Accept GET or POST."""
    return {"query": query, "payload": payload}

Webhook Signature Verification

Verify webhook signatures for security:

import hmac
import hashlib
from bifrost import workflow, config

@workflow(
    endpoint_enabled=True,
    allowed_methods=["POST"],
    public_endpoint=True
)
async def verified_webhook(payload: dict, headers: dict):
    """Webhook with signature verification."""

    secret = await config.get("webhook_secret")
    signature = headers.get("X-Signature-256")

    # Verify HMAC signature
    expected = hmac.new(
        secret.encode(),
        msg=str(payload).encode(),
        digestmod=hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, f"sha256={expected}"):
        return {"error": "Invalid signature"}, 401

    return {"status": "verified"}

Request Data

Access request data via parameters:

@workflow(
    endpoint_enabled=True,
    allowed_methods=["POST"]
)
async def process_request(
    body: dict,           # JSON body
    query_param: str = "" # Query string
):
    """Access request data."""
    return {"body": body, "query": query_param}

Response Formats

Return JSON responses:

@workflow(endpoint_enabled=True)
async def json_response():
    return {
        "status": "success",
        "data": {"id": 123}
    }

Return with status code (tuple):

@workflow(endpoint_enabled=True)
async def custom_status():
    # Return (data, status_code)
    return {"created": True}, 201

Execution Mode

Endpoints default to synchronous execution:

# Sync (immediate response)
@workflow(endpoint_enabled=True)  # execution_mode="sync" implied

# Async (returns execution ID)
@workflow(
    endpoint_enabled=True,
    execution_mode="async"
)
async def long_running_task(data: dict):
    # Returns immediately with execution ID
    # Caller polls for result
    await process_data(data)
    return {"done": True}

Example: Slack Webhook

from bifrost import workflow
import logging

logger = logging.getLogger(__name__)

@workflow(
    endpoint_enabled=True,
    allowed_methods=["POST"],
    public_endpoint=True
)
async def slack_command(payload: dict):
    """Handle Slack slash command."""
    command = payload.get("command")
    text = payload.get("text")
    user = payload.get("user_name")

    logger.info(f"Slack command: {command} from {user}")

    # Slack expects immediate response
    return {
        "response_type": "in_channel",
        "text": f"Processing: {text}"
    }

See Also

• Decorators Reference - All decorator options
• Writing Workflows - Workflow basics