Core Concepts

Architecture Overview

┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│   Source    │ ──► │    Task     │ ──► │    Sink     │
│  (Data In)  │     │ (Process)   │     │  (Data Out) │
└─────────────┘     └─────────────┘     └─────────────┘

OneStepApp

Application entry point, responsible for task registration and lifecycle management:

from onestep import OneStepApp

app = OneStepApp(
    "my-app",                    # Application name
    config={"key": "value"},     # Configuration
    state=InMemoryStateStore(),  # State store
    shutdown_timeout_s=30.0,     # Shutdown timeout
)

Task Registration

@app.task(source=..., emit=...)
async def my_task(ctx, item):
    ...

Event Listening

app.on_event(InMemoryMetrics())
app.on_event(StructuredEventLogger())

Lifecycle Hooks

@app.on_startup
async def bootstrap(app):
    ...

@app.on_shutdown
async def cleanup(app):
    ...

Source

Data input source, responsible for fetching messages:

from onestep import CronSource, IntervalSource, MemoryQueue, WebhookSource
from onestep_sql.mysql import MySQLConnector
from onestep_rabbitmq import RabbitMQConnector

# In-memory queue
source = MemoryQueue("incoming")

# Timer
source = IntervalSource.every(minutes=5)

# Cron
source = CronSource("0 * * * *")

# Webhook
source = WebhookSource(path="/webhook")

# RabbitMQ
source = RabbitMQConnector("amqp://...").queue("jobs")

# MySQL
source = MySQLConnector("mysql://...").table_queue("tasks")

RabbitMQ, MySQL, Redis Streams, AWS SQS, Cloudflare Queues, and Feishu Bitable are provided by plugin packages. Install them and import from the corresponding plugin module.

Custom Source

from onestep import Source, Delivery

class MySource(Source):
    async def fetch(self) -> list[Delivery]:
        # Fetch messages
        ...
    
    async def ack(self, delivery: Delivery):
        # Acknowledge message
        ...

Sink

Data output target, responsible for publishing messages:

from onestep import MemoryQueue
from onestep_sql.mysql import MySQLConnector
from onestep_rabbitmq import RabbitMQConnector

# In-memory queue
sink = MemoryQueue("output")

# RabbitMQ
sink = RabbitMQConnector("amqp://...").queue("results")

# MySQL
sink = MySQLConnector("mysql://...").table_sink("results")

Custom Sink

from onestep import Sink

class MySink(Sink):
    async def publish(self, body, meta=None):
        # Publish message
        ...

Delivery

The message delivery object:

The runtime gets a Delivery from Source.fetch(), then passes delivery.payload to the task function. Custom Source implementations need to implement ack(), retry() and fail(); built-in connectors already handle acknowledgement, retry, and failure semantics.

Task Context

Task execution context:

@app.task(source=...)
async def my_task(ctx, item):
    # ctx.app - OneStepApp instance
    # ctx.config - Application config
    # ctx.state - Task state
    # ctx.current - Current execution info
    ...

Config Access

app = OneStepApp("demo", config={"region": "cn"})


@app.task(source=...)
async def task(ctx, item):
    region = ctx.config["region"]

State Management

@app.task(source=...)
async def task(ctx, item):
    count = await ctx.state.get("count", 0)
    await ctx.state.set("count", count + 1)

Message Flow

Basic Flow

@app.task(source=source, emit=sink)
async def process(ctx, item):
    return {"result": item}  # Return value sent to sink

Multi-Stage Flow

queue1 = MemoryQueue("stage1")
queue2 = MemoryQueue("stage2")


@app.task(source=MemoryQueue("input"), emit=queue1)
async def stage1(ctx, item):
    return item * 2


@app.task(source=queue1, emit=queue2)
async def stage2(ctx, item):
    return item + 1


@app.task(source=queue2)
async def final(ctx, item):
    print(f"Result: {item}")

Managed Execution

onestep 1.9 introduced Managed Execution mode, which persists task state, results, and leases to a database (PostgreSQL and MySQL), suitable for long-running tasks (such as AI Agent invocations). See PostgreSQL Tracked Execution and MySQL Tracked Execution for deployment guides.

Architecture

FastAPI / Gateway                  Worker
┌──────────────┐                  ┌──────────────────┐
│ExecutionClient│  ──submit──►    │ExecutionBackend   │◄── PostgresExecutionSource
│  .submit()    │                  │  (PostgreSQL)     │     .claim()
│  .get()       │                  │                   │     heartbeat/complete
│  .cancel()    │                  └───────────────────┘
└──────────────┘

Submitting an Execution

from onestep import ExecutionClient
from onestep_sql.postgres import PostgresExecutionBackend

backend = PostgresExecutionBackend(
    dsn="postgresql+psycopg://app:secret@db/app",
    auto_create=True,
)
client = ExecutionClient(backend, namespace="agent-api")

async with client:
    execution = await client.submit(
        "run_agent",
        {"prompt": "..."},
        idempotency_key=request_id,
    )
    # Poll for result
    result = await execution.result()

Worker Consumption

from onestep_sql.postgres import PostgresExecutionSource

source = PostgresExecutionSource(
    dsn="postgresql+psycopg://app:secret@db/app",
    namespace="agent-api",
    task_names=("run_agent",),
    worker_id="agent-worker-1",
)

Each execution source can only be configured with one task name, which must match the app task name bound to that source.

State Machine

Task states include queued → running → succeeded / failed / cancelled / expired, with intermediate states retrying and cancel_requested. Execution is an immutable snapshot; call get() or list() again to get the latest state.

Leases & Reliability

Executions use leases to guarantee at-least-once delivery: workers renew via heartbeat(), and expired leases are reclaimed by claim(). Cancellation is cooperative; external side effects from handlers still require business-level idempotency.

See PostgreSQL Tracked Execution and Core Reliability for details.

Error Handling

Retry

from onestep import MaxAttempts

@app.task(
    source=...,
    retry=MaxAttempts(max_attempts=3, delay_s=1.0)
)
async def might_fail(ctx, item):
    ...

Dead Letter Queue

@app.task(
    source=main_queue,
    dead_letter=dead_letter_queue
)
async def risky_task(ctx, item):
    ...

Timeout

@app.task(source=..., timeout_s=30.0)
async def long_task(ctx, item):
    ...

Running Modes

Direct Run

if __name__ == "__main__":
    app.run()

CLI Run

onestep run module:app

Async Run

import asyncio

async def main():
    await app.serve()

asyncio.run(main())

Next Steps

• Connector - Connector details
• Retry - Retry strategies
• Middleware - Event hooks