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                │
       └─────────────────────────────────┘

Marketing Platforms: If you're integrating with CleverTap, see Marketing Platform Integration for a simpler webhook-based approach that doesn't require a custom backend.

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 - they notify you when calls start, complete, fail, etc.

For detailed webhook configuration, see Call Status Webhooks.

Note: For business logic during calls (e.g., saving user data to your API), use the api_call action in tools. See Tool Actions for details.

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

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

Tool API Call Authentication

When tools need to call your protected backend APIs during a call (e.g., logging user data, updating records), you need to pass authentication credentials through the user_context.

Authentication Flow

┌────────────────────────────────────────────────────────────────────────────┐
│                           AUTHENTICATION FLOW                               │
├────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│  1. DISPATCH TIME (your backend)                                           │
│  ┌─────────────────────────────────────────────────────────────────────┐   │
│  │  POST /api/v1/calls                                                  │   │
│  │  {                                                                   │   │
│  │    "workflow_slug": "meal-logging",                                  │   │
│  │    "user_id": "user-123",                                            │   │
│  │    "user_context": {                                                 │   │
│  │      "auth_token": "jwt-or-api-key",  ◄── Your backend passes this   │   │
│  │      "user_name": "John",                                            │   │
│  │      "id": "user-123"                                                │   │
│  │    }                                                                 │   │
│  │  }                                                                   │   │
│  └─────────────────────────────────────────────────────────────────────┘   │
│                              │                                              │
│                              ▼                                              │
│  2. STORED IN CALL                                                         │
│  ┌─────────────────────────────────────────────────────────────────────┐   │
│  │  Call.input_context = user_context                                   │   │
│  │  (encrypted at rest)                                                 │   │
│  └─────────────────────────────────────────────────────────────────────┘   │
│                              │                                              │
│                              ▼                                              │
│  3. TOOL EXECUTION (during call)                                           │
│  ┌─────────────────────────────────────────────────────────────────────┐   │
│  │  Tool action:                                                        │   │
│  │  {                                                                   │   │
│  │    "type": "api_call",                                               │   │
│  │    "url": "https://your-api.com/meals",                              │   │
│  │    "headers": {                                                      │   │
│  │      "Authorization": "Bearer {{user.auth_token}}"  ◄── Resolved!   │   │
│  │    }                                                                 │   │
│  │  }                                                                   │   │
│  └─────────────────────────────────────────────────────────────────────┘   │
│                              │                                              │
│                              ▼                                              │
│  4. HTTP REQUEST TO YOUR BACKEND                                           │
│  ┌─────────────────────────────────────────────────────────────────────┐   │
│  │  POST https://your-api.com/meals                                     │   │
│  │  Authorization: Bearer jwt-or-api-key                                │   │
│  └─────────────────────────────────────────────────────────────────────┘   │
│                                                                             │
└────────────────────────────────────────────────────────────────────────────┘

Implementation Example

1. Your backend dispatches a call with auth credentials:

async def dispatch_meal_logging_call(user_id: str, user: User):
    """Dispatch a meal logging call for a user."""
    
    # Generate a short-lived token for the voice session
    auth_token = generate_voice_session_token(user_id, expires_in=3600)
    
    response = await client.post(
        f"{AGENT_STUDIO_URL}/api/v1/calls",
        headers={"Authorization": f"Bearer {API_KEY}"},
        json={
            "workflow_slug": "meal-logging",
            "user_id": user_id,
            "user_context": {
                # Auth token for tool API calls
                "auth_token": auth_token,
                
                # User data for personalization
                "id": str(user.id),
                "name": user.display_name,
                "language": user.preferred_language,
            },
        },
    )
    return response.json()

2. Tool uses the auth token in headers:

{
  "name": "save_meal",
  "config": {
    "parameters": [
      {"name": "meal_data", "type": "string", "required": true}
    ],
    "actions": [
      {
        "type": "api_call",
        "url": "https://your-api.com/api/v1/meals",
        "method": "POST",
        "headers": {
          "Content-Type": "application/json",
          "Authorization": "Bearer {{user.auth_token}}"
        },
        "body": {
          "user_id": "{{user.id}}",
          "meal": "{{params.meal_data}}"
        },
        "timeout": 15,
        "retry_count": 3
      }
    ]
  }
}

3. Your backend validates the token:

@app.post("/api/v1/meals")
async def create_meal(
    request: Request,
    user: User = Depends(validate_voice_session_token),  # Validates auth_token
):
    """Create a meal record from voice call."""
    data = await request.json()
    
    # User is authenticated via the token passed in user_context
    meal = await create_meal_record(
        user_id=user.id,
        meal_data=data["meal"],
    )
    
    return {"meal_id": meal.id}

Security Best Practices

Use short-lived tokens - Generate tokens that expire after the expected call duration (e.g., 1 hour)
Scope tokens appropriately - The voice session token should only have permissions needed for the call (e.g., meals:write, not full user access)
Token types:
- JWT with expiry: Self-contained, no database lookup needed
- Opaque tokens: Require validation against your auth service
- Service-to-service API keys: If the tool calls internal services
Don't pass sensitive data unnecessarily - Only include what the tools actually need

Template Resolution

Tools can use {{path}} templates to resolve values from context:

Template	Resolves From	Example
`{{user.auth_token}}`	`user_context.auth_token`	`"Bearer xyz..."`
`{{user.id}}`	`user_context.id`	`"user-123"`
`{{params.meal_type}}`	Tool parameters	`"Breakfast"`
`{{workflow.logged_meals}}`	Workflow state	`["Breakfast", "Lunch"]`

See Tool Actions for the complete template syntax.

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
Use short-lived tokens for tool auth - Don't pass long-lived credentials in user_context

Backend Integration Guide

On this page