Documentation Index
Fetch the complete documentation index at: https://mintlify.com/MotiaDev/motia/llms.txt
Use this file to discover all available pages before exploring further.
Overview
FlowContext is passed as the second argument to all step handlers. It provides access to enqueue functionality, state management, streams, logging, and trigger information.
Class: FlowContext[TEnqueueData]
from motia import FlowContext
async def handler(input_data, ctx: FlowContext):
# Access context properties and methods
await ctx.enqueue({"topic": "next", "data": input_data})
TEnqueueData represents the type of data being enqueued.
Properties
enqueue
Callable[[Any], Awaitable[None]]
Function to enqueue events to other steps. Takes event data and returns a Promise.
Unique identifier for tracing this execution across steps.
State manager for persisting and retrieving scoped state. See StateManager. Logger instance for structured logging.
Dictionary of available streams, keyed by stream name.
Information about the trigger that fired this handler.
The raw input value received by the handler.
Methods
is_queue
def is_queue(self) -> bool
True if the trigger is a queue trigger.
is_api
Returns True if the trigger is an HTTP/API request.
is_cron
def is_cron(self) -> bool
True if the trigger is a cron event.
is_state
def is_state(self) -> bool
True if the trigger is a state change.
is_stream
def is_stream(self) -> bool
True if the trigger is a stream event.
get_data
def get_data(self) -> Any
- For HTTP triggers: returns
request.body
- For queue triggers: returns the queue data directly
- For cron triggers: returns
None
- For state/stream triggers: returns the input value
match
async def match(self, handlers: dict[str, Callable]) -> Any
handlers
dict[str, Callable]
required
Dictionary mapping trigger types to handler functions:
"queue": async (input) -> Any"http" or "api": async (request) -> ApiResponse"cron": async () -> Any"state": async (input) -> Any"stream": async (input) -> Any"default": async (input) -> Any (fallback)
The result from the matched handler.
Class: StateManager
Manages scoped state for your application.
Methods
get
async def get(self, scope: str, key: str) -> Any | None
The scope/group identifier.
The item key within the scope.
The stored value, or None if not found.
set
async def set(self, scope: str, key: str, value: Any) -> Any
The scope/group identifier.
The item key within the scope.
update
async def update(self, scope: str, key: str, ops: list[dict[str, Any]]) -> Any
The scope/group identifier.
The item key within the scope.
ops
list[dict[str, Any]]
required
List of update operations to apply.
delete
async def delete(self, scope: str, key: str) -> Any | None
The scope/group identifier.
The deleted value, or None if not found.
list
async def list(self, scope: str) -> list[Any]
The scope/group identifier.
List of all items in the scope.
list_groups
async def list_groups(self) -> list[str]
List of all scope identifiers.
clear
async def clear(self, scope: str) -> None
The scope/group identifier to clear.
Class: Stream[TData]
Manages distributed stream state.
Methods
get
async def get(self, group_id: str, item_id: str) -> TData | None
The stream item, or None if not found.
set
async def set(self, group_id: str, item_id: str, data: TData) -> Any
Result of the set operation.
delete
async def delete(self, group_id: str, item_id: str) -> None
list
async def list(self, group_id: str) -> list[TData]
List of all items in the group.
update
async def update(self, group_id: str, item_id: str, ops: list[dict[str, Any]]) -> Any
list_groups
async def list_groups(self) -> list[str]
Type: TriggerInfo
Information about the trigger that fired.
Fields
type
'http' | 'queue' | 'cron' | 'state' | 'stream'
The trigger type.
Index of the trigger in the step’s trigger list.
HTTP path (API triggers only).
HTTP method (API triggers only).
Queue topic (queue triggers only).
Cron expression (cron triggers only).
Usage Examples
Using State
async def handler(data, ctx: FlowContext):
# Get state
counter = await ctx.state.get("global", "request-count") or 0
# Update state
await ctx.state.set("global", "request-count", counter + 1)
# List all items in scope
items = await ctx.state.list("global")
Using Streams
async def handler(data, ctx: FlowContext):
chat_stream = ctx.streams["chat"]
# Add message to stream
await chat_stream.set(
group_id="room-123",
item_id="msg-456",
data={"text": "Hello", "user": "alice"}
)
# Get all messages in room
messages = await chat_stream.list(group_id="room-123")
Enqueuing Events
async def handler(data, ctx: FlowContext):
# Process data
result = process(data)
# Enqueue to next step
await ctx.enqueue({
"topic": "next-step",
"data": result
})
Multi-Trigger Routing
async def handler(input_data, ctx: FlowContext):
return await ctx.match({
"http": async def handle_http(req):
return {"status": 200, "body": {"data": req.body}},
"queue": async def handle_queue(data):
await ctx.enqueue({"topic": "processed", "data": data}),
"cron": async def handle_cron():
await cleanup_old_data(),
"default": async def handle_default(data):
ctx.logger.info(f"Unhandled trigger: {ctx.trigger.type}")
})
Checking Trigger Type
async def handler(input_data, ctx: FlowContext):
if ctx.is_api():
# Handle HTTP request
return {"status": 200, "body": {"ok": True}}
elif ctx.is_queue():
# Handle queue message
data = ctx.get_data()
await process_queue_data(data)
elif ctx.is_cron():
# Handle scheduled task
await run_periodic_task()
