Events

Overview

Events in Sample workflows allow you to trigger workflows and control their execution based on external or internal events. This creates a reactive workflow system where workflows can respond to changes in your application state.

Events functionality requires samplehc >= 0.3.0 and workflows-py >= 0.1.18.

Emitting Events

You can emit events from your workflow functions using the SampleHC client. Events consist of a name and an optional payload.

Basic Event Emission

from client_manager import SampleHealthcareClient

def emit_order_updated(ctx: WorkflowRunContext):
    """Handle the update action - triggers eligibility if status is new"""
    order = get_order(ctx)
    client = get_client(ctx)
    
    client.v2.events.emit(
        name="order-updated",
        payload={
            "order_id": order["order_id"], 
            "bill_type": get_bill_type(ctx)
        },
    )

Event Structure

Events have the following structure:

name: A string identifier for the event type (e.g., “order-updated”, “order-created”)
payload: Any JSON-serializable data relevant to the event

Listening to Events

Workflows can listen to events using two mechanisms: start_on and cancel_on.

Starting Workflows on Events

Use start_on to automatically start a workflow when specific events are received:

from workflows_py.workflow import Workflow

workflow = Workflow(
    get_display_col_values={"Order ID": lambda ctx: ctx.get_start_data()["order_id"]},
    start_on={
        "order-created": lambda ctx: create_insurance_verification_start_data_from_start_on_ctx(ctx),
        "order-updated": lambda ctx: create_insurance_verification_start_data_from_start_on_ctx(ctx),
    },
)

Start Handler Return Values

The start_on handler function can return:

An object with optional start_data field: Starts the workflow with the provided start data
None: Does not start the workflow

def handle_order_created(ctx):
    # Start the workflow with custom data
    return {
        "start_data": {
            "order_id": ctx.event.payload["order_id"],
            "priority": "high"
        }
    }

def handle_low_priority_order(ctx):
    # Don't start workflow for low priority orders
    if ctx.event.payload.get("priority") == "low":
        return None
    
    # Start workflow for other priorities
    return {"start_data": {"order_id": ctx.event.payload["order_id"]}}

# These handlers would be used in your workflow configuration like this:
workflow = Workflow(
    get_display_col_values={"Order ID": lambda ctx: ctx.get_start_data()["order_id"]},
    start_on={
        "order-created": handle_order_created,
        "order-updated": handle_low_priority_order,
    },
)

Canceling Workflows on Events

Use cancel_on to automatically cancel running workflows when specific events are received:

workflow = Workflow(
    # ... other configuration
    cancel_on={
        "order-updated": lambda ctx: ctx.get_start_data()["order_id"] == ctx.event.payload.get("order_id"),
    },
)

Cancel Handler Return Values

The cancel_on handler function should return:

True: Cancel the workflow run
False: Keep the workflow running

def should_cancel_on_order_update(ctx):
    # Cancel if this workflow is processing the same order
    workflow_order_id = ctx.get_start_data()["order_id"]
    event_order_id = ctx.event.payload.get("order_id")
    
    return workflow_order_id == event_order_id

Event Context

When handling events, you have access to an event context that provides:

StartOnContext

Available in start_on handlers:

def handle_start_event(ctx: StartOnContext):
    # Access the event that triggered this handler
    event_name = ctx.event.name
    event_payload = ctx.event.payload
    
    # Access authentication token (for API calls)
    backend_token = ctx.backend_token
    
    # Return start data or None
    return {"start_data": {"triggered_by": event_name}}

CancelOnContext

Available in cancel_on handlers:

def handle_cancel_event(ctx: CancelOnContext):
    # Access the event that triggered this handler
    event_name = ctx.event.name
    event_payload = ctx.event.payload
    
    # Access the workflow's original start data
    start_data = ctx.get_start_data()
    
    # Return boolean decision
    return should_cancel_workflow(start_data, event_payload)

Common Patterns

Order Processing Workflow

workflow = Workflow(
    get_display_col_values={
        "Order ID": lambda ctx: ctx.get_start_data()["order_id"],
        "Status": lambda ctx: ctx.get_start_data().get("status", "pending")
    },
    start_on={
        "order-created": lambda ctx: {
            "start_data": {
                "order_id": ctx.event.payload["order_id"],
                "status": "processing"
            }
        }
    },
    cancel_on={
        "order-cancelled": lambda ctx: (
            ctx.get_start_data()["order_id"] == ctx.event.payload["order_id"]
        )
    }
)

Multi-Event Workflow

workflow = Workflow(
    start_on={
        "patient-admitted": lambda ctx: create_admission_workflow_data(ctx),
        "insurance-verified": lambda ctx: create_insurance_workflow_data(ctx),
        "urgent-case": lambda ctx: create_urgent_workflow_data(ctx),
    },
    cancel_on={
        "patient-discharged": lambda ctx: (
            ctx.get_start_data()["patient_id"] == ctx.event.payload["patient_id"]
        ),
        "case-closed": lambda ctx: (
            ctx.get_start_data()["case_id"] == ctx.event.payload["case_id"]
        ),
    }
)

Best Practices

Event Naming

Use descriptive, consistent event names (e.g., “order-created”, “patient-admitted”)
Include entity type and action (e.g., “user-registered”, “document-processed”)
Use kebab-case for event names

Payload Design

Include essential identifiers in the payload (IDs, timestamps)
Keep payloads lightweight - avoid large objects
Include context needed for decision making

Error Handling

def safe_event_handler(ctx):
    try:
        # Your event handling logic
        return {"start_data": process_event(ctx.event)}
    except Exception as e:
        # Log error and decide whether to start workflow
        logger.error(f"Error processing event: {e}")
        return None  # Don't start workflow on error

Performance Considerations

Keep event handlers lightweight and fast
Avoid heavy computations in start_on and cancel_on handlers

Browser Agents

Runs

Help Request

Event name: sample:browser-agent:help-request
Description: Emitted by a running browser agent when it needs help from a human operator (for example, to resolve ambiguity or request manual intervention).
Payload structure:
- browserAgentRunId: The ID of the browser agent run that requested help
- browserAgentSlug: The slug identifier of the browser agent
- request: The help request message from the browser agent

Example payload

{
  "browserAgentRunId": "bar_1234567890abcdef",
  "browserAgentSlug": "my-browser-agent",
  "request": "Please confirm the last step was completed successfully."
}

Typical uses

Start or unlock a workflow step awaiting human input from a browser agent.
Notify an operations dashboard that a specific browser-agent run needs assistance.

Listening from a workflow

from workflows_py.workflow import Workflow

def handle_help_request(ctx):
    # Extract all relevant information from the event
    browser_agent_run_id = ctx.event.payload.get("browserAgentRunId")
    browser_agent_slug = ctx.event.payload.get("browserAgentSlug")
    help_text = ctx.event.payload.get("request")
    
    # Return start data with all context
    return {
        "start_data": {
            "browserAgentRunId": browser_agent_run_id,
            "browserAgentSlug": browser_agent_slug,
            "helpRequest": help_text
        }
    }

workflow = Workflow(
    get_display_col_values={
        "Agent": lambda ctx: ctx.get_start_data()["browserAgentSlug"],
        "Request": lambda ctx: ctx.get_start_data()["helpRequest"]
    },
    start_on={
        # Subscribe to all browser-agent help requests
        "sample:browser-agent:help-request": handle_help_request,
    },
)

Resolving Help Requests

Once a help request has been addressed, you can resolve it using the browser agents API:

from client_manager import SampleHealthcareClient

def resolve_help_request(ctx: WorkflowRunContext):
    client = get_client(ctx)
    start_data = ctx.get_start_data()
    
    # Resolve the help request
    client.v2.browser_agents.resolve_help_request(
        slug=start_data["browserAgentSlug"],
        browser_agent_run_id=start_data["browserAgentRunId"],
        help_request_id=start_data["helpRequestId"],  # You'll need to track this
        resolution="User confirmed the step was completed successfully"
    )

Use specific event matching to avoid unnecessary workflow starts/cancellations

Sample System Emitted Events

Sample automatically emits certain events when specific actions occur in your system. These events are prefixed with sample: and can be used to trigger workflows automatically.

Email

All email-related events are emitted under the sample:email:* pattern.

received

When an email is sent to your sample mailbox ([email protected]), Sample automatically emits a sample:email:received event. Event Name: sample:email:received Payload Structure:

{
  "subject": "Email subject line",
  "body": "Plain text email body",
  "html": "HTML email body (if available)",
  "from": ["[email protected]"],
  "to": ["[email protected]"],
  "attachments": [
    {
      "fileName": "document.pdf",
      "id": "file-metadata-id"
    }
  ]
}

Example Usage:

from workflows_py.workflow import Workflow

workflow = Workflow(
    get_display_col_values={
        "Subject": lambda ctx: ctx.get_start_data()["subject"],
        "From": lambda ctx: ", ".join(ctx.get_start_data()["from"])
    },
    start_on={
        "sample:email:received": lambda ctx: {
            "start_data": {
                "subject": ctx.event.payload["subject"],
                "body": ctx.event.payload["body"],
                "from": ctx.event.payload["from"],
                "attachments": ctx.event.payload["attachments"]
            }
        }
    }
)

Kno2

All Kno2-related events are emitted under the sample:kno2:* pattern when messages are received through your Kno2 integration.

message-receive

When a message (such as a fax) is received through your Kno2 connection, Sample automatically emits a sample:kno2:${connection.slug}:message-receive event, where ${connection.slug} corresponds to the specific Kno2 connection that received the message. Event Name: sample:kno2:${connection.slug}:message-receive Payload Structure:

{
  "message": {
    "id": "msg_abc123def456",
    "body": null,
    "type": "Intake",
    "isNew": false,
    "origin": "eFax",
    "status": "Received",
    "isDraft": false,
    "patient": {
      "fullName": "John Doe",
      "lastName": "Doe",
      "firstName": "John",
      "middleName": null,
      "birthDate": null,
      "gender": null,
      "patientId": null,
      "patientIds": [],
      "telephone": null,
      "streetAddress1": null,
      "streetAddress2": null,
      "city": null,
      "state": null,
      "postalCode": null,
      "country": null,
      "telecom": []
    },
    "subject": "Medical Records Request",
    "isUrgent": false,
    "priority": "NotUrgent",
    "toAddress": "555-123-4567",
    "fromAddress": "555-987-6543",
    "properties": {
      "faxPageCount": "4",
      "originalFaxPageCount": "4"
    },
    "sourceType": "Fax",
    "attachments": [
      {
        "id": "att_xyz789",
        "key": "attachment-key-123",
        "fileName": "medical_records.pdf",
        "mimeType": "application/pdf",
        "messageId": "msg_abc123def456",
        "sizeInBytes": 207212,
        "documentType": "",
        "attachmentMeta": {
          "documentDate": "2025-01-15T10:30:00.000",
          "documentType": "",
          "confidentiality": "Normal"
        }
      }
    ],
    "createdDate": "2025-01-15T10:30:00.000",
    "messageDate": "2025-01-15T10:29:45.000",
    "messageType": "default",
    "patientName": "John Doe",
    "organizationId": "org_sample123"
  },
  "attachments": [
    {
      "id": "fmd_abc123",
      "fileName": "medical_records.pdf"
    }
  ]
}

Key Fields:

message.id: Unique identifier for the message
message.subject: Subject line of the fax/message
message.toAddress: Recipient fax number
message.fromAddress: Sender fax number
message.sourceType: Type of message (e.g., “Fax”)
message.attachments: Array of attachment metadata from Kno2
attachments: Array of Sample file metadata IDs for downloaded attachments

Example Usage:

from workflows_py.workflow import Workflow

workflow = Workflow(
    get_display_col_values={
        "From": lambda ctx: ctx.get_start_data()["message"]["fromAddress"],
        "Subject": lambda ctx: ctx.get_start_data()["message"]["subject"],
        "Pages": lambda ctx: ctx.get_start_data()["message"]["properties"]["faxPageCount"]
    },
    start_on={
        "sample:kno2:production:message-receive": lambda ctx: {
            "start_data": {
                "message": ctx.event.payload["message"],
                "attachments": ctx.event.payload["attachments"],
                "from_fax": ctx.event.payload["message"]["fromAddress"],
                "to_fax": ctx.event.payload["message"]["toAddress"]
            }
        }
    }
)

# Process the received fax attachments
def process_kno2_fax(ctx: WorkflowRunContext):
    attachments = ctx.get_start_data()["attachments"]
    
    for attachment in attachments:
        # Process each attachment using its file metadata ID
        file_id = attachment["id"]
        file_name = attachment["fileName"]
        # Your processing logic here

# To listen to messages from multiple Kno2 connections:
workflow = Workflow(
    start_on={
        "sample:kno2:production:message-receive": lambda ctx: process_kno2_message(ctx),
        "sample:kno2:staging:message-receive": lambda ctx: process_kno2_message(ctx),
    }
)

Troubleshooting

Common Issues

Workflow not starting on events:

Verify event names match exactly between emit and listen
Check that handler functions return appropriate values
Ensure version requirements are met

Unexpected workflow cancellations:

Review cancel_on handler logic
Check event payload structure
Verify cancellation conditions are specific enough

Missing event data:

Ensure event payload contains required fields
Check that get_start_data() returns expected structure
Verify event emission includes necessary context

Workflows

Screens

Other guides

Overview

Emitting Events

Basic Event Emission

Event Structure

Listening to Events

Starting Workflows on Events

Start Handler Return Values

Canceling Workflows on Events

Cancel Handler Return Values

Event Context

StartOnContext

CancelOnContext

Common Patterns

Order Processing Workflow

Multi-Event Workflow

Best Practices

Event Naming

Payload Design

Error Handling

Performance Considerations

Browser Agents

Runs

Help Request

Resolving Help Requests

Sample System Emitted Events

Email

received

Kno2

message-receive

Troubleshooting

Common Issues

Workflows

Screens

Other guides

​Overview

​Emitting Events

​Basic Event Emission

​Event Structure

​Listening to Events

​Starting Workflows on Events

​Start Handler Return Values

​Canceling Workflows on Events

​Cancel Handler Return Values

​Event Context

​StartOnContext

​CancelOnContext

​Common Patterns

​Order Processing Workflow

​Multi-Event Workflow

​Best Practices

​Event Naming

​Payload Design

​Error Handling

​Performance Considerations

​Browser Agents

​Runs

​Help Request

​Resolving Help Requests

​Sample System Emitted Events

​Email

​received

​Kno2

​message-receive

​Troubleshooting

​Common Issues

Overview

Emitting Events

Basic Event Emission

Event Structure

Listening to Events

Starting Workflows on Events

Start Handler Return Values

Canceling Workflows on Events

Cancel Handler Return Values

Event Context

StartOnContext

CancelOnContext

Common Patterns

Order Processing Workflow

Multi-Event Workflow

Best Practices

Event Naming

Payload Design

Error Handling

Performance Considerations

Browser Agents

Runs

Help Request

Resolving Help Requests

Sample System Emitted Events

Email

received

Kno2

message-receive

Troubleshooting

Common Issues