---
title: Python
description: Install the Stormedo Python SDK, send single or batch requests, inspect delivery history, and manage schedules from synchronous or asynchronous applications.
---

import InstallTabs from '../../../../components/docs/InstallTabs.astro';

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

## Install

<InstallTabs
	syncKey="python-package-manager"
	commands={[
		{ label: 'pip', command: 'pip install stormedo' },
		{ label: 'uv', command: 'uv add stormedo' },
		{ label: 'poetry', command: 'poetry add stormedo' },
		{ label: 'pdm', command: '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

```python
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:

```python
stormedo.close()
await async_stormedo.aclose()
```

## 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
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:

```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)
```

## Next step

Read [Delivery verification](/docs/concepts/delivery-verification/) before accepting signed deliveries at your Python endpoint.