osmTalk Docs
SDKs

Python SDK

Official osmtalk package for Python 3.10+. Sync and async clients.

pip install osmtalk

Status — beta: the osmtalk Python package is not yet published to PyPI. The source is ready in packages/sdk-python/ but no PyPI release has been cut. Reach out to contact@osmapi.com for beta access, or use the TypeScript SDK (production-ready on npm as @osmapi/osmtalk-sdk) in the meantime.

Requires Python 3.10+. Built on httpx — battery-included for both sync and async use.

Basic usage

from osmtalk import Osmtalk

client = Osmtalk(api_key="...")

call = client.calls.outbound(
    agent_id="agent_xxx",
    phone_number_id="pn_xxx",
    destination="+919876543210",
    dynamic_variables={
        "first_name": "Arjun",
        "appointment_date": "Tuesday May 12 at 3 PM",
    },
)
print("Call started:", call["callId"])

Resources

client.agents       # list, get, create, update, delete, connect, publish_version, list_versions, get_version, rollback_to_version
client.calls        # outbound, get, end, transfer
client.campaigns    # CRUD, start/pause/resume/stop, upload_leads_csv, upload_leads, list_leads, report
client.dnc          # list, add, bulk_add, remove
client.eval         # simulate, create_test_case, list_test_cases, run_test_case, run_all, list_runs, get_run
client.settings     # get, get_storage, update_storage, get_webhook, update_webhook, get_compliance, update_compliance
client.platform     # get_rates, list_providers, get_presets, get_model_health, list_templates, get_template

Public catalog (no auth required)

# Quality presets with LIVE cost estimates (INR per minute, low/typical/high)
presets = client.platform.get_presets(channel="phone")
print(presets["global"]["balanced"]["costEstimate"])
# {'low': 2.1, 'typical': 3.4, 'high': 5.8, 'currency': 'INR', ...}

# Provider health — check BEFORE launching to surface flaky providers
health = client.platform.get_model_health()
if health["openai"]["status"] == "down":
    # switch to anthropic before placing the call
    ...

# Industry-organized agent templates
templates = client.platform.list_templates(category="healthcare")
for t in templates["templates"]:
    print(t["id"], "-", t["tagline"])

Common patterns

Outbound campaign

camp = client.campaigns.create(
    name="Q2 Renewals",
    agent_id="agent_xxx",
    phone_number_id="pn_xxx",
    maxConcurrent=5,
    schedule={"timezone": "Asia/Kolkata", "windowStart": "10:00", "windowEnd": "18:00"},
    retryPolicy={"maxAttempts": 3, "backoffMinutes": 60, "retryOn": ["no_answer", "busy"]},
    webhookUrl="https://your-crm/webhooks/osmtalk",
)

with open("leads.csv") as f:
    print(client.campaigns.upload_leads_csv(camp["id"], f.read()))

client.campaigns.start(camp["id"])
print(client.campaigns.report(camp["id"])["counts"])

Async client

import asyncio
from osmtalk import AsyncOsmtalk

async def main():
    async with AsyncOsmtalk(api_key="...") as client:
        result = await client.calls.outbound(
            agent_id="agent_xxx",
            phone_number_id="pn_xxx",
            destination="+919876543210",
        )
        print(result)

asyncio.run(main())

Webhook verification

The SDK ships a verify_webhook_signature helper — constant-time, validates the sha256=<hex> shape, fails closed on missing input.

from osmtalk import verify_webhook_signature
from flask import Flask, request, abort

SECRET = "your-webhook-secret"
app = Flask(__name__)

@app.post("/webhooks/osmtalk")
def webhook():
    if not verify_webhook_signature(
        request.data,                              # bytes — the RAW body
        request.headers.get("X-OsmTalk-Signature"),
        SECRET,
    ):
        abort(401)
    event = request.get_json()
    if event["event"] == "campaign.lead_completed" and event["lead"]["disposition"] == "qualified":
        # Push to your CRM
        ...
    return {"ok": True}

FastAPI version:

from fastapi import FastAPI, Request, HTTPException
from osmtalk import verify_webhook_signature

app = FastAPI()
SECRET = "your-webhook-secret"

@app.post("/webhooks/osmtalk")
async def webhook(request: Request):
    raw = await request.body()  # bytes
    if not verify_webhook_signature(raw, request.headers.get("x-osmtalk-signature"), SECRET):
        raise HTTPException(401)
    event = await request.json()
    # ... handle event
    return {"ok": True}

Important: verify against the raw request body, never the parsed JSON. Frameworks that auto-parse JSON usually expose the raw bytes via request.data (Flask) or await request.body() (FastAPI/Starlette).

Error handling

from osmtalk import Osmtalk, OsmtalkError

client = Osmtalk(api_key="...")
try:
    client.calls.outbound(agent_id="...", phone_number_id="...", destination="...")
except OsmtalkError as e:
    print(f"HTTP {e.status}: {e.body}")

Options

Osmtalk(
    api_key="...",
    base_url="https://api.osmtalk.com",  # override for self-hosted
    timeout=30.0,                         # seconds
)

Versioning

Pin a specific agent version on every call:

client.calls.outbound(
    agent_id="agent_xxx",
    phone_number_id="pn_xxx",
    destination="+919876543210",
    agent_version=7,
)

Simulation (pre-launch testing)

sim = client.eval.simulate("agent_xxx", [
    {"role": "user", "content": "Hi, who's calling?"},
    {"role": "user", "content": "Tell me about renewal options"},
])

for turn in sim["transcript"]:
    print(f"{turn['role']}: {turn['content']}")