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, Elasticsearch/OpenSearch, ClickHouse, MongoDB, and 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+
pip install 'onestep[elasticsearch]' # Elasticsearch/OpenSearch bulk sink
pip install 'onestep[clickhouse]'    # ClickHouse table sink
pip install 'onestep[mongodb]'       # MongoDB polling, change streams, and sink

Or scaffold a ready-to-run project from a scenario template:

onestep init my-worker --template redis   # interval | webhook | redis | sql-cdc
cd my-worker
pip install -e .   # scaffold dependencies already pin the extras the template needs
onestep run worker.yaml

onestep init --help lists all templates; each generates a minimal worker.yaml plus handler package and prints the pip install line it needs.

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

Logging

onestep run writes application logs, framework logs, and task lifecycle events to stdout at INFO level by default. Application logger names do not need to use the onestep namespace:

import logging

logger = logging.getLogger("billing.kpi_sync")

Use --log-level DEBUG to include fetched, started, and sink-success details. Use --no-task-events to disable the lifecycle logger installed by the CLI:

onestep run your_package.tasks:app --log-level DEBUG
onestep run your_package.tasks:app --no-task-events

An explicit --log-level overrides a level configured by the loaded target, including YAML app.logging.level. Without the option, a target-configured level is preserved; otherwise the CLI uses INFO. When the CLI installs the stdout handler, that resolved level applies to arbitrary application logger names and the onestep namespace. Existing logging handlers and custom StructuredEventLogger instances are preserved; when a host has configured its own handler, it also retains ownership of root logger levels.

Structured JSON logs

Use --log-format json to emit one JSON object per line instead of text — ready for Loki, ELK, or any JSON log collector without a parsing pipeline:

onestep run your_package.tasks:app --log-format json

The format can also be pinned in YAML via app.logging.format, so deployments (K8s/compose) get structured logs without changing the startup command:

app:
  name: billing-sync
  logging:
    level: INFO
    format: json

An explicit --log-format flag overrides the YAML value; without either, text is used.

Every line the CLI stdout handler emits becomes a JSON object with ts, level, logger, and message keys. Task lifecycle events logged by the built-in StructuredEventLogger additionally promote their structured fields to the top level, so platforms can index them directly:

{"ts": "2026-09-03T01:14:01.808690+00:00", "level": "INFO", "logger": "onestep.events", "message": "task succeeded", "event_kind": "succeeded", "app_name": "billing", "task_name": "sync", "source_name": "queue.in", "attempts": 1, "duration_s": 0.0004, "failure_kind": null, "failure_message": null}

Other log records keep their extra={...} attributes under a nested extra key. Unserializable values fall back to their repr so logging never raises. The formatter is also usable directly for embedded setups:

import logging

from onestep import JsonLogFormatter

handler = logging.StreamHandler()
JsonLogFormatter.attach(handler)
logging.getLogger().addHandler(handler)

The default text behavior is unchanged.

Direct app.run() and app.serve() calls do not modify host logging or install task event logging. Embedded applications retain full control of process logging.

Local task diagnostics

Run exactly one task attempt from JSON, or replay a captured failure, without a worker or control plane:

onestep task run your_package.tasks:app --task sync_billing --input input.json
onestep task replay your_package.tasks:app --task sync_billing --envelope captures/failure.json
onestep check your_package.tasks:app --connect

Diagnostics execute the real handler, task hooks, retry decision, and sink routing. Sink I/O is suppressed by default; --send opens, sends to, and closes the selected sinks. Handler and hook code may still perform external side effects in either mode. --timeout defaults to 60 seconds and is enforced in a spawned process, including for synchronously blocked code.

delivery_action is always a prediction because source ack/retry/fail methods are synthetic. In dry-run, would_dead_letter means dead-lettering would occur if the configured dead-letter sink publishes successfully. Use --send to observe that result. A forced timeout during --send can leave a partial external write and a later retry can duplicate it.

check --connect calls open() and close() only when both methods are callable. State/cursor stores without that lifecycle are reported as not_probeable; the command never calls load(), save(), or delete() as a connectivity probe.

Opt-in failure capture makes production envelopes replayable:

from onestep import FailureCaptureConfig, OneStepApp

app = OneStepApp(
    "billing-sync",
    failure_capture=FailureCaptureConfig(
        directory="captures",
        mode="terminal",
        redact_paths=("/body/customer/token",),
    ),
)

Capture files are versioned, private, atomically written, and reject lossy serialization. terminal records only effective terminal failures; all also records retryable attempts. Common values including datetime, UUID, bytes, Decimal, enum, tuple/namedtuple, set, and frozenset round-trip losslessly. Unsupported custom values produce an explicit capture error and no file rather than a degraded record. See docs/yaml-task-definition.md for YAML policy.

Render the worker topology

onestep render prints the topology of any Python or YAML target as a Mermaid flowchart, ready to paste into GitHub, Notion, or Obsidian:

onestep render worker.yaml

graph LR
  %% app: billing-sync
  n0["extract_entities<br/>concurrency=4 · retry=NoRetry · timeout=300s"]
  n1["sqs-orders<br/>MemoryQueue"]
  n2["mysql.meta_sink<br/>MemoryQueue"]
  n1 --> n0
  n0 -->|"emit"| n2

Each task node lists its concurrency, retry policy, and timeout. Edges are labeled emit (with the transform ref when a binding sets one), when/otherwise for conditional routes, and dashed dead_letter edges. Resources shared by multiple tasks appear once, so chained topologies are drawn as connected graphs.

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:

The three database bulk sinks accept one mapping or a non-empty sequence of mappings and await every backend chunk acknowledgement. onestep remains at-least-once: a retry can repeat committed items or chunks, so use stable document IDs, upsert keys, or a dedup-aware ClickHouse schema when duplicates matter. A partial commit whose final write set is unknown is reported as UNCERTAIN and is not automatically replayed.

The Elasticsearch plugin targets the common Elasticsearch/OpenSearch HTTP bulk surface rather than either vendor's Python client. MongoDB polling and change streams can use in-memory state for development, but production restart guarantees require an explicit durable cursor store; change streams emit raw events and default to full_document: updateLookup.

Which connector should I use?

Prefer interval for prototypes and scheduled jobs, a queue connector (Redis/RabbitMQ/SQS/Cloudflare/Kafka) when work arrives as events or needs competing consumers, and the CDC/incremental sources when the source of truth is a database table. onestep init --template {interval,webhook,redis,sql-cdc} scaffolds the four most common setups as ready-to-run projects.

Minimal YAML per connector

Each snippet shows only the resources: block; pair it with a task that binds your handler:

tasks:
  - name: run
    source: <source-resource>
    emit: [<sink-resource>]        # optional
    handler:
      ref: your_pkg.tasks:run

Core (built-in):

resources:
  tick:
    type: interval
    seconds: 60
    immediate: true
  nightly:
    type: cron
    expression: "0 3 * * *"
  intake:
    type: webhook
    path: /hooks/in
    methods: [POST]
  notify:
    type: http_sink
    url: https://example.invalid/hook

Redis Streams (pip install 'onestep[redis,yaml]'):

resources:
  redis:
    type: redis
    url: redis://localhost:6379
  jobs:
    type: redis_stream
    connector: redis
    stream: jobs
    group: workers
    create_group: true

RabbitMQ (pip install 'onestep[rabbitmq,yaml]'):

resources:
  rmq:
    type: rabbitmq
    url: amqp://guest:guest@localhost/
  jobs:
    type: rabbitmq_queue
    connector: rmq
    queue: incoming_jobs
    prefetch: 50

AWS SQS (pip install 'onestep[sqs,yaml]'):

resources:
  sqs:
    type: sqs
    region_name: us-east-1
  jobs:
    type: sqs_queue
    connector: sqs
    url: https://sqs.us-east-1.amazonaws.com/123456789012/jobs

Cloudflare Queues (pip install 'onestep[cloudflare,yaml]'):

resources:
  cf:
    type: cf_queues
    account_id: your-account-id
    api_token: your-api-token
  jobs:
    type: cf_queue
    connector: cf
    queue_id: your-queue-id

MySQL (pip install 'onestep[mysql,yaml]') — incremental polling, binlog CDC, and a table sink share one connector; CDC needs a cursor store to resume:

resources:
  db:
    type: mysql
    dsn: mysql+pymysql://user:password@localhost:3306/app
  cursor:
    type: mysql_cursor_store
    connector: db
  changed_rows:
    type: mysql_incremental
    connector: db
    table: orders
    key: id
    cursor: [updated_at, id]
    state: cursor
  cdc:
    type: mysql_binlog
    connector: db
    server_id: 18491
    schemas: [app]
    tables: [orders]
    state: cursor
    state_key: orders-cdc
  processed:
    type: mysql_table_sink
    connector: db
    table: processed_orders
    mode: upsert
    keys: [id]

PostgreSQL (pip install 'onestep[postgres,yaml]'):

resources:
  db:
    type: postgres
    dsn: postgresql+psycopg://user:password@localhost:5432/app
  cursor:
    type: postgres_cursor_store
    connector: db
  changed_rows:
    type: postgres_incremental
    connector: db
    table: orders
    key: id
    cursor: [updated_at, id]
    state: cursor
  jobs:
    type: postgres_execution_source
    connector: db
    namespace: my-worker
    task_names: [run_report]
  processed:
    type: postgres_table_sink
    connector: db
    table: processed_orders

Kafka (pip install 'onestep[kafka,yaml]', Python 3.10+):

resources:
  kafka:
    type: kafka
    bootstrap_servers: localhost:9092
  orders:
    type: kafka_topic
    connector: kafka
    topic: orders.events
    group_id: orders-workers

MongoDB (pip install 'onestep[mongodb,yaml]') — state is omitted here so polling/change streams use in-memory cursors for development; wire a durable cursor store (e.g. postgres_cursor_store) via state: for restart guarantees:

resources:
  mongo:
    type: mongodb
    uri: mongodb://localhost:27017
    database: app
  changes:
    type: mongodb_change_stream
    connector: mongo
    collection: events
  archive:
    type: mongodb_collection_sink
    connector: mongo
    collection: archive
    mode: upsert
    keys: [event_id]

Elasticsearch/OpenSearch (pip install 'onestep[elasticsearch,yaml]'):

resources:
  search:
    type: elasticsearch
    hosts: ["https://localhost:9200"]
  indexed:
    type: elasticsearch_bulk_sink
    connector: search
    index: events
    operation: index

ClickHouse (pip install 'onestep[clickhouse,yaml]'):

resources:
  db:
    type: clickhouse
    dsn: http://localhost:8123/default
  events:
    type: clickhouse_table_sink
    connector: db
    table: events

Feishu Bitable (pip install onestep-feishu-bitable):

resources:
  bitable:
    type: feishu_bitable
    app_id: your-app-id
    app_secret: your-app-secret
  rows:
    type: feishu_bitable_incremental
    connector: bitable
    app_token: your-app-token
    table_id: tblxxx
    cursor_field: updated_at

Or install everything at once:

pip install 'onestep[all]'

MySQL/PostgreSQL consolidation (issue #133): onestep-sql is now the canonical distribution for both MySQL and PostgreSQL. New deployments should install onestep-sql[mysql] / onestep-sql[postgres] (or onestep[mysql] / onestep[postgres]). The legacy onestep-mysql / onestep-postgres packages remain available as thin forwarding shims — existing pip install onestep-mysql and from onestep_mysql import ... imports keep working unchanged. All 14 YAML resource type names are unchanged. See the migration guide for details.

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 render worker.yaml           # worker topology as a Mermaid diagram
onestep init billing-sync            # scaffold a minimal YAML project
onestep build worker.yaml --out dist/worker.zip

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

Build a deployable worker package

onestep build packages a YAML worker project into a zip that a worker agent can download and run. It validates the target first, collects the YAML entrypoint, local Python modules referenced by handler, hook, and conditional routing refs, dependency declaration files such as pyproject.toml, requirements.txt, and uv.lock, packaging metadata such as README and license files, and writes an onestep-package.json manifest into the zip.

onestep build worker.yaml --strict --out dist/worker.zip

For files that cannot be inferred from imports, add build hints to pyproject.toml:

[tool.onestep.build]
entrypoint = "worker.yaml"
include = ["templates/**"]
exclude = ["templates/private/**"]

Use --env-file .env to provide local values for the pre-build check. .env files are excluded from packages by default; deploy-time configuration should be provided through the worker agent or control plane. The package manifest records the entrypoint so compatible control-plane uploads can infer it automatically; when uploading to an older control plane, pass the same entrypoint explicitly. Use --json to emit the build report for automation.

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.7.2
  

  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 the onestep-control-plane application over a single WebSocket and accept remote commands — with no connector or task-code changes.

The host execution agent lives in apps/work-agent and is published separately as onestep-worker-agent. It connects outbound to the control plane and starts assigned workflow packages as local onestep subprocesses.

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. Optional service-level metadata can be reported with reporter.service_description or ONESTEP_SERVICE_DESCRIPTION and shown in the control plane:

reporter:
  service_description: Synchronizes billing data into the warehouse

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/framework-evolution-roadmap.md — ordered framework milestones and exit gates
• docs/stable-instance-identity.md — reporter identity resolution
• docs/agent-ws-protocol.md — agent WS protocol
• deploy/ — deployment templates

License

MIT