Skip to main content

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

Steps are the core building blocks of Motia applications. They define handlers that respond to triggers like HTTP requests, queue messages, cron schedules, state changes, or stream events.

​
Function: step

from motia import step

my_step = step(config, handler)
Create a step definition or builder.
​
config
StepConfig | dict[str, Any]
required
Step configuration object or dictionary containing:
  • name (str): Unique step name
  • triggers (list): List of trigger configurations
  • enqueues (list): Optional list of topics to enqueue
  • description (str): Optional description
  • flows (list[str]): Optional flow associations
  • infrastructure (InfrastructureConfig): Optional resource config
​
handler
Callable[..., Awaitable[Any]] | None
Optional async handler function. If provided, returns a StepDefinition. If omitted, returns a StepBuilder.
​
return
StepDefinition | StepBuilder
Returns StepDefinition if handler is provided, otherwise StepBuilder for chaining.

​
Class: StepBuilder

Builder pattern for creating step definitions.

​
Methods

​
handle

def handle(self, handler: StepHandler) -> StepDefinition
Attach a handler to complete the step definition.
​
handler
Callable[..., Awaitable[Any]]
required
Async function that handles the step execution.
​
return
StepDefinition
Complete step definition with config and handler.

​
Class: StepDefinition

A complete step definition containing both configuration and handler.

​
Attributes

​
config
StepConfig
The step configuration.
​
handler
StepHandler
The async handler function.

​
Type: StepConfig

from motia import StepConfig
from motia.triggers import http, queue

config = StepConfig(
    name="my-step",
    triggers=[http("POST", "/api/process")],
    enqueues=["output-topic"],
    description="Process API requests"
)

​
Fields

​
name
str
required
Unique identifier for the step.
​
triggers
list[TriggerConfig]
required
List of trigger configurations (API, queue, cron, state, or stream).
​
enqueues
list[str | Enqueue]
default:"[]"
Topics to enqueue to. Can be topic names (strings) or Enqueue objects with label and conditional settings.
​
virtual_enqueues
list[str | Enqueue] | None
Virtual enqueue topics for visualization without actual enqueue calls.
​
virtual_subscribes
list[str] | None
Virtual subscriptions for visualization purposes.
​
description
str | None
Human-readable description of the step.
​
flows
list[str] | None
Flow names this step belongs to.
​
include_files
list[str] | None
Additional files to include in deployment.
​
infrastructure
InfrastructureConfig | None
Resource configuration for handler and queue settings.

​
Type: InfrastructureConfig

from motia.types import InfrastructureConfig, HandlerConfig, QueueConfig

infra = InfrastructureConfig(
    handler=HandlerConfig(ram=256, cpu=512, timeout=60),
    queue=QueueConfig(type="fifo", max_retries=5)
)

​
Fields

​
handler
HandlerConfig | None
Handler resource configuration:
  • ram (int): Memory in MB (default: 128)
  • cpu (int): CPU units (optional)
  • timeout (int): Timeout in seconds (default: 30)
​
queue
QueueConfig | None
Queue configuration:
  • type (“fifo” | “standard”): Queue type (default: “standard”)
  • max_retries (int): Maximum retry attempts (default: 3)
  • visibility_timeout (int): Visibility timeout in seconds (default: 30)
  • delay_seconds (int): Delay before processing (default: 0)

​
Usage Examples

​
HTTP Trigger Step

from motia import step, StepConfig
from motia.triggers import http

my_step = step(
    StepConfig(
        name="api-handler",
        triggers=[http("POST", "/api/users")],
        description="Handle user creation"
    ),
    handler=async def handle(req, ctx):
        user_data = req.body
        # Process user...
        return {"status": 201, "body": {"id": "user-123"}}
)

​
Queue Trigger with Builder Pattern

from motia import step, StepConfig
from motia.triggers import queue

async def process_message(data, ctx):
    print(f"Processing: {data}")
    await ctx.enqueue({"topic": "next-step", "data": data})

my_step = step(
    StepConfig(
        name="queue-processor",
        triggers=[queue("input-topic")],
        enqueues=["next-step"]
    )
).handle(process_message)

​
Multi-Trigger Step

from motia import step, StepConfig
from motia.triggers import http, queue, cron

async def multi_handler(input_data, ctx):
    # Use ctx.match() to route based on trigger type
    return await ctx.match({
        "http": lambda req: handle_http(req),
        "queue": lambda data: handle_queue(data),
        "cron": lambda: handle_scheduled(),
    })

my_step = step(
    StepConfig(
        name="multi-trigger",
        triggers=[
            http("GET", "/health"),
            queue("events"),
            cron("0 0 * * *")
        ]
    ),
    handler=multi_handler
)

​
With Infrastructure Config

from motia import step, StepConfig
from motia.types import InfrastructureConfig, HandlerConfig, QueueConfig
from motia.triggers import queue

my_step = step(
    StepConfig(
        name="heavy-processor",
        triggers=[queue("large-files")],
        infrastructure=InfrastructureConfig(
            handler=HandlerConfig(
                ram=512,
                cpu=1024,
                timeout=300
            ),
            queue=QueueConfig(
                type="fifo",
                max_retries=5,
                visibility_timeout=600
            )
        )
    ),
    handler=async def handle(data, ctx):
        # Process large files...
        pass
)

Build docs developers (and LLMs) love

Get started for freeTalk to us