Skip to main content
The Python SDK is continually evolving, with new endpoints being added regularly. Stay tuned for updates!

Overview

The Roark Python SDK provides a streamlined way to interact with the Roark API. This guide covers the two most common workflows: uploading calls and running metrics, and executing simulation run plans.

Prerequisites

Before you begin, ensure you have:

Installation

Choose your preferred package manager:
pip install roark-analytics

Initialize the Client

import os
from roark_analytics import Roark

client = Roark(
    bearer_token=os.environ.get("ROARK_API_BEARER_TOKEN"),
)
Set ROARK_API_BEARER_TOKEN in your environment, or replace with your actual API key from Roark.

Upload a Call and Run Metrics

Upload a call recording, then run metric definitions against it using a collection job.
1

Upload a Call

call = client.call.create(
    recording_url="https://example.com/recording.mp3",
    started_at="2024-01-15T10:00:00Z",
    interface_type="PHONE",
    call_direction="INBOUND",
    agent={
        "name": "Support Agent",
        "customId": "agent-123",
    },
    customer={
        "phoneNumberE164": "+15551234567",
    },
    # Optional: custom properties for filtering
    properties={
        "department": "sales",
        "campaignId": "summer-2024",
    },
)
2

Get Metric Definitions

List available metric definitions to choose which ones to run:
definitions = client.metric.list_definitions()

# Find the metrics you want to collect
task_completion = next(
    m for m in definitions.data if m.metric_id == "task_completion"
)
satisfaction = next(
    m for m in definitions.data if m.metric_id == "customer_satisfaction"
)
3

Create a Metric Collection Job

Run the selected metrics against your uploaded call:
job = client.metric_collection_job.create(
    call_ids=[call.data.id],
    metrics=[
        {"id": task_completion.id},
        {"id": satisfaction.id},
    ],
)
4

Poll for Completion

import time

status = "PENDING"
while status not in ("COMPLETED", "FAILED"):
    result = client.metric_collection_job.get_by_id(job.data.id)
    status = result.data.status
    print(f"Progress: {result.data.completed_items}/{result.data.total_items}")
    if status not in ("COMPLETED", "FAILED"):
        time.sleep(2)
5

Fetch Results

metrics = client.call.list_metrics(call.data.id)
print(metrics.data)
If you have metric policies configured, metrics are collected automatically when calls are uploaded — no collection job needed.

Run a Simulation Run Plan

Create and execute a simulation run plan to systematically test your voice AI agent.
1

Create the Run Plan

Define scenarios, personas, agent endpoints, and metrics to include:
plan = client.simulation_run_plan.create(
    name="Production Regression Suite",
    direction="INBOUND",
    max_simulation_duration_seconds=300,
    scenarios=[
        {"id": "scenario-uuid-1"},
        {"id": "scenario-uuid-2"},
    ],
    personas=[
        {"id": "persona-uuid-1"},
        {"id": "persona-uuid-2"},
    ],
    agent_endpoints=[
        {"id": "agent-endpoint-uuid"},
    ],
    metrics=[
        {"id": "metric-definition-uuid-1"},
        {"id": "metric-definition-uuid-2"},
    ],
    iteration_count=1,
    max_concurrent_jobs=10,
    auto_run=True,  # Start immediately after creation
)

# plan.data.run_plan_job contains the job if auto_run is True
2

Or Start an Existing Plan

If you already have a run plan, start a new job for it:
job = client.simulation_run_plan_job.start("plan-uuid")
3

Monitor Progress

job_result = client.simulation_run_plan_job.get_by_id(
    job.data.simulation_run_plan_job_id
)

print(f"Status: {job_result.data.status}")
print(f"Jobs: {len(job_result.data.simulation_jobs)}")
4

View Results

Results including metric evaluations are available in the Roark dashboard, or you can fetch metrics for individual simulation calls via the API:
# Get metrics for a specific simulation call
metrics = client.call.list_metrics("simulation-call-id")

Additional Examples

Track function calls and tool usage during the conversation:
call = client.call.create(
    recording_url="https://example.com/recording.mp3",
    started_at="2024-01-15T10:00:00Z",
    interface_type="PHONE",
    call_direction="INBOUND",
    agent={
        "name": "Booking Agent",
        "customId": "booking-agent-1",
    },
    customer={
        "phoneNumberE164": "+15551234567",
    },
    tool_invocations=[
        {
            "name": "check_availability",
            "description": "Check available appointment slots",
            "startOffsetMs": 5000,
            "endOffsetMs": 5500,
            "parameters": {
                "date": "2024-01-20",
                "serviceType": "consultation",
            },
            "result": {"slots": ["9:00 AM", "2:00 PM", "4:00 PM"]},
            "agent": {"customId": "booking-agent-1"},
        },
        {
            "name": "book_appointment",
            "description": "Book an appointment for the customer",
            "startOffsetMs": 15000,
            "endOffsetMs": 15800,
            "parameters": {
                "date": "2024-01-20",
                "time": "2:00 PM",
                "customerName": "John Doe",
            },
            "result": "Appointment confirmed",
            "agent": {"customId": "booking-agent-1"},
        },
    ],
)
Reference an agent by its Roark ID or custom ID:
# By Roark ID
call = client.call.create(
    recording_url="https://example.com/recording.mp3",
    started_at="2024-01-15T10:00:00Z",
    interface_type="WEB",
    call_direction="OUTBOUND",
    agent={
        "roarkId": "550e8400-e29b-41d4-a716-446655440000",
    },
    customer={
        "phoneNumberE164": "+15551234567",
    },
)

# By custom ID
call2 = client.call.create(
    recording_url="https://example.com/recording.mp3",
    started_at="2024-01-15T10:00:00Z",
    interface_type="PHONE",
    call_direction="INBOUND",
    agent={
        "customId": "my-agent-id",
    },
    customer={
        "phoneNumberE164": "+15551234567",
    },
)
Run metrics across a batch of existing calls:
job = client.metric_collection_job.create(
    call_ids=[
        "call-uuid-1",
        "call-uuid-2",
        "call-uuid-3",
    ],
    metrics=[
        {"id": "metric-definition-uuid-1"},
        {"id": "metric-definition-uuid-2"},
    ],
)

# total_items = len(call_ids) * len(metrics)
print(f"Processing {job.data.total_items} items")
For asynchronous operations, use the async client:
import os
import asyncio
from roark_analytics import AsyncRoark

client = AsyncRoark(
    bearer_token=os.environ.get("ROARK_API_BEARER_TOKEN"),
)

async def main():
    call = await client.call.create(
        recording_url="https://example.com/recording.mp3",
        started_at="2024-01-15T10:00:00Z",
        interface_type="PHONE",
        call_direction="INBOUND",
        agent={"name": "Support Agent"},
        customer={"phoneNumberE164": "+15551234567"},
    )

    job = await client.metric_collection_job.create(
        call_ids=[call.data.id],
        metrics=[{"id": "metric-definition-uuid"}],
    )

asyncio.run(main())

Additional Resources

API Reference

Explore our comprehensive API documentation

Example Repository

View example implementations and use cases

PyPI Package

View package details and stats on PyPI

GitHub Repository

Browse the source code and contribute