Python SDK
Install and authenticate
pip install surge-sdkSet your API key as an environment variable:
export SURGE_API_KEY=sk_live_your_key_hereThe 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:
| Exception | When |
|---|---|
APIStatusError | API returned a 4xx or 5xx response |
APIConnectionError | Network error or timeout |
AuthenticationError | Invalid or missing API key (401) |
RateLimitError | Rate 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 "", 200The Webhook class follows the Standard Webhooks spec and handles replay protection automatically.