Outbound Campaign

This guide shows how to run a batch outbound call campaign: reading contacts from a CSV, scheduling calls, tracking completions via webhooks, and generating a results report.

Use case: Appointment reminders — call 500 customers to confirm their upcoming appointments.

---

Prerequisites

• An active agent (see Build a Voice Agent)
• A provisioned phone number assigned to the agent
• A webhook endpoint to receive SessionEnded events

---

1. Prepare your contacts CSV

id,phone,name,appointment_date,appointment_time
cust_001,+14155550101,Jane Smith,2025-06-20,10:00 AM
cust_002,+14155550102,Bob Johnson,2025-06-20,11:30 AM
cust_003,+14155550103,Alice Brown,2025-06-20,2:00 PM

---

2. Write the campaign script

import csv
import time
import voisnap
from datetime import datetime, timedelta, timezone
 
client = voisnap.VoisnapClient(api_key="vsnp_live_...")
 
AGENT_ID = "agt_01HXK8Z3MNPQRS"
 
# Campaign window: June 17 9:00 AM – 5:00 PM UTC
CAMPAIGN_START = datetime(2025, 6, 17, 13, 0, 0, tzinfo=timezone.utc)  # 9 AM ET
CALL_INTERVAL_SECONDS = 30  # one call every 30 seconds (120/hour)
 
def load_contacts(csv_path: str) -> list[dict]:
    with open(csv_path) as f:
        return list(csv.DictReader(f))
 
def schedule_campaign(contacts: list[dict]) -> list[dict]:
    scheduled_calls = []
 
    for i, contact in enumerate(contacts):
        scheduled_time = CAMPAIGN_START + timedelta(seconds=i * CALL_INTERVAL_SECONDS)
 
        # Build a personalized opening message
        first_message = (
            f"Hi {contact['name'].split()[0]}, this is Aria calling from Acme "
            f"to confirm your appointment on {contact['appointment_date']} "
            f"at {contact['appointment_time']}. Is this still a good time for you?"
        )
 
        try:
            call = client.outbound_calls.create(
                agent_id=AGENT_ID,
                to_number=contact["phone"],
                scheduled_for=scheduled_time.isoformat(),
                override_first_message=first_message,
                max_retries=2,
                retry_delay_minutes=20,
                metadata={
                    "customer_id": contact["id"],
                    "customer_name": contact["name"],
                    "appointment_date": contact["appointment_date"],
                    "appointment_time": contact["appointment_time"],
                    "campaign": "appointment_reminders_june_2025",
                },
            )
            scheduled_calls.append({
                "call_id": call.id,
                "customer_id": contact["id"],
                "phone": contact["phone"],
                "scheduled_for": scheduled_time.isoformat(),
                "status": "scheduled",
            })
            print(f"✓ Scheduled {contact['name']} ({contact['phone']}) for {scheduled_time}")
 
        except Exception as e:
            print(f"✗ Failed to schedule {contact['name']}: {e}")
            scheduled_calls.append({
                "call_id": None,
                "customer_id": contact["id"],
                "phone": contact["phone"],
                "scheduled_for": scheduled_time.isoformat(),
                "status": "failed_to_schedule",
                "error": str(e),
            })
 
    return scheduled_calls
 
if __name__ == "__main__":
    contacts = load_contacts("contacts.csv")
    print(f"Scheduling {len(contacts)} calls...")
    results = schedule_campaign(contacts)
    print(f"\nDone! {sum(1 for r in results if r['status'] == 'scheduled')} calls scheduled.")
    print(f"First call at: {CAMPAIGN_START}")
    last_time = CAMPAIGN_START + timedelta(seconds=(len(contacts) - 1) * CALL_INTERVAL_SECONDS)
    print(f"Last call at:  {last_time}")

---

3. Set up a webhook to track results

Create a webhook for the SessionEnded event:

webhook = client.webhooks.create(
    agent_id=AGENT_ID,
    url="https://your-server.com/webhooks/campaign",
    events=["SessionEnded", "AnalysisCompleted"],
    secret="whsec_your_secret"
)

In your webhook handler (FastAPI example):

from fastapi import FastAPI, Request, HTTPException
import hmac, hashlib, time, json
from database import db  # your database client
 
app = FastAPI()
WEBHOOK_SECRET = "whsec_your_secret"
 
@app.post("/webhooks/campaign")
async def campaign_webhook(request: Request):
    signature = request.headers.get("X-Webhook-Signature", "")
    timestamp = request.headers.get("X-Webhook-Timestamp", "")
    delivery_id = request.headers.get("X-Webhook-Delivery-Id", "")
    body = await request.body()
 
    # Verify signature
    if abs(time.time() - int(timestamp)) > 300:
        raise HTTPException(400, "Timestamp too old")
    signed = f"{timestamp}.{body.decode()}"
    expected = "sha256=" + hmac.new(
        WEBHOOK_SECRET.encode(), signed.encode(), hashlib.sha256
    ).hexdigest()
    if not hmac.compare_digest(expected, signature):
        raise HTTPException(401, "Invalid signature")
 
    # Idempotency check
    if await db.webhook_events.exists(delivery_id=delivery_id):
        return {"status": "already_processed"}
 
    event = json.loads(body)
 
    if event["type"] == "SessionEnded":
        call_data = event["data"]
        await db.campaign_results.upsert(
            conversation_id=call_data["conversationId"],
            customer_id=call_data.get("metadata", {}).get("customer_id"),
            status=call_data["endReason"],
            duration_seconds=call_data["durationSeconds"],
            cost_usd=call_data["cost"]["totalUsd"],
        )
 
    if event["type"] == "AnalysisCompleted":
        analysis = event["data"]
        outcome = analysis["customFields"].get("resolution_status", "unknown")
        await db.campaign_results.update(
            conversation_id=analysis["conversationId"],
            outcome=outcome,
            sentiment=analysis["sentiment"],
            summary=analysis["summary"],
        )
 
    await db.webhook_events.insert(delivery_id=delivery_id)
    return {"status": "ok"}

---

4. Poll for call statuses

For calls that haven't triggered a webhook yet, poll periodically:

import asyncio
 
async def poll_campaign_status(call_ids: list[str]):
    status_counts = {"scheduled": 0, "in_progress": 0, "completed": 0,
                     "no_answer": 0, "failed": 0, "voicemail": 0}
 
    for call_id in call_ids:
        call = client.outbound_calls.get(call_id)
        status_counts[call.status] = status_counts.get(call.status, 0) + 1
 
    total = len(call_ids)
    done = status_counts["completed"] + status_counts["no_answer"] + \
           status_counts["failed"] + status_counts["voicemail"]
 
    print(f"\n=== Campaign Progress ({done}/{total} done) ===")
    for status, count in status_counts.items():
        if count > 0:
            print(f"  {status}: {count} ({count/total:.1%})")

---

5. Generate a results report

def generate_report(agent_id: str, date_from: str, date_to: str):
    # Request CSV export
    report = client.reports.create(
        type="outbound_campaign",
        agent_id=agent_id,
        date_from=date_from,
        date_to=date_to,
        format="csv",
        include_fields=[
            "conversationId", "toNumber", "status", "duration",
            "outcome", "sentiment", "cost", "scheduledFor", "metadata"
        ]
    )
 
    # Wait for generation
    while report.status == "generating":
        time.sleep(3)
        report = client.reports.get(report.id)
 
    # Download
    url = client.reports.download_url(report.id)
    import urllib.request
    urllib.request.urlretrieve(url, "campaign-results.csv")
    print("Report saved to campaign-results.csv")

---

Best practices

Rate limiting: Voisnap's outbound calls endpoint allows 30 calls/minute. Spread your scheduling over time rather than creating all calls at once.

Calling hours: Schedule calls only during business hours in the recipient's timezone. Voisnap does not automatically enforce calling hours — implement this in your scheduling logic.

No-answer handling: Use max_retries: 2 with retry_delay_minutes: 30 for appointment reminders. Don't retry more than twice for cold outreach.

Voicemail: If voicemail detection is enabled, set a concise voicemail.message on the agent — 15–20 seconds max. Include a callback number.