augstash-sdk 1.0.0

Official Python SDK for the AugStash WhatsApp BSP Platform

augstash-sdk — Python

Official Python SDK for the AugStash WhatsApp BSP Platform by AugmenTrait Pvt. Ltd.

---

Installation

pip install augstash-sdk

Requirements: Python ≥ 3.10. No third-party dependencies — uses only the standard library.

---

Quick Start

from augstash_client import AugStashClient, AugStashError

client = AugStashClient(
    base_url="https://api.augstash.com",  # your AugStash deployment URL
    api_key="as_live_xxxxxxxxxxxx"        # API key from Settings → Developer
)

# List connected WABA accounts
connections = client.list_waba_connections()
print(connections)

# Send a WhatsApp message
import json
result = client.send_message(
    waba_id="1234567890",
    phone_number_id="9876543210",
    contact_phone="[phone]",
    meta_payload_json=json.dumps({
        "messaging_product": "whatsapp",
        "to": "[phone]",
        "type": "text",
        "text": {"body": "Hello from AugStash!"}
    })
)
print(result["metaMessageId"])

---

Authentication

Every API call requires a Bearer API key issued per tenant.

Generate your key: AugStash Dashboard → Settings → Developer → API Keys

client = AugStashClient(base_url, api_key)

Keep your API key secret. Never commit it to source control.
Use environment variables:

import os
client = AugStashClient(
    base_url=os.environ["AUGSTASH_BASE_URL"],
    api_key=os.environ["AUGSTASH_API_KEY"]
)

---

API Reference

WABA Management

# List all connected WABA accounts
connections = client.list_waba_connections()

# List phone numbers (optionally filter by WABA UUID)
phones      = client.list_phone_numbers()
waba_phones = client.list_phone_numbers(waba_id="waba-uuid")

# Get WABA health — quality rating, tier, daily limit per phone
health = client.get_waba_health("meta-waba-id")
# → { "wabaId": ..., "connectionStatus": ..., "phones": [...] }

# Submit display name with SLM pre-check before Meta submission
result = client.submit_display_name(
    waba_id="waba-id",
    proposed_name="Acme Pharma",
    legal_name="Acme Pharmaceuticals Ltd.",
    phone_number_id="phone-number-id"
)
# → { "displayNameStatus": "PENDING" | "BLOCKED_BY_SLM", "slmScore": 85, ... }

# Disconnect a WABA cleanly (pauses campaigns, revokes Meta webhook)
client.disconnect_waba("waba-id")

# Complete credential rotation after token expiry
client.complete_credential_rotation("waba-id", "new-meta-auth-code")

Contacts

# List with pagination and search
page = client.list_contacts(page=0, size=50, search="Ravi")
# → { "content": [...], "totalElements": 320, "totalPages": 7 }

# Create contact
contact = client.create_contact({
    "phone": "[phone]",
    "name":  "Ravi Kumar",
    "email": "[e-mail]",
    "tags":  ["vip", "mumbai"]
})

# Update contact
client.update_contact("contact-uuid", {"name": "Ravi K."})

# Bulk import (up to 1,000 contacts per request)
summary = client.bulk_import_contacts([
    {"phone": "[phone]", "name": "Ravi Kumar"},
    {"phone": "[phone]", "name": "Priya Sharma"}
])
# → { "imported": 2, "failed": 0 }

# Record manual opt-out
client.opt_out_contact("contact-uuid")

Campaigns

# Create a campaign in DRAFT status
campaign = client.create_campaign({
    "name":                   "Diwali Offer 2025",
    "templateId":             "template-uuid",
    "segmentId":              "segment-uuid",
    "attributionGoal":        "augvanta.finance.invoice.paid",
    "attributionWindowHours": 48
})

# Launch immediately (must be DRAFT or SCHEDULED)
client.launch_campaign(campaign["id"])

# Pause / Resume
client.pause_campaign(campaign["id"])
client.resume_campaign(campaign["id"])

# Full funnel analytics
analytics = client.get_campaign_analytics(campaign["id"])
# → { "sentCount": 1500, "deliveredCount": 1450, "readCount": 980,
#     "deliveryRate": 0.966, "readRate": 0.676, "optOutRate": 0.004 }

# Per-contact drill-down (paginated)
recipients = client.get_campaign_recipients(campaign["id"], page=0, size=100)

# Export recipients as CSV
csv_bytes = client.export_campaign_csv(campaign["id"])
with open("campaign_report.csv", "wb") as f:
    f.write(csv_bytes)

Templates

import json

# Create a template
client.create_template({
    "name":       "invoice_payment_reminder",
    "category":   "UTILITY",
    "language":   "en",
    "wabaId":     "meta-waba-id",
    "components": json.dumps({
        "body": {
            "type": "BODY",
            "text": "Hi {{1}}, your invoice of ₹{{2}} is due on {{3}}."
        }
    })
})

# Browse available template packs
packs = client.list_template_packs()
# → [{"id": "bfsi", "name": "BFSI Template Pack", "templateCount": "6"}, ...]

# Import a pack — creates PENDING drafts ready for Meta submission
result = client.import_template_pack("bfsi", "meta-waba-id")
# → { "imported": 6, "skipped": 0, "total": 6 }

# Import Healthcare pack
client.import_template_pack("healthcare", "meta-waba-id")

Inbox

# List agent conversations
inbox = client.list_conversations(page=0, size=20)

# Assign conversation to agent or team
client.assign_conversation("conv-id", agent_id="agent-user-id")
client.assign_conversation("conv-id", team_id="team-uuid")

# Reply (auto-translates to contact's detected language)
client.reply_to_conversation("conv-id", {
    "wabaId":          "waba-id",
    "phoneNumberId":   "phone-id",
    "contactPhone":    "[phone]",
    "metaPayloadJson": json.dumps({
        "messaging_product": "whatsapp",
        "to": "[phone]",
        "type": "text",
        "text": {"body": "Your order is confirmed!"}
    })
})

# Resolve conversation (triggers CSAT + SLM summary generation)
client.resolve_conversation("conv-id")

# SLM-generated reply suggestions
suggestions = client.get_reply_suggestions("conv-id", "when will my order arrive?")
# → { "suggestions": ["Your order will arrive by...", "Let me check...", "..."] }

# Conversation SLM summary
summary = client.get_conversation_summary("conv-id")
# → { "summary": "Customer asked about order status. Agent confirmed delivery by 5th." }

# Canned responses
responses = client.list_canned_responses()
matches   = client.search_canned_responses(shortcode="/refund")
matches   = client.search_canned_responses(keyword="payment")

# CSAT report
csat = client.get_csat_report()
# → { "avgScore": 4.3, "responseCount": 120,
#     "distribution": {1: 2, 2: 5, 3: 10, 4: 45, 5: 58} }

FAQ Bot (A9.5)

# Option 1: upload plain text directly
faq_text = open("product-faq.txt").read()
client.upload_faq("Product FAQ v2.1", faq_text)
# → { "status": "UPLOADED", "chunkCount": 23 }

# Option 2: extract text from PDF first (using pdfplumber)
# pip install pdfplumber
import pdfplumber
with pdfplumber.open("faq.pdf") as pdf:
    faq_text = "\n\n".join(page.extract_text() for page in pdf.pages)
client.upload_faq("Product FAQ v2.1", faq_text)

# Check what's uploaded
info = client.get_faq_info()
# → { "title": "...", "chunkCount": 23, "uploadedAt": "...", "updatedAt": "..." }

# Test the bot
answer = client.ask_faq("What is the return policy?")
# → { "answer": "Our return policy allows returns within 30 days...", "confidence": 0.87 }

# Confidence interpretation:
# ≥ 0.70 → answered automatically, not escalated to agent
# < 0.70 → escalated to human agent

# Delete FAQ knowledge base
client.delete_faq()

Analytics

# Agent performance report (A12.3)
stats = client.get_agent_performance()
# → [{ "agentId": "...", "displayName": "Ananya", "totalConversations": 145,
#       "avgCsatScore": 4.5, "csatResponses": 89, "available": True }]

# Contact list health (A12.5)
health = client.get_contact_health()
# → { "reachabilityPct": 87.3, "optOutsLast7Days": 12,
#     "weeklyGrowth": [{"weekStart": "2025-01-01", "newContacts": 45}, ...],
#     "activeAnomalies": 0 }

# Click-to-WhatsApp ROAS report (B2.2)
roas = client.get_ctwa_roas()
# → { "totalCtwaConversations": 320, "byAdSource": [...] }

# Flow funnel analytics (A8.4)
funnel = client.get_flow_analytics("flow-definition-uuid")
# → { "totalSessions": 450, "completionRate": 62.5,
#     "nodeStats": [{"nodeId": "welcome", "entries": 450, "dropOffRate": 5.0}] }

# Active anomaly events (A9.6)
anomalies = client.get_active_anomalies()
# → [{ "anomalyType": "OPT_OUT_RATE_SPIKE", "severity": "HIGH",
#      "description": "...", "campaignsPaused": True }]

# Resolve an anomaly
client.resolve_anomaly(anomalies[0]["id"])

Developer / Sandbox (A15.1–2)

# Enable sandbox — no real WhatsApp messages sent
client.toggle_sandbox(True)

status = client.get_sandbox_status()
# → { "sandboxEnabled": True, "tenantId": "...", "plan": "STARTER" }

# Webhook delivery log
log = client.get_webhook_log(limit=100)
# → { "total": 850, "sent": 830, "suppressed": 12, "queued": 8,
#     "deliveryRate": 97.6, "entries": [...] }

# Replay a failed message by its Augstash message ID
client.replay_webhook("augstash-message-id")

# Disable sandbox when ready for production
client.toggle_sandbox(False)

---

Error Handling

from augstash_client import AugStashClient, AugStashError

try:
    client.launch_campaign("invalid-id")
except AugStashError as e:
    print(f"API error {e.status_code}: {e}")
    # status_code: 400 | 401 | 403 | 404 | 422 | 429 | 500

---

Complete Example — Onboarding Automation

import os
import json
from augstash_client import AugStashClient

client = AugStashClient(
    base_url=os.environ["AUGSTASH_BASE_URL"],
    api_key=os.environ["AUGSTASH_API_KEY"]
)

# 1. Check WABA health
health = client.get_waba_health("meta-waba-id")
for phone in health["phones"]:
    if phone["qualityRating"] == "RED":
        print(f"WARNING: {phone['phoneNumber']} is RED quality!")

# 2. Import BFSI templates
result = client.import_template_pack("bfsi", "meta-waba-id")
print(f"Imported {result['imported']} BFSI templates")

# 3. Upload FAQ knowledge base
with open("product-faq.txt") as f:
    client.upload_faq("Product FAQ", f.read())
print("FAQ uploaded")

# 4. Create and launch a campaign
campaign = client.create_campaign({
    "name":       "EMI Reminder — November",
    "templateId": "emi-template-uuid",
    "segmentId":  "active-borrowers-segment-uuid"
})
client.launch_campaign(campaign["id"])
print(f"Campaign launched: {campaign['id']}")

# 5. Monitor analytics
import time
time.sleep(300)  # wait 5 minutes
analytics = client.get_campaign_analytics(campaign["id"])
print(f"Sent: {analytics['sentCount']}, "
      f"Delivered: {analytics['deliveredCount']}, "
      f"Read: {analytics['readCount']}")

---

License

MIT © AugmenTrait Pvt. Ltd.