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"})

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

Running Integration Tests

Integration tests require a running III Engine instance. Make sure to have it built or installed before running tests.

Install dev dependencies:
```
cd motia && uv sync --all-extras
```
Run tests:
```
uv run pytest
```

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 integration test suite covers:

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

Metadata

Release files for motia 0.2.2a20260220201719

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 0.2.2a20260220201719
File	Size	Uploaded
motia-0.2.2a20260220201719.tar.gz	98.6 kB	Feb 20, 2026	Details

Built distribution (wheel)

Table of built distributions (wheels) for motia 0.2.2a20260220201719
File	Interpreter	ABI	Platform	Reset
motia-0.2.2a20260220201719-py3-none-any.whl 27.9 kB Feb 20, 2026	Python 3	none	any	Details

Total release size: 126.5 kB

Release files / motia-0.2.2a20260220201719.tar.gz

Download URL	motia-0.2.2a20260220201719.tar.gz
Size	98.6 kB
Tags	Source
SHA-256 checksum How to use checksums	`978bea02888d7d365504a63b0f359df141d8181a07d12467950a5929accf1680`
BLAKE2b-256 checksum How to use checksums	`4f79a7707df4f7b82e9ff5ad5aa2c0dc7c38d70cbb3b8217e0b1bdd9681e1bcb`
Upload date	Feb 20, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/6.1.0 CPython/3.13.7`

Release files / motia-0.2.2a20260220201719-py3-none-any.whl

Download URL	motia-0.2.2a20260220201719-py3-none-any.whl
Size	27.9 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`16c6f622f7e425594bd99b7f4306450090839f3dd829d217daf7b030b0320df0`
BLAKE2b-256 checksum How to use checksums	`c2c54d256ffe64717e8fa32d7943628f268e2b422f4c31c6c0c84ab8dd48c84a`
Upload date	Feb 20, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/6.1.0 CPython/3.13.7`