Python SDK for the Callbotics Core API
Project description
Callbotics SDK
Official Python SDK for the Callbotics Core API.
Installation
pip install callbotics-sdk
# With WebSocket support
pip install callbotics-sdk[websocket]
Quick Start
from callbotics_sdk import CallboticsClient
# Initialize the client
client = CallboticsClient(
base_url="https://api.callbotics.ai",
token="your-bearer-token"
)
# List campaigns
campaigns = client.campaigns.list(page=1, per_page=10)
print(f"Found {campaigns.total_records} campaigns")
# Create a new campaign
campaign = client.campaigns.create(
name="My Campaign",
organisation_id="org-123",
agent_id="agent-456",
direction="outbound",
concurrency=5
)
# Start the campaign
client.campaigns.start(campaign["id"])
Authentication
The SDK supports two authentication methods:
Bearer Token
client = CallboticsClient(
base_url="https://api.callbotics.ai",
token="your-bearer-token"
)
API Key
client = CallboticsClient(
base_url="https://api.callbotics.ai",
api_key="your-api-key"
)
Update Token at Runtime
client.set_token("new-token")
Available Resources
| Resource | Description |
|---|---|
client.users |
User management |
client.organisations |
Organisation management |
client.campaigns |
Campaign management |
client.calls |
Call management |
client.agents |
Agent configuration |
client.contacts |
Contact list management |
client.prompts |
Prompt templates |
client.voice_configs |
Voice configuration |
client.llm_configs |
LLM configuration |
client.stt_configs |
Speech-to-Text configuration |
client.telephony |
Telephony configuration |
client.billing |
Billing and usage |
client.call_logs |
Call logs and reports |
client.analytics |
Dashboard analytics |
client.workflows |
Workflow management |
client.roles |
Role management |
client.actions |
Action management |
client.live_monitoring |
Real-time monitoring (REST + WebSocket) |
Examples
Complete Outbound Campaign Setup
This example shows the full workflow for creating and running an outbound calling campaign.
from callbotics_sdk import CallboticsClient, CallboticsError
# Initialize client
client = CallboticsClient(
base_url="https://api.callbotics.ai",
token="your-bearer-token"
)
# Step 1: Create the campaign
campaign = client.campaigns.create(
name="Q1 Sales Outreach",
organisation_id="org-123",
agent_id="agent-456", # Your AI agent ID
direction="outbound",
concurrency=10, # Max simultaneous calls
telephony_config_id="tel-789", # Telephony provider config
from_number="+18001234567" # Caller ID number
)
campaign_id = campaign["id"]
print(f"Campaign created: {campaign_id}")
# Step 2: Create a contact list for the campaign
contact_list = client.contacts.create(
list_name="Sales Leads - January",
campaign_id=campaign_id
)
contact_list_id = contact_list["id"]
print(f"Contact list created: {contact_list_id}")
# Step 3: Upload contacts from Excel/CSV file
with open("leads.xlsx", "rb") as f:
result = client.contacts.upload(contact_list_id, f, filename="leads.xlsx")
print(f"Contacts uploaded: {result}")
# Step 4: Start the campaign
client.campaigns.start(campaign_id)
print("Campaign started!")
# Step 5: Monitor campaign progress
import time
while True:
campaign_status = client.campaigns.get(campaign_id)
status = campaign_status.get("status")
print(f"Status: {status}")
if status == "completed":
print("Campaign completed!")
break
elif status == "paused":
print("Campaign is paused")
break
time.sleep(30) # Check every 30 seconds
Outbound Calling - Single Call
Make a single outbound call with custom contact data.
from callbotics_sdk import CallboticsClient
client = CallboticsClient(
base_url="https://api.callbotics.ai",
token="your-bearer-token"
)
# Make a single outbound call
call = client.calls.create(
campaign_id="campaign-123",
to_number="+14155551234",
from_number="+18001234567", # Optional: override campaign default
contact_data={ # Custom data available to AI agent
"first_name": "John",
"last_name": "Smith",
"company": "Acme Corp",
"appointment_date": "2024-02-15",
"account_number": "ACC-12345"
}
)
print(f"Call initiated: {call['id']}")
print(f"Status: {call.get('status', 'queued')}")
Outbound Calling - Batch Calls
Queue multiple calls programmatically without using contact lists.
from callbotics_sdk import CallboticsClient, CallboticsError
client = CallboticsClient(
base_url="https://api.callbotics.ai",
token="your-bearer-token"
)
# List of contacts to call
contacts = [
{"phone": "+14155551001", "name": "Alice Johnson", "account": "A001"},
{"phone": "+14155551002", "name": "Bob Williams", "account": "A002"},
{"phone": "+14155551003", "name": "Carol Davis", "account": "A003"},
]
campaign_id = "campaign-123"
queued_calls = []
for contact in contacts:
try:
call = client.calls.create(
campaign_id=campaign_id,
to_number=contact["phone"],
contact_data={
"name": contact["name"],
"account_number": contact["account"]
}
)
queued_calls.append(call)
print(f"Queued call to {contact['name']}: {call['id']}")
except CallboticsError as e:
print(f"Failed to queue call to {contact['name']}: {e.message}")
print(f"\nTotal calls queued: {len(queued_calls)}")
Campaign Management - Full Lifecycle
from callbotics_sdk import CallboticsClient
client = CallboticsClient(
base_url="https://api.callbotics.ai",
token="your-bearer-token"
)
# List all campaigns for an organization
campaigns = client.campaigns.list(
organisation_id="org-123",
status="ongoing", # Filter: pending, ongoing, paused, completed
direction="outbound", # Filter: outbound, inbound
page=1,
per_page=20
)
print(f"Found {campaigns.total_records} campaigns")
for campaign in campaigns.items:
print(f" - {campaign['name']} ({campaign['status']})")
# Get specific campaign details
campaign = client.campaigns.get("campaign-123")
print(f"\nCampaign: {campaign['name']}")
print(f" Direction: {campaign['direction']}")
print(f" Concurrency: {campaign['concurrency']}")
print(f" Status: {campaign['status']}")
# Campaign controls
campaign_id = "campaign-123"
# Start campaign - begins dialing contacts
client.campaigns.start(campaign_id)
print("Campaign started")
# Pause campaign - stops new calls, existing calls continue
client.campaigns.pause(campaign_id)
print("Campaign paused")
# Reset campaign - resets all contact statuses to retry
client.campaigns.reset(campaign_id)
print("Campaign reset")
# Update campaign settings
client.campaigns.update(
campaign_id=campaign_id,
name="Updated Campaign Name",
concurrency=15 # Increase concurrent calls
)
print("Campaign updated")
# Check available concurrency
concurrency = client.campaigns.get_available_concurrency(campaign_id)
print(f"Available concurrency: {concurrency}")
Contact List Management
from callbotics_sdk import CallboticsClient
client = CallboticsClient(
base_url="https://api.callbotics.ai",
token="your-bearer-token"
)
campaign_id = "campaign-123"
# Get sample Excel template for proper format
sample_excel = client.contacts.get_sample_excel(campaign_id)
with open("contact_template.xlsx", "wb") as f:
f.write(sample_excel)
print("Downloaded contact template")
# Create a new contact list
contact_list = client.contacts.create(
list_name="February Leads",
campaign_id=campaign_id
)
list_id = contact_list["id"]
# Upload contacts from file
with open("february_leads.xlsx", "rb") as f:
result = client.contacts.upload(list_id, f, filename="february_leads.xlsx")
print(f"Upload result: {result}")
# List all contact lists for a campaign
contact_lists = client.contacts.list(campaign_id=campaign_id)
for cl in contact_lists.items:
print(f" - {cl['list_name']}: {cl.get('total_contacts', 0)} contacts")
# Get active (pending) contacts
active = client.contacts.get_active_contacts(campaign_id)
print(f"Active contacts waiting to be called: {len(active)}")
# Get available timezones for scheduling
timezones = client.contacts.get_timezones()
print(f"Available timezones: {len(timezones)}")
Complete AI Agent Setup (LLM + TTS + STT)
This example shows how to configure LLM, Voice (TTS), STT, create an AI agent, and use it in a campaign.
from callbotics_sdk import CallboticsClient
client = CallboticsClient(
base_url="https://api.callbotics.ai",
token="your-bearer-token"
)
# ============================================================
# STEP 1: Create LLM Configuration
# ============================================================
llm_config = client.llm_configs.create(
name="GPT-4 Sales Agent",
provider="openai", # openai, anthropic, etc.
model="gpt-4-turbo",
temperature=0.7,
max_tokens=1024
)
llm_config_id = llm_config["id"]
print(f"LLM Config created: {llm_config_id}")
# ============================================================
# STEP 2: Create Voice Configuration (TTS - Text-to-Speech)
# ============================================================
voice_config = client.voice_configs.create(
name="Professional Female Voice",
provider="elevenlabs", # elevenlabs, azure, google, etc.
voice_id="21m00Tcm4TlvDq8ikWAM" # Provider-specific voice ID
)
voice_config_id = voice_config["id"]
print(f"Voice Config created: {voice_config_id}")
# ============================================================
# STEP 3: Create STT Configuration (Speech-to-Text)
# ============================================================
stt_config = client.stt_configs.create(
name="English STT - Deepgram",
provider="deepgram", # deepgram, google, azure, etc.
language="en-US"
)
stt_config_id = stt_config["id"]
print(f"STT Config created: {stt_config_id}")
# ============================================================
# STEP 4: Create Prompt Template
# ============================================================
prompt = client.prompts.create(
name="Sales Outreach Prompt",
system_prompt="""You are a friendly sales representative for Acme Corp.
Your goal is to schedule a product demo with the prospect.
Be professional, helpful, and respect their time.
If they're not interested, thank them politely and end the call.""",
initial_message="Hi {{first_name}}, this is Sarah from Acme Corp. How are you today?"
)
prompt_id = prompt["id"]
print(f"Prompt created: {prompt_id}")
# ============================================================
# STEP 5: Create AI Agent (combines all configurations)
# ============================================================
agent = client.agents.create(
name="Sales Demo Scheduler",
organisation_id="org-123",
prompt_id=prompt_id,
llm_config_id=llm_config_id,
voice_config_id=voice_config_id,
stt_config_id=stt_config_id
)
agent_id = agent["id"]
print(f"Agent created: {agent_id}")
# ============================================================
# STEP 6: Create Campaign with the Agent
# ============================================================
campaign = client.campaigns.create(
name="Q1 Demo Outreach",
organisation_id="org-123",
agent_id=agent_id, # Use our configured agent
direction="outbound",
concurrency=10,
telephony_config_id="tel-789",
from_number="+18001234567"
)
campaign_id = campaign["id"]
print(f"Campaign created: {campaign_id}")
# ============================================================
# STEP 7: Upload contacts and start campaign
# ============================================================
contact_list = client.contacts.create(
list_name="Demo Prospects",
campaign_id=campaign_id
)
with open("prospects.xlsx", "rb") as f:
client.contacts.upload(contact_list["id"], f)
# Start the campaign
client.campaigns.start(campaign_id)
print("Campaign started with AI agent!")
Managing LLM Configurations
# List available LLM configs
llm_configs = client.llm_configs.list()
for config in llm_configs.items:
print(f"{config['name']} - {config['provider']}/{config['model']}")
# Update LLM config
client.llm_configs.update(
config_id="llm-123",
temperature=0.5,
max_tokens=2048
)
# Assign LLM to a company/organization
client.llm_configs.assign_to_company(
config_id="llm-123",
company_id="company-456"
)
# Get LLMs assigned to a company
assigned = client.llm_configs.get_assigned_llms("company-456")
Managing Voice Configurations (TTS)
# List voice configs
voices = client.voice_configs.list()
for voice in voices.items:
print(f"{voice['name']} - {voice['provider']}")
# Create voice config with additional settings
voice = client.voice_configs.create(
name="Custom Voice",
provider="elevenlabs",
voice_id="voice-id-from-provider",
stability=0.5,
similarity_boost=0.75
)
# Assign to company
client.voice_configs.assign_to_company(
config_id=voice["id"],
company_id="company-456"
)
Managing STT Configurations
# List STT configs
stt_configs = client.stt_configs.list()
for stt in stt_configs.items:
print(f"{stt['name']} - {stt['provider']} ({stt.get('language', 'en')})")
# Create STT config
stt = client.stt_configs.create(
name="Spanish STT",
provider="deepgram",
language="es"
)
# Assign to company
client.stt_configs.assign_to_company(
config_id=stt["id"],
company_id="company-456"
)
Updating Agent Configuration
# Update agent to use different LLM/Voice/STT
client.agents.update(
agent_id="agent-123",
llm_config_id="new-llm-config-id",
voice_config_id="new-voice-config-id",
stt_config_id="new-stt-config-id"
)
# Update just the prompt
client.agents.update(
agent_id="agent-123",
prompt_id="new-prompt-id"
)
User Management
# List users
users = client.users.list(page=1, per_page=20)
for user in users.items:
print(f"{user['email']} - {user['role_name']}")
# Create a user
new_user = client.users.create(
email="john@example.com",
password="secure-password",
first_name="John",
last_name="Doe",
role_id="role-123"
)
# Update a user
client.users.update(user_id="user-123", first_name="Jane")
# Delete a user
client.users.delete(user_id="user-123")
Analytics
# Get call counts
counts = client.analytics.get_calls_count(
org_id="org-123",
start_date="2024-01-01",
end_date="2024-01-31"
)
# Get disposition distribution
distribution = client.analytics.get_disposition_distribution(
campaign_id="campaign-123"
)
Real-Time Monitoring (WebSocket)
The SDK includes WebSocket support for real-time monitoring.
Live Campaigns
import asyncio
async def monitor_campaigns():
async with client.live_monitoring.connect_campaigns() as ws:
async for message in ws:
if message["type"] == "live_campaigns":
data = message["data"]
print(f"Active campaigns: {data['total_active_campaigns']}")
for campaign in data["campaigns"]:
print(f" - {campaign['campaign_name']}: {campaign['active_calls_count']} calls")
asyncio.run(monitor_campaigns())
Live Transcript
async def monitor_transcript(call_id: str):
async with client.live_monitoring.connect_transcript(call_id) as ws:
async for message in ws:
if message["type"] == "transcript_init":
print("Initial transcript received")
elif message["type"] == "transcript_delta":
for item in message["data"]["items"]:
# item contains the transcript message
print(item)
asyncio.run(monitor_transcript("call-123"))
Campaign Calls
async def monitor_campaign_calls(campaign_id: str):
async with client.live_monitoring.connect_campaign_calls(campaign_id) as ws:
async for message in ws:
if message["type"] == "campaign_calls":
data = message["data"]
print(f"Calls in {data['campaign_name']}: {data['calls_count']}")
asyncio.run(monitor_campaign_calls("campaign-123"))
Pagination
List methods return a PaginatedResponse object:
response = client.campaigns.list(page=1, per_page=10)
print(f"Total: {response.total_records}")
print(f"Page: {response.current_page} of {response.total_pages}")
print(f"Has next: {response.has_next}")
print(f"Has prev: {response.has_prev}")
for item in response.items:
print(item)
Error Handling
from callbotics_sdk import (
CallboticsError,
AuthenticationError,
NotFoundError,
ValidationError,
RateLimitError,
ServerError,
)
try:
campaign = client.campaigns.get("invalid-id")
except NotFoundError as e:
print(f"Campaign not found: {e.message}")
except ValidationError as e:
print(f"Validation error: {e.message}")
print(f"Errors: {e.errors}")
except AuthenticationError as e:
print(f"Auth failed: {e.message}")
except RateLimitError as e:
print(f"Rate limited. Retry after {e.retry_after} seconds")
except ServerError as e:
print(f"Server error: {e.message}")
except CallboticsError as e:
print(f"API error [{e.status_code}]: {e.message}")
Context Manager Support
# Sync
with CallboticsClient(base_url="...", token="...") as client:
campaigns = client.campaigns.list()
# Async
async with CallboticsClient(base_url="...", token="...") as client:
campaigns = client.campaigns.list()
Requirements
- Python 3.9+
- httpx >= 0.24.0
- websockets >= 11.0 (optional, for WebSocket support)
License
MIT License
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file callbotics_sdk-1.1.0.tar.gz.
File metadata
- Download URL: callbotics_sdk-1.1.0.tar.gz
- Upload date:
- Size: 25.6 kB
- Tags: Source
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.12.12
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
01e51848ebcfa9ed6f2e04a95554fe2f80b5647bd4d5a85de362646541b7c583
|
|
| MD5 |
35f45f51057d4c8a1820296a4b73324f
|
|
| BLAKE2b-256 |
536536f318497b7e5d29fbf9b2da60042b75514db32424e95eb7cf83f3327b8b
|
File details
Details for the file callbotics_sdk-1.1.0-py3-none-any.whl.
File metadata
- Download URL: callbotics_sdk-1.1.0-py3-none-any.whl
- Upload date:
- Size: 33.0 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? No
- Uploaded via: twine/6.2.0 CPython/3.12.12
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
f8b264b3e2e462633ee69ab0aee1e1592e81f8d5a2657db509e7d42a43acdb93
|
|
| MD5 |
7578bbf0e55345ad7f37c352ecbc87f2
|
|
| BLAKE2b-256 |
8a19e0d4d30e813feb020945819a60da7369558e56f26cce116679726e8146e5
|