SDKs

Python SDK

The Python SDK has a synchronous Halyard client and an asyncio AsyncHalyard client with the same methods. Workers run as a plain process, under Django or alongside FastAPI.

halyardv3.4.0 Source Changelog

On this page

Install
Configure the client
Create a job
Run a worker
Publish an event
Verify webhooks
Handle errors
Typed payloads
Framework integrations
Django
FastAPI
Reference

Install

Add the package to your project:

pip install halyard

Other package managers

pip install halyard# with Pydantic support for typed payloadspip install "halyard[pydantic]"# or with uvuv add halyard

Requirements:

Python 3.10 or newer
Pydantic 2 (optional) for typed payloads

Keep your API key and webhook secret out of source control. The examples on this page read them from the environment:

.env

HALYARD_KEY=hy_test_51Hq8Zx2rTmV9cKpL3dHALYARD_WEBHOOK_SECRET=whsec_7mYb2Qe9LkT4

Configure the client

Create one client and share it across your app. It holds a connection pool, so making a new client per request wastes connections.

app/halyard_client.py

import os from halyard import Halyard halyard = Halyard(    os.environ["HALYARD_KEY"],    region="eu",       # keep data in Frankfurt    timeout=10.0,      # per request, in seconds    max_retries=3,     # network errors, 429 and 5xx)

Client options

regionLiteral["us", "eu", "ap"]default: "us"

Region your account lives in. Requests go to that region's API host.

timeoutfloatdefault: 30.0

Seconds before a single request is aborted. Retries get a fresh timeout.

max_retriesintdefault: 2

Automatic retries for connection errors, 429 and 5xx responses, with exponential backoff.

base_urlstr

Override the API host, for example to point at a local mock server in tests.

http_clienthttpx.Client

Bring your own httpx client for proxies, custom TLS or instrumentation.

Create a job

A job needs a queue, a task name and a JSON payload. It runs as soon as a worker is free, or at run_at if you set one.

Python

from datetime import datetime, timedelta, timezone from app.halyard_client import halyard job = halyard.jobs.create(    queue="emails",    task="send-welcome-email",    payload={"user_id": "usr_8KfQ2"},    run_at=datetime.now(timezone.utc) + timedelta(minutes=10),  # optional    max_attempts=5,) print(job.id, job.status)  # job_3Rk9Tq7Xw2 scheduled

Arguments are keyword-only and match the API fields. run_at must be timezone-aware; a naive datetime raises ValueError instead of guessing. With AsyncHalyard, the same call is await halyard.jobs.create(...).

Run a worker

Workers long-poll Halyard for jobs on the queues you name and call the handler registered for each task. Run as many worker processes as you like; Halyard hands each job to exactly one of them at a time.

app/worker.py

import os from halyard import Job, Worker from app.models import Userfrom app.mail import mailer worker = Worker(os.environ["HALYARD_KEY"], queues=["emails"], concurrency=10)  @worker.task("send-welcome-email")def send_welcome_email(job: Job) -> None:    user = User.get(job.payload["user_id"])    mailer.send(to=user.email, template="welcome")    job.log(f"Sent welcome email to {user.email}")  if __name__ == "__main__":    worker.run()  # handles SIGTERM: finishes running jobs, then exits

Return and the attempt succeeds. Raise and Halyard records the traceback and schedules a retry with backoff. Raise halyard.NonRetryableError to skip the remaining attempts. Handlers can be async def too; the worker runs them on its own event loop.

Publish an event

Events describe something that happened. Halyard fans each one out to every subscription that matches its type, whether that subscription starts a job or calls a webhook. Publishing is fire-and-forget: you get an evt_ ID back in a few milliseconds.

Python

event = halyard.events.publish(    type="order.paid",    data={"order_id": "ord_77Hc1", "amount": 4200, "currency": "eur"},) print(event.id)  # evt_9Vb2Lm4Qa1

Verify webhooks

Halyard signs every webhook delivery with your endpoint's secret. Verify the signature before you trust the body; the SDK does the HMAC check, the timestamp check and the JSON parsing in one call.

app/webhooks.py

import os from halyard import SignatureVerificationError from app.halyard_client import halyard  def handle_webhook(raw_body: bytes, signature: str) -> int:    try:        event = halyard.webhooks.verify(            raw_body,            signature,            os.environ["HALYARD_WEBHOOK_SECRET"],        )    except SignatureVerificationError:        return 400     if event.type == "job.failed":        print("Job failed", event.data["job"]["id"])    return 204

verify() checks the HMAC SHA-256 signature and rejects timestamps older than five minutes. Pass tolerance=600 to change the window. Always hand it the raw request bytes, never a re-serialized dict.

Handle errors

Failed API calls raise a typed error with the API's type, code and the request_id to quote to support. Retryable failures have already been retried by the time you see them.

Python

import halyard as hy try:    halyard.jobs.create(queue="emails", task="send-welcome-email", payload=payload)except hy.RateLimitError as err:    print(f"Slow down, retry in {err.retry_after}s")    raiseexcept hy.InvalidRequestError as err:    print(f"Bad parameter {err.param}: {err.message}")    raiseexcept hy.HalyardError as err:    print(err.type, err.code, err.request_id)  # req_4Tn8Wc2Ke7    raise

Python SDK error classes
Error	Raised when
HalyardError	Base class. Every error below inherits from it and carries `type`, `code`, `status` and `request_id`.
InvalidRequestError	400 or 422. A parameter is missing or invalid; `param` names it.
AuthenticationError	401. The key is missing, revoked or for the wrong environment.
PermissionDeniedError	403. The key's scopes don't allow this call.
NotFoundError	404. The ID doesn't exist or belongs to another project.
ConflictError	409. Includes `idempotency_mismatch` when a key is reused with a different body.
RateLimitError	429. `retry_after` holds the seconds to wait.
APIConnectionError	The request never got a response: DNS, TLS or timeout.

The full list of types and codes is on the Errors page.

Typed payloads

Payloads are JSON on the wire, but they don't have to be untyped in your code. Describe each task's payload once and the SDK checks both ends: the code that enqueues and the handler that runs.

app/tasks.py

from typing import Literal from pydantic import BaseModel from halyard import Job  class WelcomeEmail(BaseModel):    user_id: str    locale: Literal["en", "de"] = "en"  @worker.task("send-welcome-email")def send_welcome_email(job: Job[WelcomeEmail]) -> None:    job.payload.user_id  # str, validated before your code runs  halyard.jobs.create(    queue="emails",    task="send-welcome-email",    payload=WelcomeEmail(user_id="usr_8KfQ2"),)

Annotate the handler with Job[YourModel] and the worker validates each payload before calling you. A payload that fails validation is dead-lettered at once with the Pydantic errors in the job log, instead of being retried. TypedDict and dataclasses work too.

Framework integrations

The SDK works anywhere your language runs. These are the patterns we recommend for the frameworks we see most.

Django

Add halyard.contrib.django to INSTALLED_APPS. It reads HALYARD_KEY from settings, enqueues after the surrounding transaction commits, and adds a manage.py command for workers.

settings.py

INSTALLED_APPS = [    # ...    "halyard.contrib.django",] HALYARD = {    "KEY": env("HALYARD_KEY"),    "REGION": "eu",    "ENQUEUE_ON_COMMIT": True,}

accounts/views.py

from halyard.contrib.django import halyard  def sign_up(request):    user = User.objects.create_user(**form.cleaned_data)    # Sent only if the transaction commits.    halyard.jobs.create(        queue="emails",        task="send-welcome-email",        payload={"user_id": str(user.pk)},    )

Terminal

python manage.py halyard_worker --queues emails --concurrency 10

FastAPI

Use AsyncHalyard so enqueueing never blocks the event loop, and read the raw body with await request.body() to verify webhooks.

main.py

import os from fastapi import FastAPI, HTTPException, Requestfrom halyard import AsyncHalyard, SignatureVerificationError app = FastAPI()halyard = AsyncHalyard(os.environ["HALYARD_KEY"])  @app.post("/webhooks/halyard", status_code=204)async def halyard_webhook(request: Request) -> None:    try:        event = halyard.webhooks.verify(            await request.body(),            request.headers.get("Halyard-Signature", ""),            os.environ["HALYARD_WEBHOOK_SECRET"],        )    except SignatureVerificationError:        raise HTTPException(status_code=400, detail="Invalid signature")    await handle(event)

Reference

Every method on the client maps to an endpoint in the API reference. Source, issues and release notes for the Python SDK live at code.halyard.example/sdks/halyard-python.

Jobs APIEvery job field, status and endpoint, with request and response examples.Events APIPublish events and manage the subscriptions that route them.WebhooksDelivery, retries, signature scheme and every event type.ErrorsError types and codes, and what to do about each one.

Other SDKs