Backend Integration Guide

This guide covers how to integrate Agent Studio with your backend systems to trigger voice calls from events and receive call status notifications.

Overview

Agent Studio provides two integration patterns:

Outbound: Your backend triggers calls via REST API
Inbound: Agent Studio sends webhooks for call events

┌─────────────┐     REST API      ┌───────────────┐
│   Backend   │ ────────────────► │ Agent Studio  │
│   Service   │                   │     API       │
└─────────────┘                   └───────────────┘
       ▲                                 │
       │         Webhooks                │
       └─────────────────────────────────┘

Authentication

Creating API Keys

Navigate to Dashboard → Settings → API Keys
Click Create API Key
Select scopes:
- calls:initiate - Trigger calls
- calls:read - Read call status
- workflows:read - List available workflows
- webhooks:write - Configure webhooks
Choose environment: Live (production) or Test (staging)
Save the key securely - it's only shown once

Using API Keys

Include the key in the Authorization header:

curl -X POST https://api.yourdomain.com/api/v1/calls \
  -H "Authorization: Bearer as_live_your_api_key_here" \
  -H "Content-Type: application/json" \
  -d '{"workflow_slug": "daily-checkup", "user_id": "user-123"}'

Triggering Calls

Basic Call Dispatch

import httpx

AGENT_STUDIO_URL = "https://api.yourdomain.com"
API_KEY = "as_live_your_api_key_here"

async def dispatch_call(
    workflow_slug: str,
    user_id: str,
    context: dict | None = None,
) -> dict:
    """Dispatch a voice call."""
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{AGENT_STUDIO_URL}/api/v1/calls",
            headers={"Authorization": f"Bearer {API_KEY}"},
            json={
                "workflow_slug": workflow_slug,
                "user_id": user_id,
                "user_context": context or {},
            },
        )
        response.raise_for_status()
        return response.json()

# Usage
result = await dispatch_call(
    workflow_slug="onboarding",
    user_id="user-123",
    context={
        "user_name": "John",
    },
)
print(f"Call dispatched: {result['call_id']}")

Response

{
  "call_id": "550e8400-e29b-41d4-a716-446655440000",
  "room_name": "call-550e8400",
  "token": "eyJ...",  // LiveKit token for joining
  "url": "wss://livekit.yourdomain.com"
}

Testing Single Agents

For testing individual agents without a full workflow:

response = await client.post(
    f"{AGENT_STUDIO_URL}/api/v1/calls/test-agent",
    headers={"Authorization": f"Bearer {API_KEY}"},
    json={
        "agent_name": "greeter",
        "user_id": "test-user",
        "user_context": {"test": True},
    },
)

Event-Driven Integration

Kafka Consumer Example

Trigger calls from Kafka events:

"""
Kafka consumer that triggers calls on user events.

Install: pip install aiokafka httpx
"""
import asyncio
import json
from aiokafka import AIOKafkaConsumer
import httpx

AGENT_STUDIO_URL = "https://api.yourdomain.com"
API_KEY = "as_live_your_api_key_here"

async def dispatch_call(client: httpx.AsyncClient, **kwargs) -> dict:
    response = await client.post(
        f"{AGENT_STUDIO_URL}/api/v1/calls",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json=kwargs,
    )
    response.raise_for_status()
    return response.json()

async def handle_user_signup(event: dict, client: httpx.AsyncClient):
    """Trigger onboarding call for new users."""
    result = await dispatch_call(
        client,
        workflow_slug="onboarding",
        user_id=event["user_id"],
        user_context={
            "user_name": event.get("name", "there"),
            "signup_time": event.get("timestamp"),
        },
    )
    print(f"Dispatched onboarding call {result['call_id']} for {event['user_id']}")

async def main():
    consumer = AIOKafkaConsumer(
        "user-events",
        bootstrap_servers="localhost:9092",
        group_id="agent-studio-trigger",
        value_deserializer=lambda m: json.loads(m.decode()),
    )
    
    await consumer.start()
    
    async with httpx.AsyncClient(timeout=30) as client:
        try:
            async for message in consumer:
                event = message.value
                event_type = event.get("type")
                
                if event_type == "user.signup":
                    await handle_user_signup(event, client)
                    
        finally:
            await consumer.stop()

if __name__ == "__main__":
    asyncio.run(main())

Cron Job Example

Schedule daily check-in calls:

"""
Daily check-in caller.

Run with cron: 0 9 * * * python daily_checkin.py
"""
import asyncio
import httpx

AGENT_STUDIO_URL = "https://api.yourdomain.com"
API_KEY = "as_live_your_api_key_here"

async def get_users_for_checkin() -> list[dict]:
    """Fetch users who need check-in calls today."""
    # Your logic to get users
    return [
        {"id": "user-1", "name": "Alice"},
        {"id": "user-2", "name": "Bob"},
    ]

async def dispatch_checkin(client: httpx.AsyncClient, user: dict):
    response = await client.post(
        f"{AGENT_STUDIO_URL}/api/v1/calls",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={
            "workflow_slug": "daily-checkin",
            "user_id": user["id"],
            "user_context": {"user_name": user["name"]},
            "metadata": {"source": "daily-cron"},
        },
    )
    if response.status_code == 201:
        print(f"Scheduled call for {user['name']}")
    else:
        print(f"Failed for {user['name']}: {response.text}")

async def main():
    users = await get_users_for_checkin()
    
    async with httpx.AsyncClient(timeout=30) as client:
        tasks = [dispatch_checkin(client, user) for user in users]
        await asyncio.gather(*tasks)

if __name__ == "__main__":
    asyncio.run(main())

Receiving Call Status Webhooks

Agent Studio can notify your backend when call lifecycle events occur. These webhooks are for call status tracking only - they notify you when calls start, complete, fail, etc.

Note: For business logic during calls (e.g., saving data to your API), use Tool Webhooks instead. Tool webhooks are triggered by agent actions during the call.

Configuring Call Status Webhooks

# Create webhook configuration
response = await client.post(
    f"{AGENT_STUDIO_URL}/api/v1/webhooks",
    headers={"Authorization": f"Bearer {API_KEY}"},
    json={
        "url": "https://your-backend.com/webhooks/agent-studio",
        "filter": {
            "events": ["call.completed", "call.failed"],
            "include_transcript": True,
            "include_metrics": True,
        },
    },
)

# Save the secret!
webhook_secret = response.json()["secret"]
print(f"Webhook secret: {webhook_secret}")

Webhook Payload

{
  "event": {
    "id": "evt_abc123",
    "type": "call.completed",
    "created": 1705680000,
    "tenant_id": "tenant-uuid",
    "livemode": true,
    "data": {
      "call_id": "call-uuid",
      "room_name": "room-abc",
      "workflow_slug": "onboarding",
      "user_id": "user-123",
      "status": "completed",
      "duration_seconds": 180,
      "agent_history": ["greeter", "support"],
      "transcript": [
        {"role": "assistant", "content": "Hello!", "agent": "greeter"},
        {"role": "user", "content": "Hi there"}
      ],
      "metrics": {
        "duration_seconds": 180,
        "agent_count": 2,
        "message_count": 15
      },
      "metadata": {"source": "kafka"}
    }
  },
  "delivery_id": "dlv_xyz789",
  "attempt": 1,
  "max_attempts": 5
}

Verifying Webhook Signatures

Critical: Always verify webhook signatures to prevent spoofing.

import hmac
import hashlib
import time
from fastapi import FastAPI, Request, HTTPException

app = FastAPI()
WEBHOOK_SECRET = "whsec_your_secret_here"

def verify_signature(payload: bytes, signature_header: str) -> bool:
    """Verify webhook signature."""
    try:
        parts = dict(p.split("=") for p in signature_header.split(","))
        timestamp = int(parts["t"])
        signature = parts["v1"]
    except (KeyError, ValueError):
        return False
    
    # Reject old timestamps (prevent replay attacks)
    if abs(time.time() - timestamp) > 300:  # 5 minute tolerance
        return False
    
    # Compute expected signature
    signed_payload = f"{timestamp}.{payload.decode()}"
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        signed_payload.encode(),
        hashlib.sha256,
    ).hexdigest()
    
    return hmac.compare_digest(signature, expected)

@app.post("/webhooks/agent-studio")
async def handle_webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("X-Webhook-Signature", "")
    
    if not verify_signature(body, signature):
        raise HTTPException(status_code=401, detail="Invalid signature")
    
    event = await request.json()
    event_type = event["event"]["type"]
    data = event["event"]["data"]
    
    if event_type == "call.completed":
        await handle_call_completed(data)
    elif event_type == "call.failed":
        await handle_call_failed(data)
    
    return {"status": "received"}

async def handle_call_completed(data: dict):
    """Track call completion for analytics/billing."""
    call_id = data["call_id"]
    user_id = data["user_id"]
    duration = data.get("duration_seconds", 0)
    workflow = data.get("workflow_slug")
    
    # Log for analytics
    print(f"Call {call_id} completed: user={user_id}, workflow={workflow}, duration={duration}s")
    
    # Track for billing, analytics, audit logs, etc.
    # await record_call_completion(call_id, user_id, duration)

async def handle_call_failed(data: dict):
    """Alert on failed calls for monitoring."""
    call_id = data["call_id"]
    user_id = data["user_id"]
    reason = data.get("status_reason")
    
    print(f"Call {call_id} failed: user={user_id}, reason={reason}")
    
    # Alert ops team, trigger retry logic, etc.
    # await alert_call_failure(call_id, user_id, reason)

Event Types

These are call lifecycle events for tracking call status:

Event	When	Use Case
`call.started`	Call initiated	Track call volume
`call.connected`	Agent connected	Monitor connection success
`call.completed`	Normal completion	Billing, analytics, audit
`call.failed`	Connection failed	Alerting, retry logic
`call.disconnected`	Unexpected drop	Quality monitoring
`call.agent.changed`	Agent handoff	Workflow analytics

Important: These webhooks are for call tracking only. Business data (e.g., saving user input) should be handled via Tool Webhooks during the call.

Polling for Status

If webhooks aren't suitable, poll for call status:

async def wait_for_completion(
    client: httpx.AsyncClient,
    call_id: str,
    timeout: float = 300,
) -> dict:
    """Poll until call completes."""
    terminal = {"completed", "failed", "disconnected", "timeout"}
    start = time.time()
    
    while time.time() - start < timeout:
        response = await client.get(
            f"{AGENT_STUDIO_URL}/api/v1/calls/{call_id}",
            headers={"Authorization": f"Bearer {API_KEY}"},
        )
        call = response.json()
        
        if call["status"] in terminal:
            return call
        
        await asyncio.sleep(2)
    
    raise TimeoutError(f"Call {call_id} did not complete")

Error Handling

Retry Strategy

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10),
)
async def dispatch_with_retry(**kwargs):
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{AGENT_STUDIO_URL}/api/v1/calls",
            headers={"Authorization": f"Bearer {API_KEY}"},
            json=kwargs,
        )
        response.raise_for_status()
        return response.json()

Common Errors

Status	Meaning	Action
401	Invalid API key	Check key is correct
403	Missing scope	Add required scope to key
404	Workflow not found	Verify workflow_slug
429	Rate limited	Implement backoff
500	Server error	Retry with backoff

Best Practices

Store webhook secrets securely - Use environment variables or secret managers
Verify all webhooks - Never trust unverified payloads
Use idempotency - Handle duplicate webhook deliveries gracefully
Log call IDs - Include call_id in your logs for debugging
Monitor failures - Alert on failed call dispatches
Use test environment - Test integrations with as_test_ keys first

Backend Integration Guide

On this page