Migrating from Motia (Python)

Motia was a higher-level framework built on top of iii-sdk. It handled file scanning, middleware wiring, trigger registration, and production packaging automatically. These conveniences came at a cost: they hid iii’s three core primitives — Workers, Functions, and Triggers — behind opaque abstractions that limited what you could build. By moving to iii-sdk directly, you unlock the full power of the engine:

Add new workers that register their own functions and triggers, enabling multi-worker orchestration across services, languages, and runtimes.
Connect from the browser using iii-browser-sdk with Worker RBAC for secure, real-time frontends — no REST layer needed. See Use iii in the browser.
Treat Motia as one worker among many instead of a standalone monolith. Your existing Motia code becomes just another worker in a larger iii deployment.
Understand the primitives directly. Working with register_function, register_trigger, and register_worker builds a mental model that transfers across all iii SDKs and documentation.

Before diving into this migration, we recommend reading the iii documentation to understand Workers, Functions, and Triggers. The quickstart and Everything is a Worker pages are good starting points.

For the Node / TypeScript migration, see Migrating from Motia (Node.js).

Step 1 — Update dependencies

Remove motia and add iii-sdk.

pip uninstall motia
pip install iii-sdk==0.11.0

Before (Motia)	After (iii-sdk)
`motia` (pip)	`iii-sdk==0.11.0` (pip)
`motia build`	`python -m build` or a container image
`motia dev`	`python -m src.main` with a file watcher

The Python package is distributed as iii-sdk on PyPI but imported as iii (e.g., from iii import register_worker).

Step 2 — Initialize the SDK

Create src/lib/iii_client.py. This replaces the implicit connection that Motia managed for you.

import os
from iii import Logger, register_worker, InitOptions

iii = register_worker(
    address=os.environ.get("III_URL", "ws://localhost:49134"),
    options=InitOptions(worker_name="api-worker"),
)

logger = Logger()

Every from motia import logger in your codebase changes to from src.lib.iii_client import logger.

Step 3 — Register functions and triggers directly

Motia auto-registered functions and triggers from exported config objects. With iii-sdk Python, you call iii.register_function and iii.register_trigger directly at module scope — there’s no helper to build. Each handler module registers its function and trigger(s) when imported. This mirrors the Motia config + handler structure without the implicit discovery. Continue to Step 4 to see the handler patterns.

iii-sdk Python supports engine-native HTTP middleware via middleware_function_ids. See Use HTTP middleware for the full pattern.

Step 4 — Migrate handlers

HTTP

Motia
iii-sdk

from motia import http, ApiRequest, ApiResponse

config = {
    "name": "Health",
    "description": "Health check",
    "triggers": [http("GET", "/health")],
    "enqueues": [],
}

def handler(_request: ApiRequest) -> ApiResponse:
    return ApiResponse(status=200, body={"ok": True})

def health(_data):
    return {"statusCode": 200, "body": {"ok": True}}

iii.register_function("health", health)
iii.register_trigger({
    "type": "http",
    "function_id": "health",
    "config": {"api_path": "/health", "http_method": "GET"},
})

HTTP with middleware

Motia
iii-sdk

from motia import http, ApiRequest, ApiResponse

def require_auth(request: ApiRequest) -> bool:
    return request.headers.get("authorization") is not None

config = {
    "name": "ListTags",
    "description": "List tags",
    "triggers": [http("GET", "/tag", middleware=[require_auth])],
    "enqueues": [],
}

def handler(request: ApiRequest) -> ApiResponse:
    org_id = request.headers.get("x-org-id")
    return ApiResponse(status=200, body={"tags": list_tags_for_org(org_id)})

from src.middleware.auth import require_auth  # registers "middleware::require-auth"

def list_tags(data):
    headers = data.get("headers") or {}
    org_id = headers.get("x-org-id")
    return {"statusCode": 200, "body": {"tags": list_tags_for_org(org_id)}}

iii.register_function("list-tags", list_tags)
iii.register_trigger({
    "type": "http",
    "function_id": "list-tags",
    "config": {
        "api_path": "/tag",
        "http_method": "GET",
        "middleware_function_ids": ["middleware::require-auth"],
    },
})

iii-sdk Python uses engine-native middleware registered as regular functions with a middleware:: prefix. See Use HTTP middleware for the full pattern.

Cron

Motia
iii-sdk

from motia import cron, logger

config = {
    "name": "DailyReport",
    "description": "Generate daily report",
    "triggers": [cron("0 0 9 * * * *")],
    "enqueues": [],
}

def handler(input: None) -> None:
    generate_report()

def daily_report(_data) -> None:
    generate_report()

iii.register_function("daily-report", daily_report)
iii.register_trigger({
    "type": "cron",
    "function_id": "daily-report",
    "config": {"expression": "0 0 9 * * * *"},
})

Queue (durable subscriber)

Motia
iii-sdk

from motia import queue, logger
from typing import Any

config = {
    "name": "ProcessOrder",
    "description": "Process created orders",
    "triggers": [queue("order.created")],
    "enqueues": [],
}

def handler(input: Any) -> None:
    process_order(input)

def process_order(data) -> None:
    process_order_impl(data)

iii.register_function("process-order", process_order)
iii.register_trigger({
    "type": "durable:subscriber",
    "function_id": "process-order",
    "config": {"topic": "order.created"},
})

To publish to a topic, use iii.trigger instead of Motia’s enqueue:

iii.trigger({
    "function_id": "iii::durable::publish",
    "payload": {"topic": "order.created", "data": order_payload},
})

State

Motia
iii-sdk

from motia import StateTriggerInput, state, logger

config = {
    "name": "OnStateChange",
    "description": "React to state changes",
    "triggers": [state()],
    "enqueues": [],
}

def handler(input: StateTriggerInput) -> None:
    handle_state_change(input)

def on_state_change(data) -> None:
    handle_state_change(data)

iii.register_function("on-state-change", on_state_change)
iii.register_trigger({
    "type": "state",
    "function_id": "on-state-change",
    "config": {},
})

Stream

Motia
iii-sdk

from motia import StreamTriggerInput, stream, logger

config = {
    "name": "ChatMessage",
    "description": "Handle chat messages",
    "triggers": [stream("chat", group_id="room-1")],
    "enqueues": [],
}

def handler(input: StreamTriggerInput) -> None:
    handle_chat_message(input)

def chat_message(data) -> None:
    handle_chat_message(data)

iii.register_function("chat-message", chat_message)
iii.register_trigger({
    "type": "stream",
    "function_id": "chat-message",
    "config": {"stream_name": "chat", "group_id": "room-1"},
})

Multiple triggers on one function

A function can be driven by multiple triggers. This was possible in Motia via the triggers array, and maps directly to multiple register_trigger calls for the same function_id:

def order_handler(data) -> None:
    handle_order(data)

iii.register_function("order-handler", order_handler)
iii.register_trigger({
    "type": "http",
    "function_id": "order-handler",
    "config": {"api_path": "/orders", "http_method": "POST"},
})
iii.register_trigger({
    "type": "durable:subscriber",
    "function_id": "order-handler",
    "config": {"topic": "order.retry"},
})

Stream and State operations

In Motia, you used the Stream class and stateManager singleton to read and write data. Under the hood, these called iii.trigger() with built-in function IDs (stream::get, state::set, etc.). With iii-sdk, you call iii.trigger() directly. See the Stream and State worker docs for the full list of operations.

Stream
State

from src.lib.iii_client import iii

todo = iii.trigger({
    "function_id": "stream::get",
    "payload": {"stream_name": "todos", "group_id": "user-1", "item_id": "todo-1"},
})

iii.trigger({
    "function_id": "stream::set",
    "payload": {"stream_name": "todos", "group_id": "user-1", "item_id": "todo-1", "data": {"title": "Buy milk"}},
})

iii.trigger({
    "function_id": "stream::delete",
    "payload": {"stream_name": "todos", "group_id": "user-1", "item_id": "todo-1"},
})

items = iii.trigger({
    "function_id": "stream::list",
    "payload": {"stream_name": "todos", "group_id": "user-1"},
})

from src.lib.iii_client import iii

theme = iii.trigger({
    "function_id": "state::get",
    "payload": {"scope": "settings", "key": "theme"},
})

iii.trigger({
    "function_id": "state::set",
    "payload": {"scope": "settings", "key": "theme", "value": "dark"},
})

iii.trigger({
    "function_id": "state::delete",
    "payload": {"scope": "settings", "key": "theme"},
})

all_settings = iii.trigger({
    "function_id": "state::list",
    "payload": {"scope": "settings"},
})

Additional function IDs: stream::update, stream::list_groups, stream::send for streams; state::update, state::list_groups for state.

Stream lifecycle hooks (`onJoin` / `onLeave`)

In Motia, stream files could define onJoin and onLeave callbacks inside the stream config. The framework registered a single shared function for each hook and dispatched by stream name internally. With iii-sdk, you register these as regular functions with stream:join and stream:leave triggers. The handler receives a dict with stream_name, group_id, id, and context.

Motia
iii-sdk

config = {
    "name": "todo",
    "schema": todo_schema,
    "base_config": {"storage_type": "default"},
}

def on_join(subscription, context, auth_context):
    return {"unauthorized": False}

def on_leave(subscription, context, auth_context):
    pass  # cleanup logic

from src.lib.iii_client import iii

def stream_on_join(event):
    stream_name = event.get("stream_name")
    group_id = event.get("group_id")
    item_id = event.get("id")
    context = event.get("context")
    return {"unauthorized": False}

iii.register_function("stream::on-join", stream_on_join)
iii.register_trigger({
    "type": "stream:join",
    "function_id": "stream::on-join",
    "config": {},
})

def stream_on_leave(event):
    stream_name = event.get("stream_name")
    group_id = event.get("group_id")
    item_id = event.get("id")
    context = event.get("context")
    # cleanup logic

iii.register_function("stream::on-leave", stream_on_leave)
iii.register_trigger({
    "type": "stream:leave",
    "function_id": "stream::on-leave",
    "config": {},
})

If you had multiple streams with different onJoin / onLeave logic, dispatch by event["stream_name"] inside the handler or register separate function IDs per stream.

Stream authentication (`authenticateStream`)

In Motia, motia_config.py (or motia.config.ts for JS projects) exported an authentication function that ran during WebSocket upgrade. With iii-sdk, you register a function directly and reference its ID in the engine’s stream module config.

Motia
iii-sdk

# motia_config.py
def authenticate_stream(req, context):
    return {"context": {"userId": "sergio"}}

from src.lib.iii_client import iii

def stream_authenticate(data):
    headers = data.get("headers")
    path = data.get("path")
    query_params = data.get("query_params")
    addr = data.get("addr")
    return {"context": {"userId": "sergio"}}

iii.register_function("stream::authenticate", stream_authenticate)

The engine calls this function during WebSocket upgrade. Reference the function ID in your config.yaml:

workers:
  - name: iii-stream
    config:
      auth_function: stream::authenticate

Key differences:

Motia used queryParams (camelCase); iii-sdk uses query_params (snake_case).
No trigger registration is needed for auth — it is config-driven via auth_function.
The motia_config.py file is no longer needed; delete it.

Step 5 — Update request and response shapes

Request properties

Motia wrapped iii-sdk’s HTTP request into friendlier property names. With iii-sdk you receive the raw shape.

Property	Motia (`input.*`)	iii-sdk (`input.*`)
Route params	`params`	`path_params`
Query string	`query`	`query_params`
JSON body	`body`	`body`
Headers	`headers`	`headers`
HTTP method	`method`	`method`
Request stream	`requestBody`	`request_body`
Response stream	`response`	`response`

Motia Python’s ApiRequest already exposed path_params, query_params, body, headers, method. When switching to iii-sdk, handlers receive those fields in a plain dict instead of a Pydantic model, so attribute access (request.path_params) becomes dict-key access (data["path_params"]).

# Before (Motia — Pydantic model, attribute access)
folder_id = request.path_params.get("folderId")
parent_folder_id = request.query_params.get("parentFolderId")

# After (iii-sdk — raw dict, key access)
folder_id = (data.get("path_params") or {}).get("folderId")
parent_folder_id = (data.get("query_params") or {}).get("parentFolderId")

`query_params` value types

In iii-sdk, query_params values are str | list[str]. Use a helper to safely extract the first value:

def first_query_param(data: dict, name: str) -> str | None:
    value = (data.get("query_params") or {}).get(name)
    if value is None:
        return None
    return value[0] if isinstance(value, list) else value

Response shape

iii-sdk uses statusCode instead of status. This applies to every handler return and middleware response.

# Before (Motia)
return ApiResponse(status=200, body={"tags": all_tags})
return ApiResponse(status=404, body={"error": "Not found"})

# After (iii-sdk)
return {"statusCode": 200, "body": {"tags": all_tags}}
return {"statusCode": 404, "body": {"error": "Not found"}}

Step 6 — Set up the entry point

Motia discovered *_step.py files automatically. With iii-sdk Python, create src/main.py that imports every handler module as a side effect.

# src/main.py
import signal
import time

from src.lib.iii_client import iii  # noqa: F401 — connects on import

# Import every handler module; each registers its function + trigger at import time.
from src.handlers import health  # noqa: F401
from src.handlers import auth  # noqa: F401
from src.handlers.orders import create_order  # noqa: F401
from src.handlers.orders import list_orders  # noqa: F401
from src.handlers.reports import daily_report  # noqa: F401
# ... import every handler module


def _shutdown(*_args) -> None:
    iii.shutdown()
    raise SystemExit(0)


if __name__ == "__main__":
    signal.signal(signal.SIGINT, _shutdown)
    signal.signal(signal.SIGTERM, _shutdown)
    while True:
        time.sleep(1)

Each import executes the register_function and register_trigger calls at the module level, registering the function and its triggers with the engine.

File renaming

Rename all *_step.py files to remove the _step suffix:

src/steps/health_step.py  →  src/handlers/health.py
src/steps/auth_step.py    →  src/handlers/auth.py

The directory structure is yours to decide — handlers/, routes/, rest/, or any layout that fits your project.

Step 7 — Development workflow

The iii-exec worker has a built-in file watcher. In your config.yaml, declare watch globs alongside the exec command — no extra dependencies, no wrapper process:

workers:
  - name: iii-exec
    config:
      watch:
        - src/**/*.py
      exec:
        - python
        - -m
        - src.main

Any change matching a watch glob restarts the exec pipeline.

Step 8 — Production build

Python has no direct esbuild equivalent. The recommended production path is a virtualenv (or container image) with pinned dependencies. Add to pyproject.toml:

[project]
name = "my-worker"
version = "0.1.0"
dependencies = [
    "iii-sdk==0.11.0",
]

Build and run:

pip install .
python -m src.main

For containerized deploys, a minimal Dockerfile:

FROM python:3.11-slim
WORKDIR /app
COPY pyproject.toml ./
RUN pip install --no-cache-dir .
COPY src ./src
CMD ["python", "-m", "src.main"]

Update your production config to run the installed module (omit watch in production):

workers:
  - name: iii-exec
    config:
      exec:
        - python
        - -m
        - src.main

Unlike esbuild, you do not bundle Python sources — the runtime imports modules directly. Ship your source tree (or a wheel) rather than a single file.

Changelog

0.11.0

Migrating from Motia (Python)

Step 1 — Update dependencies

Step 2 — Initialize the SDK

Step 3 — Register functions and triggers directly

Step 4 — Migrate handlers

HTTP

HTTP with middleware

Cron

Queue (durable subscriber)

State

Stream

Multiple triggers on one function

Stream and State operations

Stream lifecycle hooks (`onJoin` / `onLeave`)

Stream authentication (`authenticateStream`)

Step 5 — Update request and response shapes

Request properties

`query_params` value types

Response shape

Step 6 — Set up the entry point

File renaming

Step 7 — Development workflow

Step 8 — Production build

Migration checklist

Changelog

0.11.0

​Step 1 — Update dependencies

​Step 2 — Initialize the SDK

​Step 3 — Register functions and triggers directly

​Step 4 — Migrate handlers

​HTTP

​HTTP with middleware

​Cron

​Queue (durable subscriber)

​State

​Stream

​Multiple triggers on one function

​Stream and State operations

​Stream lifecycle hooks (onJoin / onLeave)

​Stream authentication (authenticateStream)

​Step 5 — Update request and response shapes

​Request properties

​query_params value types

​Response shape

​Step 6 — Set up the entry point

​File renaming

​Step 7 — Development workflow

​Step 8 — Production build

​Migration checklist

Step 1 — Update dependencies

Step 2 — Initialize the SDK

Step 3 — Register functions and triggers directly

Step 4 — Migrate handlers

HTTP

HTTP with middleware

Cron

Queue (durable subscriber)

State

Stream

Multiple triggers on one function

Stream and State operations

Stream lifecycle hooks (`onJoin` / `onLeave`)

Stream authentication (`authenticateStream`)

Step 5 — Update request and response shapes

Request properties

`query_params` value types

Response shape

Step 6 — Set up the entry point

File renaming

Step 7 — Development workflow

Step 8 — Production build

Migration checklist