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

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

import os

from stormedo import AsyncStormedo

async_stormedo = AsyncStormedo(os.environ["STORMEDO_TOKEN"])
request = await async_stormedo.send(
    "https://your-app.example/webhooks/stormedo",
    body={"type": "order.created", "order_id": "ord_123"},
)

Close clients during shutdown

Close each client when the application stops:

stormedo.close()
await async_stormedo.aclose()

Forward exact bytes

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

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:

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

stormedo.cancel_request(request_id)
replay = stormedo.replay_request(request_id)

print(replay.id)

Cancellation preserves the original request and attempt history. Replay creates a new request linked to the original.

Manage recurring schedules

Create schedules with cron or fixed-interval timing:

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.

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)

Next step

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