Python

The Stormedo Python SDK supports Python 3.10 and newer. It provides synchronous and asynchronous clients with the same operations.

Install

pip

pip install stormedo

uv

uv add stormedo

poetry

poetry add stormedo

pdm

pdm add stormedo

Send from a synchronous application

Python

import os
from datetime import timedelta

from stormedo import Stormedo

stormedo = Stormedo(os.environ["STORMEDO_TOKEN"])
request = stormedo.send(
    "https://your-app.example/webhooks/stormedo",
    body={
        "type": "order.created",
        "order_id": "ord_123",
    },
    delay=timedelta(minutes=5),
    max_attempts=5,
)

print(request.id, request.status)

Load STORMEDO_TOKEN from server-side configuration. Do not commit the key to source control. Create one client, reuse it, and close it during application shutdown.

Send from an asynchronous application

Save this as send_async.py and run python send_async.py with STORMEDO_TOKEN set. The test destination accepts the event without requiring your own receiver.

Python

import asyncio
import os

from stormedo import AsyncStormedo

async def main():
    stormedo = AsyncStormedo(os.environ["STORMEDO_TOKEN"])
    try:
        request = await stormedo.send(
            "https://api.stormedo.com/ping",
            body={"message": "Hello from async Python"},
            max_attempts=4,
            idempotency_key="async-first-ping-001",
        )
        print(request.id, request.status)
    finally:
        await stormedo.aclose()

asyncio.run(main())

Close clients during shutdown

The standalone async example closes its client in finally, including on failure. For a long-running service, create one client at startup and close it from the application’s shutdown hook. Close a synchronous client with stormedo.close() and an asynchronous client with await stormedo.aclose() inside an async function.

The following synchronous examples reuse the stormedo client from the first section.

Forward exact bytes

Python

request = stormedo.send(
    "https://your-app.example/events",
    raw_body=b"event=created",
    content_type="application/x-www-form-urlencoded",
    headers={"Authorization": "Bearer destination-token"},
)

Use raw_body when a signature or downstream parser depends on the exact byte sequence.

Send a batch

Python

from datetime import timedelta

from stormedo import BatchRequest

requests = stormedo.send_batch(
    [
        BatchRequest(
            url="https://your-app.example/webhooks/orders",
            body={"type": "order.created", "order_id": "ord_123"},
        ),
        BatchRequest(
            url="https://your-app.example/webhooks/inventory",
            body={"type": "inventory.sync", "warehouse_id": "wh_7"},
            delay=timedelta(hours=1),
        ),
    ]
)

print([request.id for request in requests])

delay and timeout accept timedelta. deliver_at accepts a timezone-aware datetime.

Inspect request history

list_requests supports status, search, method, source, time range, cursor, and limit filters:

Python

page = stormedo.list_requests(
    status="failed",
    method="POST",
    limit=25,
)

print(page.items)

if page.next_cursor:
    next_page = stormedo.list_requests(
        status="failed",
        method="POST",
        limit=25,
        cursor=page.next_cursor,
    )

    print(next_page.items)

request_id = "req_33uZSQsVaf8aZjDuzkuAq"
detail = stormedo.get_request(request_id)
attempts = stormedo.list_request_attempts(request_id)

print(detail.status, len(attempts))

Cancel and replay

Python

cancellation = stormedo.cancel_request(request_id)
print(cancellation.status)

Cancellation preserves history. cancellation_requested acknowledges the signal; it is not a final request status. An in-flight attempt can still finish.

Separately, replay a request that you have confirmed is succeeded, failed, or cancelled:

Python

completed_request_id = "req_33uZSQsVaf8aZjDuzkuAq"
replay = stormedo.replay_request(
    completed_request_id, idempotency_key="order-123-replay-001"
)
print(replay.id)

Replace the illustrative ID with your completed request’s ID. Do not cancel and immediately replay without checking state; replay of active work returns 409. Replay has a new request ID, so use a stable payload event ID if repeating the business effect must be prevented.

Manage recurring schedules

Create schedules with cron or fixed-interval timing:

Python

from datetime import timedelta

from stormedo import cron_timing, interval_timing

schedule = stormedo.create_schedule(
    name="sync-inventory",
    timing=cron_timing("0 */6 * * *", "Asia/Kolkata"),
    url="https://your-app.example/inventory/sync",
    body={"full_sync": True},
    max_attempts=4,
)

stormedo.update_schedule(
    schedule.id,
    timing=interval_timing(timedelta(hours=4)),
)

schedules = stormedo.list_schedules()
definition = stormedo.get_schedule(schedule.id)

print(len(schedules), definition.status)

stormedo.pause_schedule(schedule.id)
stormedo.resume_schedule(schedule.id)
stormedo.delete_schedule(schedule.id)

cron_timing accepts a five-field cron expression and an IANA timezone. interval_timing accepts a timedelta containing a whole number of minutes from one minute through 30 days. Pass a future timezone-aware datetime as starts_at to choose the first run time.

Handle API errors

API and network failures raise StormedoError. Use code for program logic and request_id when contacting Stormedo support.

Python

import os

from stormedo import Stormedo, StormedoError

stormedo = Stormedo(os.environ["STORMEDO_TOKEN"])
try:
    stormedo.get_request("req_33uZSQsVaf8aZjDuzkuAq")
except StormedoError as error:
    print(error.code, error.status, error.request_id, error.details)
    raise

Send options and defaults

send(url, **options) returns an EnqueuedRequest. JSON sends use /requests; raw sends use /send.

Option	Type and behavior
body	Any JSON value. Omission sends no body; explicit None sends JSON null.
raw_body	str or bytes. Mutually exclusive with body and metadata.
content_type	String up to 255 bytes, for raw sends only.
method	GET, POST, PUT, PATCH, or DELETE; default POST.
headers	Destination mapping of strings; up to 64 entries and 32 KiB combined.
delay	timedelta with a positive whole-second total, up to 30 days.
deliver_at	Timezone-aware datetime within 30 days of server time. Cannot be combined with delay; allowed past times run immediately.
timeout	timedelta with a whole-second total from 1 through 300; default 50 seconds per destination attempt.
max_attempts	Total attempts, 1 through 20. Default 3 for JSON sends and batches, 1 for raw sends and schedules.
queue	Label, 1 to 64 letters, numbers, dots, hyphens, or underscores; default default. No FIFO or concurrency guarantee.
metadata	JSON sends only. Up to 32 string entries; 64-byte keys, 1024-byte values, 16 KiB combined. Not forwarded.
idempotency_key	Stable business-operation key, 1 to 255 bytes; generated per call when omitted.

Fractional-second durations and naive datetimes are rejected. Each delivered body is limited to 256 KiB. See reserved headers.

Results and operations

The SDK unwraps API data into dataclass objects with snake-case attributes. Read request.id, not request["data"]["id"]. Results use RFC 3339 timestamp strings; verifier timestamps are timezone-aware datetime objects. Collection results are tuples.

Operation	Result and options
send	EnqueuedRequest: id, project_id, replayed_from_request_id, status, scheduled_at, created_at.
send_batch(items, idempotency_key=...)	Tuple of enqueued requests in input order. 1 to 100 BatchRequest items, with optional per-item keys.
list_requests(...)	Page with items, next_cursor, history_retention_days, history_cutoff_at, has_hidden_history. from_ and to filters take RFC 3339 strings. Default limit 50, maximum 100.
get_request(id)	Request definition and completed attempts; body omitted and header values redacted.
list_request_attempts(id)	Tuple of completed attempts, empty until the first completes.
cancel_request(id)	id and status, either cancellation_requested or cancelled.
replay_request(id, idempotency_key=...)	New enqueued request.
create_schedule(..., idempotency_key=...)	Schedule summary.
list_schedules()	Tuple of schedule summaries.
get_schedule(id)	Schedule detail with request replacing top-level url and max_attempts.
update_schedule(id, ...)	Updated schedule summary.
pause_schedule(id), resume_schedule(id)	Updated schedule summary.
delete_schedule(id)	None, from an HTTP 204.

Every operation has an awaited equivalent on AsyncStormedo. Full response fields are in API objects. The SDK converts the schedule HTTP field retry to total max_attempts.

Schedule creation accepts name, timing, url, JSON body, headers, and max_attempts. In update_schedule, put destination changes in the request mapping, with the full intended URL/body/attempt budget. Omitted headers preserve values; {} clears them. Omitted or None schedule bodies send no body, unlike explicit None in an ordinary JSON send. Never write [REDACTED] header values back into a schedule.

Transport failures and validation

The default API base URL is https://api.stormedo.com; constructor base_url supports controlled test endpoints. Keep one client per application lifetime and call close() or aclose() on shutdown. Do not use the synchronous client inside an async request handler.

The SDK uses an HTTPX timeout of 30 seconds for API transport operations. This is an inactivity timeout, not a total wall-clock deadline, and the client currently has no public timeout override. The send timeout option controls destination attempts, not this publisher call.

The SDK retries transport errors once for sends, batches, reads, cancel, replay, schedule creation, pause, and resume, preserving the original idempotency key. It does not automatically retry HTTP error responses, schedule updates, or schedule deletion. Application cancellation does not cancel durable work already accepted.

API and exhausted transport failures raise StormedoError. Its status and request_id may be None for transport failures; network_error is the transport error code. Invalid local input can raise TypeError or ValueError. A timeout or server error may leave acceptance unknown; follow submission recovery.

Next step

Read Delivery verification before accepting signed deliveries at your Python endpoint.