Reeds Python library

The official Python library for the Reeds APIs: one client for every Reeds service.

Developer docs · API reference · API changelog · Library changelog

from reeds import Reeds
from reeds.shipping import Quote

reeds = Reeds()  # reads REEDS_CLIENT_ID and REEDS_CLIENT_SECRET

quote = reeds.shipping.quotes.create(pickup=pickup, dropoff=dropoff, package={"description": "Sneakers", "quantity": 1, "weight_kg": 1.5})
if isinstance(quote, Quote):
    delivery = reeds.shipping.deliveries.create(quote_id=quote.id, reference="ORDER-10045")
    print(delivery.tracking_code)  # HQ3W5F2

Contents

• Services
• Why use the library
• Install
• Credentials and modes
• Shipping: quickstart, coverage and quotes, deliveries, statuses, labels, wallet and invoices, sandbox
• Async
• Webhooks: FastAPI, Flask, Django
• Lists and pagination
• Errors
• Retries, idempotency and timeouts
• Types
• Configuration
• Versioning · Support · License

Services

Reeds offers its services as separate APIs. The library gives each one a property on the same client, so one install and one set of credentials cover them all:

Webhooks are shared: one endpoint receives every service's events, all verified by reeds.webhooks. A new service arrives in a minor version of this library, so upgrading never breaks the services you already use.

Why use the library

Install

pip install reeds-sdk

Or uv add reeds-sdk, or poetry add reeds-sdk. Import it as reeds. It needs Python 3.11 or later, and installs HTTPX2 and Pydantic 2.

Credentials and modes

Get your client id and secret from Dashboard → Developers. Each mode has its own pair:

import os

from reeds import Reeds

reeds = Reeds(client_id=os.environ["REEDS_CLIENT_ID"], client_secret=os.environ["REEDS_CLIENT_SECRET"])

reeds.mode  # "test" or "live", from the client id
reeds.scopes()  # the services your credentials can use, for example ["shipping"]

client_id and client_secret default to the REEDS_CLIENT_ID and REEDS_CLIENT_SECRET environment variables. Keep the secret on your server: never put it in a browser or a mobile app.

Shipping

The Shipping API books pickups and deliveries: within a state, and between states along the routes Reeds runs.

Quickstart

from reeds import Reeds
from reeds.shipping import QuoteUnavailable

reeds = Reeds()

# 1. Price it.
quote = reeds.shipping.quotes.create(
    pickup={"name": "Tola Fabrics", "phone": "08051234567", "address": "23 Allen Avenue", "area": "Ikeja", "lga": "Ikeja", "state": "Lagos"},
    dropoff={"name": "Chinedu Eze", "phone": "08098765432", "address": "14 Admiralty Way", "area": "Lekki Phase 1", "lga": "Eti-Osa", "state": "Lagos"},
    package={"description": "2 pairs of sneakers", "quantity": 1, "weight_kg": 1.5},
)

if isinstance(quote, QuoteUnavailable):
    # Not an error: Reeds doesn't serve this route. Tell your customer why.
    raise SystemExit(quote.reason.message)
print(quote.fee.total, quote.schedule)  # kobo, and the pickup and delivery promise

# 2. Book it. The wallet is charged and the delivery gets its tracking code.
delivery = reeds.shipping.deliveries.create(quote_id=quote.id, reference="ORDER-10045")

# 3. Follow it with webhooks (see below), or read it.
latest = reeds.shipping.deliveries.get(delivery.id)
print(latest.status, latest.tracking_url)

Parameters are keyword arguments, and field names are the API's own (quote_id, tracking_code), so your code reads like the API reference. Amounts are integers in kobo: 280000 is ₦2,800. Times are ISO 8601 strings in Lagos time (datetime.fromisoformat reads them).

Coverage and quotes

coverage = reeds.shipping.coverage.get()  # the states, zones and inter-state routes Reeds serves
for state in coverage.states or []:
    print(state.state, state.zones)

A quote holds its price for 24 hours. A route Reeds doesn't serve comes back as a QuoteUnavailable (available is False) with a reason (ADDRESS_NOT_SERVICEABLE, INTERSTATE_NOT_AVAILABLE, ...), never as an error. Branch on reason.code, not the message.

Deliveries

Not packed yet? Book with ready=False. The delivery is charged and gets its tracking code, then waits with a MARK_READY action:

delivery = reeds.shipping.deliveries.create(quote_id=quote.id, ready=False)
# ...when the parcel is packed:
reeds.shipping.deliveries.mark_ready(delivery.id)  # now "scheduled", with its pickup and delivery times

A booked delivery that isn't marked ready within 7 days is cancelled and refunded in full.

Delivery statuses

Reeds can add statuses: give every match on status a default case.

Labels

label = reeds.shipping.deliveries.label(delivery.id)
label.write_to(f"{delivery.id}.pdf")  # or a directory, to save it as label.filename
label.data, label.content_type  # the bytes, and "application/pdf"

Wallet and invoices

wallet = reeds.shipping.wallet.get()
print(wallet.balance)  # kobo

# Live mode: start a Paystack top-up (₦50,000) and send the payer to the URL.
# wallet.funded carries the same reference when the money lands.
top_up = reeds.shipping.wallet.fund(amount=5_000_000)
print(top_up.payment_url, top_up.reference)

for entry in reeds.shipping.wallet.list_transactions():
    print(entry.type, entry.amount, entry.description)

# Accounts billed monthly:
for invoice in reeds.shipping.invoices.list(status="issued"):
    if invoice.id:
        reeds.shipping.invoices.pdf(invoice.id).write_to(f"{invoice.number}.pdf")

Test mode and the sandbox

With test credentials, build the whole flow without real money or riders:

reeds.shipping.sandbox.fund_wallet(amount=1_000_000)  # ₦10,000 in the test wallet, at once
reeds.shipping.sandbox.advance_delivery(delivery.id)  # the next status, with its webhook
reeds.shipping.sandbox.advance_delivery(delivery.id, status="delivered")  # straight to a status

Sandbox calls are refused before anything is sent when the client has live credentials.

Async

AsyncReeds has the same methods; await them. A list is awaited for one page, or walked with async for:

import asyncio

from reeds import AsyncReeds
from reeds.shipping import Quote


async def main() -> None:
    async with AsyncReeds() as reeds:
        quote = await reeds.shipping.quotes.create(pickup=pickup, dropoff=dropoff, package={"description": "Sneakers"})
        if isinstance(quote, Quote):
            delivery = await reeds.shipping.deliveries.create(quote_id=quote.id)
            print(delivery.tracking_code)

        page = await reeds.shipping.deliveries.list(limit=50)  # await: one page
        print(len(page.data), page.has_more)
        async for delivery in reeds.shipping.deliveries.list(status="delivered"):  # async for: every delivery
            print(delivery.id)


asyncio.run(main())

Create one client and reuse it: it keeps one access token and one connection pool. async with (or await reeds.close()) closes the connections.

Webhooks

Reeds posts events to your endpoint as things happen. Verify each one before you trust it: pass the raw request body, because a body that was decoded and encoded again never matches the signature. The signing secret (whsec_test_... or whsec_live_...) is next to your endpoint in Dashboard → Developers. It isn't your client secret.

construct_event checks the signature and returns a typed event. Match on its class:

import os

from reeds import WebhookSignatureError, webhooks
from reeds.shipping import DeliveryStatusChangedEvent, WalletLowBalanceEvent


def handle(body: bytes, signature: str | None) -> int:
    try:
        event = webhooks.construct_event(body, signature, os.environ["REEDS_WEBHOOK_SECRET"])
    except WebhookSignatureError as error:
        print(error.message)
        return 400

    match event:
        case DeliveryStatusChangedEvent():
            print(event.data.delivery.id, event.data.previous_status, "->", event.data.delivery.status)
        case WalletLowBalanceEvent():
            print("Top up: the wallet is at", event.data.balance)
        case _:
            pass  # Reeds adds event types over time: ignore the ones you don't use
    return 200

webhooks verifies events without credentials; Reeds().webhooks does the same and can also send test events.

FastAPI

await request.body() is the raw body:

import os

from fastapi import FastAPI, Request, Response
from reeds import WebhookSignatureError, webhooks

app = FastAPI()


@app.post("/webhooks/reeds")
async def reeds_webhook(request: Request) -> Response:
    try:
        event = webhooks.construct_event(await request.body(), request.headers.get("reeds-signature"), os.environ["REEDS_WEBHOOK_SECRET"])
    except WebhookSignatureError as error:
        return Response(error.message, status_code=400)
    print(event.id, event.type)  # handle the event here
    return Response(status_code=200)

Flask

request.get_data() is the raw body:

import os

from flask import Flask, request
from reeds import WebhookSignatureError, webhooks

app = Flask(__name__)


@app.post("/webhooks/reeds")
def reeds_webhook() -> tuple[str, int]:
    try:
        event = webhooks.construct_event(request.get_data(), request.headers.get("Reeds-Signature"), os.environ["REEDS_WEBHOOK_SECRET"])
    except WebhookSignatureError as error:
        return error.message, 400
    print(event.id, event.type)  # handle the event here
    return "", 200

Django

request.body is the raw body. The view is exempt from CSRF: Reeds signs its requests instead. In your project the view goes in views.py and the route in urls.py:

import os

from django.http import HttpRequest, HttpResponse
from django.urls import path
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from reeds import WebhookSignatureError, webhooks


@csrf_exempt  # Reeds signs its requests; there's no CSRF token to send
@require_POST
def reeds_webhook(request: HttpRequest) -> HttpResponse:
    try:
        event = webhooks.construct_event(request.body, request.headers.get("Reeds-Signature"), os.environ["REEDS_WEBHOOK_SECRET"])
    except WebhookSignatureError as error:
        return HttpResponse(error.message, status=400)
    print(event.id, event.type)  # handle the event here
    return HttpResponse(status=200)


urlpatterns = [path("webhooks/reeds", reeds_webhook)]

Each has a model in reeds.shipping (DeliveryBookedEvent, WalletFundedEvent, ...). An event type this version doesn't know yet arrives as an UnknownEvent, so a new type never breaks your endpoint.

Good to know:

• Duplicates: an event can arrive more than once. Skip event.ids you've already processed.
• Order: events can arrive out of order. Compare event.data.delivery.updated_at.
• Rotating a secret: pass both while you rotate, as construct_event(body, signature, [new_secret, old_secret]).
• Testing your receiver: send a signed test event to your endpoint:

reeds.webhooks.send_test(type="delivery.booked")

Lists and pagination

A list returns its first page, and iterating it walks every item, fetching pages as you go:

page = reeds.shipping.deliveries.list(status="delivered", limit=50)
page.data  # up to 50, newest first
page.has_more  # True when another page follows
page.next_page()  # the next page, or None

for delivery in reeds.shipping.deliveries.list(status="delivered"):
    ...  # every delivered delivery; pages are fetched as you go

With AsyncReeds, await the list for its first page and await page.next_page(), or async for over it for every item (async).

Errors

Every failure raises a ReedsError subclass. Branch on the class or on error.code, never on the message, which Reeds can reword.

from reeds import InvalidRequestError, PaymentRequiredError, ReedsError

try:
    reeds.shipping.deliveries.create(quote_id=quote.id)
except PaymentRequiredError:
    ...  # INSUFFICIENT_BALANCE: top up, then retry with the same idempotency_key
except InvalidRequestError as error:
    print(error.fields)  # [{"field": "dropoff.lga", "message": "Enter the LGA."}]
except ReedsError as error:
    print(error.code, error.message, error.request_id, error.doc_url)
    raise

Every error has:

• status, code and message;
• fields: what to fix, on validation errors;
• doc_url: the code's page on the developer hub;
• request_id: quote it to support;
• raw: the API's error object, including fields some codes add, such as delivery_id on QUOTE_ALREADY_USED.

Retries, idempotency and timeouts

* An expired token is refreshed once and the request sent again, automatically.

Waits back off from about half a second, up to 8 seconds. Every POST carries an Idempotency-Key, and a retry sends the same one, so Reeds acts on it once. To make your own retries safe too (after a crash, or a deploy), pass a key tied to your order:

reeds.shipping.deliveries.create(quote_id=quote.id, reference="ORDER-10045", idempotency_key="order-10045-delivery")

Every method also takes timeout (seconds), max_retries and extra_headers, for that request only:

reeds.shipping.deliveries.list(limit=10, timeout=10, max_retries=0)

Types

The models and parameter types follow the API's OpenAPI spec. Import them from reeds.shipping:

from reeds.shipping import Delivery, DeliveryStatus


def is_done(delivery: Delivery) -> bool:
    final: list[DeliveryStatus] = ["delivered", "cancelled"]
    return delivery.status in final

Parameters are TypedDicts, so a dict you build ahead is checked too:

from reeds.shipping import AddressInputParam

pickup: AddressInputParam = {"name": "Tola Fabrics", "phone": "08051234567", "address": "23 Allen Avenue", "lga": "Ikeja", "state": "Lagos"}

The types accept values Reeds adds later: an enum such as DeliveryStatus is its known values or any string, so a new status type-checks and parses. Fields Reeds adds are kept on the model:

print(delivery.to_dict())  # the API's JSON, including fields this version doesn't know
delivery.model_extra  # just the fields this version doesn't know

An object that doesn't fit its model (a field missing, a value of another type) never raises: the library keeps what came, so a call that charged never looks like it failed.

Configuration

import httpx2

from reeds import Reeds

reeds = Reeds(
    client_id="rci_test_...",  # default: REEDS_CLIENT_ID
    client_secret="rcs_test_...",  # default: REEDS_CLIENT_SECRET
    base_url="https://api.getreeds.co",  # the default
    timeout=60,  # seconds per attempt
    max_retries=2,
    default_headers={"X-Your-Header": "value"},
    http_client=httpx2.Client(proxy="http://proxy.internal:8080"),  # proxies, custom TLS
)

with Reeds() as reeds:  # closes its connections at the end
    reeds.shipping.coverage.get()

AsyncReeds takes the same options, with an httpx2.AsyncClient as http_client. A client you pass in stays open: close it yourself.

Versioning

The library follows semantic versioning. Until 1.0.0, a minor version may change something you use: read the changelog before upgrading.

Within an API version (v1), Reeds adds fields, statuses and event types without notice. The library keeps them rather than dropping them, and its types allow them.

Every release is built by GitHub Actions and published to PyPI with trusted publishing, with no long-lived token.

Support

• Docs and guides: developer.getreeds.co.
• Help with a call: email support@getreeds.co with the request_id from the error (or the X-Request-Id header). Every call is also listed in Dashboard → Developers → API requests for 30 days.
• A bug in the library: open an issue.
• A security problem: email support@getreeds.co rather than opening an issue.

License

MIT © Reeds Merchants Technologies Ltd.