Python Examples

pip install httpx

import httpx
import os

class VoiceMatrixClient:
    def __init__(self, api_key: str = None):
        self.api_key = api_key or os.environ.get('VOICEMATRIX_API_KEY')
        self.base_url = 'https://api.voicematrix.ai/api/v1/ext'
        self.client = httpx.Client(
            base_url=self.base_url,
            headers={
                'X-API-Key': self.api_key,
                'Content-Type': 'application/json'
            }
        )

    def list_calls(self, limit=20, offset=0, **filters):
        params = {'limit': limit, 'offset': offset, **filters}
        response = self.client.get('/calls', params=params)
        response.raise_for_status()
        return response.json()

    def get_call(self, call_id: str):
        response = self.client.get(f'/calls/{call_id}')
        response.raise_for_status()
        return response.json()

    def make_call(self, to: str, agent_id: str, metadata: dict = None):
        data = {'to': to, 'agent_id': agent_id}
        if metadata:
            data['metadata'] = metadata
        response = self.client.post('/calls', json=data)
        response.raise_for_status()
        return response.json()

    def create_lead(self, name: str, phone: str, **kwargs):
        data = {'name': name, 'phone': phone, **kwargs}
        response = self.client.post('/leads', json=data)
        response.raise_for_status()
        return response.json()

# Usage
vm = VoiceMatrixClient()

# List recent calls
calls = vm.list_calls(limit=10, status='completed')
print(f"Found {calls['pagination']['total']} calls")

# Make outbound call
call = vm.make_call(
    to='+972501234567',
    agent_id='your-agent-uuid',
    metadata={'campaign': 'summer_sale'}
)
print(f"Call initiated: {call['id']}")

# Create lead
lead = vm.create_lead(
    name='John Doe',
    phone='+972501234567',
    email='[email protected]',
    source='python_script'
)
print(f"Lead created: {lead['id']}")

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

app = FastAPI()
WEBHOOK_SECRET = os.environ.get('WEBHOOK_SECRET')  # the whole whsec_... string, returned once on create

def verify_signature(body: bytes, signature: str, timestamp: str, tolerance: int = 300) -> bool:
    # X-Webhook-Signature = "v1=" + hex(HMAC-SHA256(secret, f"{timestamp}.{body}"))
    if not signature.startswith('v1=') or not timestamp.isdigit():
        return False
    if abs(time.time() - int(timestamp)) > tolerance:  # replay window — your choice, not enforced server-side
        return False
    expected = 'v1=' + hmac.new(
        WEBHOOK_SECRET.encode(),
        timestamp.encode() + b'.' + body,  # the raw bytes as received, never re-serialized
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(signature, expected)

@app.post('/webhooks/voicematrix')
async def handle_webhook(request: Request):
    signature = request.headers.get('X-Webhook-Signature', '')
    timestamp = request.headers.get('X-Webhook-Timestamp', '')
    body = await request.body()

    if not verify_signature(body, signature, timestamp):
        raise HTTPException(401, 'Invalid signature')

    data = await request.json()
    event = data.get('event')

    # Real catalog is exactly 5 events — see ../webhooks.md. "call.completed" was never one of them.
    if event == 'call.ended':
        print(f"Call ended: {data['data']['id']}")
    elif event == 'call.started':
        print(f"Call started: {data['data']['id']}")
    elif event == 'lead.created':
        print(f"New lead: {data['data']['name']}")
    elif event in ('lead.updated', 'agent.error'):
        pass

    return {'received': True}

def call(self, method, path, **kwargs):
    response = self.client.request(method, path, **kwargs)
    if response.status_code >= 400:
        body = response.json().get('error', {})
        raise RuntimeError(f"{body.get('code')}: {body.get('message')}")
    return response.json()