onestep

English | 简体中文

onestep is a small async task runtime for queue, polling, schedule, and webhook workloads. You declare a task with a source and optional sink, and the runtime takes care of fetching, concurrency, retries, dead-lettering, and telemetry.

• One decorator turns any async function into a managed task
• Pluggable connectors for memory, MySQL, RabbitMQ, Redis, SQS, Kafka, Feishu
• Scheduling via interval, cron, webhook, or DB-backed queues
• Production-ready: retries, dead-letter, timeouts, state stores, metrics, and an optional control-plane reporter
• Two config styles: plain Python, or declarative YAML
• Python 3.9+

Quick start

Install:

pip install onestep
# optional extras:
pip install 'onestep[yaml]'          # YAML task definitions
pip install 'onestep[control-plane]' # push telemetry to onestep-control-plane
pip install 'onestep[kafka]'         # Kafka topic source/sink, Python 3.10+

Define an app, then run it with the onestep CLI:

from onestep import IntervalSource, OneStepApp

app = OneStepApp("billing-sync")


@app.task(source=IntervalSource.every(hours=1, immediate=True, overlap="skip"))
async def sync_billing(ctx, _):
    print("syncing billing data")

onestep run your_package.tasks:app
onestep check your_package.tasks:app   # validate the target before starting

What it does

Core concepts

The whole runtime is built on four ideas:

• OneStepApp — task registry and lifecycle manager
• Source — fetches data from a queue, schedule, webhook, or polling backend
• Sink — publishes processed results downstream
• Delivery — a single fetched item exposing ack / retry / fail

from onestep import MemoryQueue, OneStepApp

app = OneStepApp("demo")
source = MemoryQueue("incoming")
sink = MemoryQueue("processed")


@app.task(source=source, emit=sink, concurrency=4)
async def double(ctx, item):
    return {"value": item["value"] * 2}


async def main():
    await source.publish({"value": 21})
    await app.serve()

Connectors

Each backend ships as its own package so you only install what you use:

Or install everything at once:

pip install 'onestep[all]'

Configuration styles

Plain Python

Best for application code. Each connector is a class you instantiate:

from onestep import OneStepApp
from onestep_redis import RedisConnector

app = OneStepApp("redis-demo")
redis = RedisConnector("redis://localhost:6379")
source = redis.stream("jobs", group="workers", batch_size=100)
out = redis.stream("processed")


@app.task(source=source, emit=out, concurrency=8)
async def process_job(ctx, item):
    return {"job": item["job"], "status": "done"}

YAML

Best for deployment wiring. Keep business logic in Python; describe the runtime — app, resources, hooks, tasks — declaratively.

app:
  name: billing-sync

resources:
  tick:
    type: interval
    minutes: 5
    immediate: true

tasks:
  - name: sync_billing
    source: tick
    handler:
      ref: your_package.handlers.billing:sync_billing

onestep run worker.yaml
onestep check --strict worker.yaml   # schema validation, unknown-field detection
onestep init billing-sync            # scaffold a minimal YAML project

The full YAML schema, resource types, conditional routing, and state binding are covered in docs/yaml-task-definition.md.

Deployment

• systemd — minimal unit + preflight check template in deploy/
• Official worker image — run YAML workers in Docker without packaging:

  docker run --rm \
    -e ONESTEP_TARGET=/workspace/worker.yaml \
    -v "$PWD:/workspace" \
    ghcr.io/mic1on/onestep-worker:1.5.0
  

  See deploy/worker-runtime-image.md.
• Embed in a web app — recommended shape for FastAPI/Django in deploy/web-service-integration.md.

Control plane

onestep can push runtime telemetry (heartbeat, topology, metrics, events) to onestep-control-plane over a single WebSocket and accept remote commands — with no connector or task-code changes.

Install the reporter plugin first:

pip install 'onestep[control-plane]'

app:
  name: billing-sync

reporter: true

Required env: ONESTEP_CONTROL_PLANE_URL, ONESTEP_CONTROL_PLANE_TOKEN.

Handlers can report low-cardinality custom counters and gauges through the same reporter. The plane stores them and can expose them from its Prometheus /metrics endpoint:

async def sync_users(ctx, payload):
    success_count = 0
    failed_count = 0
    ...
    ctx.metrics.counter("rows_success").inc(success_count)
    ctx.metrics.counter("rows_failed").inc(failed_count)
    ctx.metrics.gauge("batch_size").set(success_count + failed_count)

For identity, multi-replica guidance, env vars, and a local demo, see docs/stable-instance-identity.md.

Examples

Runnable examples live in example/. Highlights:

# 5-second interval task
SYNC_INTERVAL_SECONDS=5 PYTHONPATH=src onestep run example.cli_app:app

# end-to-end: webhook -> queue -> worker -> dead-letter, with metrics + logs
PYTHONPATH=src python3 example/runtime_showcase.py

Upgrading

1.0.0 was a runtime rewrite. If you're coming from 0.5.x, see MIGRATION-0.5-to-1.0.0.md for the old-to-new API mapping, unsupported features, and rollout guidance.

More

• docs/yaml-task-definition.md — YAML schema
• docs/core-reliability.md — stable API, delivery semantics, plugin compatibility, and release checklist
• docs/stable-instance-identity.md — reporter identity resolution
• docs/agent-ws-protocol.md — agent WS protocol
• deploy/ — deployment templates

License

MIT