Stormedo Python SDK

This package sends durable HTTP requests through Stormedo and authenticates or verifies signed deliveries from Stormedo. It supports Python 3.10 and newer.

Install

uv add stormedo

Send requests

Use Stormedo in synchronous applications:

import os
from datetime import timedelta

from stormedo import Stormedo

stormedo = Stormedo(os.environ["STORMEDO_TOKEN"])
request = stormedo.send(
    "https://example.com/webhook",
    body={"message": "hello from Stormedo"},
    delay=timedelta(minutes=5),
    max_attempts=5,
)

Use AsyncStormedo when the application already runs an event loop:

import os

from stormedo import AsyncStormedo

async_stormedo = AsyncStormedo(os.environ["STORMEDO_TOKEN"])
request = await async_stormedo.send(
    "https://example.com/webhook",
    body={"message": "hello from Stormedo"},
)

Create one client, reuse it, and call close() or aclose() during application shutdown.

Use raw_body when bytes must be sent unchanged. Headers supplied in headers are sent to the destination. Stormedo never forwards the project API key.

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

Batch items use BatchRequest values instead of dictionaries:

from datetime import timedelta

from stormedo import BatchRequest

requests = stormedo.send_batch(
    [
        BatchRequest(
            url="https://example.com/orders",
            body={"order_id": "ord_123"},
        ),
        BatchRequest(
            url="https://example.com/inventory",
            body={"warehouse_id": "wh_7"},
            delay=timedelta(hours=1),
        ),
    ]
)

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

The synchronous and asynchronous clients expose the same operations:

• send and send_batch
• list_requests, get_request, list_request_attempts, cancel_request, and replay_request
• create_schedule, list_schedules, get_schedule, update_schedule, pause_schedule, resume_schedule, and delete_schedule

Create schedules with a timing helper instead of constructing the API payload:

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://example.com/inventory/sync",
    body={"full_sync": True},
    max_attempts=4,
)

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

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.

Configure delivery verification

Create one verifier with the project ID and reuse it across requests:

import os

from stormedo import DeliveryVerifier

project_id = os.environ["STORMEDO_PROJECT_ID"]
verifier = DeliveryVerifier(project_id)

For asynchronous applications:

from stormedo import AsyncDeliveryVerifier

async_verifier = AsyncDeliveryVerifier(project_id)

Authenticate without checking the body

Use authenticate_delivery when the platform cannot provide the raw body. It authenticates Stormedo and the request context but does not prove payload integrity.

delivery = verifier.authenticate_delivery(
    signature=signature,
    request_id=request_id,
    expected_url="https://example.com/webhooks/stormedo",
    method="POST",
)

assert delivery.payload_integrity == "unchecked"

Verify the exact body

Use verify_delivery when the receiver can capture the exact raw bytes before a JSON or form parser changes them:

delivery = verifier.verify_delivery(
    signature=signature,
    request_id=request_id,
    expected_url="https://example.com/webhooks/stormedo",
    method="POST",
    raw_body=raw_body,
)

assert delivery.payload_integrity == "verified"

Receiver responsibilities

• Reject missing or duplicated Stormedo-Signature and Stormedo-Request-Id headers before calling the core verifier.
• Supply the complete public destination URL from trusted configuration. Do not trust forwarded request headers unless the application has an explicit proxy trust boundary.
• Authenticate or verify at ingress while the delivery token is current.
• Parse the application body only after verify_delivery succeeds when payload integrity is required.

Errors

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

Verification failures raise DeliveryVerificationError. Use its code to handle the failure.