Motia Framework for Python

High-level framework for building workflows with the III Engine.

Installation

uv pip install motia

Usage

Defining a Step

from motia import FlowContext, queue

config = {
    "name": "process-data",
    "triggers": [queue("data.created")],
    "enqueues": ["data.processed"],
}

async def handler(data: dict, ctx: FlowContext) -> None:
    ctx.logger.info("Processing data", data)
    await ctx.enqueue({"topic": "data.processed", "data": data})

API Steps

from motia import ApiRequest, ApiResponse, FlowContext, http

config = {
    "name": "create-item",
    "triggers": [http("POST", "/items")],
    "emits": ["item.created"],
}

async def handler(req: ApiRequest, ctx: FlowContext) -> ApiResponse:
    ctx.logger.info("Creating item", req.body)
    await ctx.enqueue({"topic": "item.created", "data": req.body})
    return ApiResponse(status=201, body={"id": "123"})

Channel-based HTTP (Streaming)

import json
from typing import Any

from motia import FlowContext, MotiaHttpArgs, http

config = {
    "name": "HttpChannelEcho",
    "triggers": [http("POST", "/http-channel/echo")],
}

async def handler(args: MotiaHttpArgs[Any], ctx: FlowContext[Any]) -> None:
    request = args.request
    response = args.response

    chunks: list[bytes] = []
    async for chunk in request.request_body.stream:
        chunks.append(chunk if isinstance(chunk, bytes) else str(chunk).encode("utf-8"))

    await response.status(200)
    await response.headers({"content-type": "application/json"})
    response.writer.stream.write(json.dumps({"receivedBytes": len(b"".join(chunks))}).encode("utf-8"))
    response.close()

Streams

from motia import Stream

# Define a stream
todo_stream = Stream[dict]("todos")

# Use the stream
item = await todo_stream.get("group-1", "item-1")
await todo_stream.set("group-1", "item-1", {"title": "Buy milk"})
await todo_stream.delete("group-1", "item-1")
items = await todo_stream.get_group("group-1")

Build & Publish

python -m build
uv publish --index cloudsmith dist/*

Features

Event-driven step definitions
API route handlers
Cron job support
Stream-based state management
Type-safe context with logging

Testing

Install dev dependencies:
```
uv sync --extra dev
```
Run the unit test suite:
```
uv run pytest -m "not integration"
```

Run the unit test suite with coverage:

uv run pytest -m "not integration" --cov=src/motia --cov-report=term-missing

Running Integration Tests

Integration tests require a running III Engine instance on the test ports below.

uv run pytest -m integration

Test Configuration

Tests use non-default ports to avoid conflicts:

Engine WebSocket: ws://localhost:49199
HTTP API: http://localhost:3199

Set III_ENGINE_PATH environment variable to point to the III engine binary.

Test Coverage

The coverage command above measures the unit-testable Python package code under src/motia.

The integration test suite covers:

Bridge connection and function registration
API triggers (HTTP endpoints)
PubSub messaging
Logging module
Motia framework integration
Stream operations (when available)
State management (when available)

Metadata

Release files for motia 1.0.4rc1

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for motia 1.0.4rc1
File	Size	Uploaded
motia-1.0.4rc1.tar.gz	127.2 kB	Mar 15, 2026	Details

Built distribution (wheel)

Table of built distributions (wheels) for motia 1.0.4rc1
File	Interpreter	ABI	Platform	Reset
motia-1.0.4rc1-py3-none-any.whl 34.9 kB Mar 15, 2026	Python 3	none	any	Details

Total release size: 162.0 kB

Release files / motia-1.0.4rc1.tar.gz

Download URL	motia-1.0.4rc1.tar.gz
Size	127.2 kB
Tags	Source
SHA-256 checksum How to use checksums	`19a3fbb0c7133fe010b9ca435c0bb2665867670dc6c9fcdd2f41dca7cd691959`
BLAKE2b-256 checksum How to use checksums	`ec7eab87d1f208fdc4830538873ca26dd786a08e50d593d2138b7cc87bf55176`
Upload date	Mar 15, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/6.1.0 CPython/3.13.7`

Release files / motia-1.0.4rc1-py3-none-any.whl

Download URL	motia-1.0.4rc1-py3-none-any.whl
Size	34.9 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`82c68c97413ea94833bc67501fa44c6122f6e0c63f9ce5c121d54ee20a034c00`
BLAKE2b-256 checksum How to use checksums	`c72e64cc16b2b72da1121d668ecf4ab567027c19285f83554676dec137011a11`
Upload date	Mar 15, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/6.1.0 CPython/3.13.7`