Croniq Runner SDK for Python

Build job execution runners for Croniq in async Python. The SDK polls a Croniq server for work, dispatches your handlers, streams structured logs back, and reports completion — idiomatic asyncio + httpx + Pydantic v2.

Install

pip install croniq-runner
# Optional: OpenTelemetry tracing
pip install "croniq-runner[otel]"

Python 3.11+ required (asyncio.TaskGroup, tomllib).

Quick start

import asyncio
import os

from croniq_runner import Runner, RunnerOptions

async def hello(ctx):
    ctx.logger.info("hello from %s (attempt %d)", ctx.job_key, ctx.attempt)
    await ctx.log("emitting a structured event", fields={"customer": "acme"})

async def main():
    runner = Runner(RunnerOptions(
        server_url="http://localhost:4000",
        api_key=os.environ["CRONIQ_API_KEY"],
        capabilities=["billing"],
        tags=["lang=python", "env=dev"],
        max_inflight=5,
    ))
    runner.add_handler("hello:world", hello)
    await runner.run()

asyncio.run(main())

The key comes from the environment on purpose — matching the Go and TypeScript samples. A literal api_key="croniq_..." in a snippet like this gets copy-pasted into source control more often than not.

Stop the runner with Ctrl-C or by calling runner.request_drain() from another coroutine — in-flight handlers get up to drain_timeout_ms to finish before the loop returns.

Features

• asyncio-first — every public coroutine returns awaitably; no sync surface.
• Pydantic v2 DTOs mirroring openapi.yaml snake_case wire format.
• Two-tier logging:
  • ctx.logger — standard logging.Logger scoped with execution_id, job_key, runner_id, attempt.
  • ctx.log_writer — streaming channel backed by a bounded asyncio.Queue with batching (32 events / 200 ms / max 100 per POST), drained before the ack.
• Server-side cancellation — PollResponse.cancel is honoured via ctx.cancellation (an asyncio.Event). Handlers should await ctx.cancellation.wait() between checkpoints, or just use await asyncio.sleep(...) — the runner cancels the underlying task when the event fires.
• Lease renewal — periodic POST /v1/work/renew heartbeat for each in-flight execution.
• Self-registration — pass schedule="5m" to add_handler and the runner POSTs to /v1/jobs/register on startup.
• OpenTelemetry — opt-in via the [otel] extra; spans wrap each handler invocation when opentelemetry-api is importable. Zero dependency otherwise.

Capabilities vs Tags

A common pitfall: don't put implementation details into capabilities. Capabilities drive server-side job routing (require/prefer in the Croniqfile). Tags are filter-only — they show up in the UI and operational views but don't influence routing.

If your runner is Python-based, that belongs in tags (lang=python, platform=linux-x64), not capabilities — so a future Rust- or .NET-runner with the same business capabilities can take over without rewriting Croniqfile entries.

Handler API

A handler is any async def fn(ctx: ExecutionContext) -> None. The ctx exposes:

Two ways to control the ack failure message:

from croniq_runner import HandlerError

async def my_handler(ctx):
    if not data_available():
        raise HandlerError("upstream feed unavailable")  # ack.error = "upstream feed unavailable"

Any other exception's str(exc) is forwarded as the error message.

Configuration

Transport security

The API key is attached to every request as an Authorization header. Over http:// it travels in cleartext — and because httpx honours HTTP_PROXY by default, it can traverse an intermediary in the clear too.

RunnerOptions and TriggerClientOptions therefore refuse a cleartext server_url at construction time unless the host is loopback:

• accepted: any https:// URL, and http:// on localhost, 127.0.0.0/8 or ::1 — so the http://localhost:4000 quickstart default keeps working;
• refused: http:// on any other host, with a ValueError naming the URL and the opt-in.

If a deployment genuinely has no TLS terminator (a lab or staging box), opt in explicitly — the SDK then starts, but logs one loud warning on the croniq_runner.security logger:

RunnerOptions(
    server_url="http://croniq.internal:4000",
    api_key=os.environ["CRONIQ_API_KEY"],
    allow_insecure_http=True,  # the API key travels in cleartext
)

Streaming logs example

from croniq_runner import LogLevel

async def long_job(ctx):
    async with ctx.log_writer as writer:  # type: ignore[reportInvalidUsage]
        async for line in slow_generator():
            await writer.write(line, level=LogLevel.INFO)

You don't actually need async with — the runner drains the writer before the ack regardless. Calling aclose() yourself just lets you control when the drain happens (e.g. before a downstream API call that should see the events first).

Triggering jobs on demand (producer)

The Runner above is the consumer side. To fire a job on demand — from a web request, a message handler, anywhere — use TriggerClient (the producer). It wraps POST /v1/trigger and is independent of the runner: a pure producer never polls, and it carries its own credentials, because triggering needs the jobs:trigger (or admin) scope that runner poll keys typically lack.

import asyncio
import os

from croniq_runner import TriggerClient, TriggerClientOptions

async def main():
    async with TriggerClient(TriggerClientOptions(
        server_url="http://localhost:4000",
        # jobs:trigger (or admin) scope — not a runner key
        api_key=os.environ["CRONIQ_TRIGGER_KEY"],
    )) as client:
        result = await client.trigger(
            "billing:invoice",
            metadata={"invoice_id": "inv_42"},
            idempotency_key="evt-2026-07-14-001",  # optional: dedup retries
        )
        print(result.execution_id, result.queued, result.deduplicated)

asyncio.run(main())

trigger() takes the same routing/metadata knobs as a scheduled fire:

…and returns a TriggerResult:

Unset optional arguments are omitted from the request body (never sent as null). A non-2xx response raises httpx.HTTPStatusError — including the per-job queue-overflow 429, so a producer batching triggers observes that backpressure instead of piling work up unbounded.

Conformance suite

The Python binding for the language-agnostic conformance suite lives under tests/conformance/. Every case is one pytest parameter; run them with:

pip install -e ".[dev]"
pytest tests/conformance

The cases live at sdks/conformance/cases/*.yaml and are loaded by file — adding a new YAML automatically adds a new test. The producer (trigger) cases at sdks/conformance/cases-trigger/*.yaml are driven the same way and exercise TriggerClient.

Development

python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev,otel]"
ruff check .
mypy
pytest

Releasing

Releases run via .github/workflows/python-sdk-release.yml and upload to PyPI through Trusted Publishing (OIDC — no API token in the repo).

1. Bump version = "X.Y.Z" in pyproject.toml.
2. Add a ## [X.Y.Z] section to CHANGELOG.md.
3. Commit, push to main.
4. Tag and push:

   git tag python-sdk-vX.Y.Z
   git push origin python-sdk-vX.Y.Z
   

5. The workflow builds, verifies the tag matches pyproject.toml, publishes to PyPI, then attaches the wheel + sdist to a GitHub Release.

One-time PyPI setup (project owner): add a Pending Publisher on pypi.org with project croniq-runner, owner nuetzliches, repository croniq, workflow python-sdk-release.yml, environment pypi.

License

Dual-licensed under MIT or Apache-2.0, matching the rest of the Croniq repository.