Surge

Python SDK

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.