Events

The event system lets you react to workflow execution in real-time.

EventBus

Import from orca.sdk.events:

from orca.sdk.events import EventBus

event_bus = EventBus()

Methods

Method	Description
subscribe(event_name, handler)	Register a handler for an event
unsubscribe(event_name, handler)	Remove a handler
emit(event_name, context)	Emit an event (internal use)

Subscribing to Events

Function Handlers

Simple callable handlers receive event name and context:

def on_thread_completed(event_name, context):
    print(f"Thread {context.thread_name} completed")

event_bus.subscribe("THREAD.COMPLETED", on_thread_completed)

Workflow Event Handlers

Subscribe directly on the workflow:

workflow.add_event_handler("METHOD.COMPLETED", on_method_complete)

Event Names

Events follow the pattern ENTITY.ID.STATUS or ENTITY.STATUS:

Thread Events

Event	When
THREAD.CREATED	Thread instance created
THREAD.MOVING	Thread labware is being transported
THREAD.EXECUTING_ACTION	Action is executing
THREAD.COMPLETED	Thread finished all methods

Method Events

Event	When
METHOD.CREATED	Method instance created
METHOD.IN_PROGRESS	Method is executing
METHOD.COMPLETED	All actions in method done

Action Events

Event	When
ACTION.CREATED	Action created
ACTION.EXECUTING_ACTION	Action is running
ACTION.COMPLETED	Action finished
ACTION.ERRORED	Action failed

Workflow Events

Event	When
WORKFLOW.CREATED	Workflow created
WORKFLOW.IN_PROGRESS	Workflow running
WORKFLOW.COMPLETED	All threads completed
WORKFLOW.ERRORED	Workflow failed

Execution Context

Handlers receive context with execution details. Import context types from orca.sdk.events:

from orca.sdk.events import ExecutionContext, WorkflowExecutionContext, ThreadExecutionContext

WorkflowExecutionContext

@dataclass
class WorkflowExecutionContext:
    workflow_id: str
    workflow_name: str

ThreadExecutionContext

@dataclass
class ThreadExecutionContext(WorkflowExecutionContext):
    thread_id: str
    thread_name: str

The ExecutionContext type is a union of all context types. Handlers receive the appropriate context type based on the event.

SystemBoundEventHandler

For handlers that need system access, extend SystemBoundEventHandler:

from orca.sdk.events import SystemBoundEventHandler

class MyHandler(SystemBoundEventHandler):
    def handle(self, event_name: str, context):
        # Access system via self._system
        device = self._system.get_device("shaker")
        print(f"Event: {event_name}, Device status: {device.is_initialized}")

The system is automatically injected when the workflow runs.

Status Enums

Import status enums for comparison:

from orca.sdk.events import ActionStatus, MethodStatus, LabwareThreadStatus

# ActionStatus values
ActionStatus.CREATED
ActionStatus.EXECUTING_ACTION
ActionStatus.COMPLETED
ActionStatus.ERRORED

# MethodStatus values
MethodStatus.CREATED
MethodStatus.IN_PROGRESS
MethodStatus.COMPLETED

# LabwareThreadStatus values
LabwareThreadStatus.CREATED
LabwareThreadStatus.MOVING
LabwareThreadStatus.EXECUTING_ACTION
LabwareThreadStatus.COMPLETED

Example: Logging Handler

from orca.sdk.events import EventBus

def log_event(event_name, context):
    print(f"[{event_name}] workflow={context.workflow_name}")
    if hasattr(context, 'thread_name'):
        print(f"  thread={context.thread_name}")
    if hasattr(context, 'method_name'):
        print(f"  method={context.method_name}")

event_bus = EventBus()
event_bus.subscribe("THREAD.COMPLETED", log_event)
event_bus.subscribe("METHOD.COMPLETED", log_event)
event_bus.subscribe("WORKFLOW.COMPLETED", log_event)

Next Steps

• Running Workflows - Execute workflows
• Workflows - Add handlers to workflows