Surge home page
SupportDashboard
Surge

Python SDK

The Python SDK wraps the Surge HTTP API with a typed client, sync and async variants, automatic pagination, and a webhook verifier. It's published as surge-sdk on PyPI and supports Python 3.9 and newer.

This page covers installing the package, sending your first request with both clients, paginating list endpoints, handling typed errors, and verifying webhook signatures in your handler.

Install and authenticate

pip install surge-sdk

Set your API key as an environment variable:

export SURGE_API_KEY=sk_live_your_key_here

The client reads it automatically:

from surge import Surge

surge = Surge()

For async code, use AsyncSurge:

from surge import AsyncSurge

surge = AsyncSurge()

Your first request

Send a message:

message = surge.messages.create(
    account_id="acct_01jrzhe8d9enptypyx360pcmxj",
    to="+18015551234",
    body="Your appointment is confirmed for Friday at 2pm.",
)

print(message.id)      # msg_01j...
print(message.status)  # "sent"

With AsyncSurge:

import asyncio

async def main():
    message = await surge.messages.create(
        account_id="acct_01j...",
        to="+18015551234",
        body="Your appointment is confirmed.",
    )
    print(message.status)

asyncio.run(main())

Pagination

The SDK returns auto-paginating iterators for list endpoints. You don't need to manage cursors manually:

# Iterates through all messages automatically
for message in surge.messages.list(account_id="acct_01j..."):
    print(message.id, message.status)

To page manually, use the raw list method:

page = surge.messages.list_page(account_id="acct_01j...", limit=25)
print(page.data)            # list of messages
print(page.pagination.next_cursor)

Retries and timeouts

The SDK retries transient failures (network errors, 5xx responses) with exponential backoff. Default: 3 retries.

Configure retries and timeout on the client:

surge = Surge(
    max_retries=5,
    timeout=30.0,  # seconds
)

To disable retries for a specific call:

message = surge.messages.create(
    account_id="acct_01j...",
    to="+18015551234",
    body="Hello",
    max_retries=0,
)

Error handling

Errors raise typed exceptions:

from surge import SurgeError, APIStatusError

try:
    surge.messages.create(
        account_id="acct_01j...",
        to="+18015551234",
        body="Hello",
    )
except APIStatusError as e:
    print(e.status_code)  # e.g. 422
    print(e.error.type)   # e.g. "opted_out"
    print(e.error.message)

Common exception types:

ExceptionWhen
APIStatusErrorAPI returned a 4xx or 5xx response
APIConnectionErrorNetwork error or timeout
AuthenticationErrorInvalid or missing API key (401)
RateLimitErrorRate limit exceeded (429)

Webhooks

Verify webhook signatures and parse events:

import os
from surge.webhooks import Webhook

wh = Webhook(os.environ["SURGE_WEBHOOK_SECRET"])

# In your webhook handler (Flask example):
@app.post("/webhooks/surge")
def surge_webhook():
    try:
        event = wh.verify(request.get_data(), request.headers)
    except Exception:
        return "", 400

    match event["type"]:
        case "message.received":
            handle_inbound(event["data"])
        case "contact.opted_out":
            handle_opt_out(event["data"])

    return "", 200

The Webhook class follows the Standard Webhooks spec and handles replay protection automatically.