Outbound Calls

import asyncio
from unpod import AsyncClient

async def main():
    async with AsyncClient() as client:  # reads UNPOD_API_KEY
        call = await client.calls.create(
            agent_id="my-agent",         # or pipe_id= for older code
            to_number="+14155550100",    # E.164 format
            from_number="+18005551234",  # optional if a number is attached to the agent
            instructions="This is a reminder call for a dental appointment.",
            data={
                "patient_name": "Jane Smith",
                "appointment_date": "2026-06-10",
                "appointment_time": "09:30 AM",
                "doctor": "Dr. Patel",
            },
        )
        print("Call initiated:", call.call_id, call.status)
        # status is "pending" - calls.create() enqueues the call and returns
        # immediately. The call is dispatched asynchronously once the account
        # has free concurrency. Poll GET /calls/{id} (or use hooks) to watch
        # it advance to ringing -> active -> completed. See Call Lifecycle.

asyncio.run(main())

from superdialog import DialogMachine
from unpod import AgentRunner, CallContext

async def handle_call(ctx: CallContext) -> None:
    # Available for both inbound and outbound calls
    print(ctx.direction)        # "outbound"
    print(ctx.instructions)     # "This is a reminder call..."
    print(ctx.data)             # {"patient_name": "Jane Smith", ...}

    # Inject context into the dialog machine
    patient_name = ctx.data.get("patient_name", "there")
    appt_date = ctx.data.get("appointment_date", "soon")
    appt_time = ctx.data.get("appointment_time", "")

    ctx.session.dialog_machine = DialogMachine(
        "reminder.yaml",
        llm="anthropic/claude-haiku-4-5",
    )
    # Inject the personalization as an assist directive
    ctx.session.dialog_machine.assist(
        f"You are calling {patient_name}. "
        f"Their appointment is on {appt_date} at {appt_time}. "
        f"Confirm they will attend, or reschedule if needed."
    )

    await ctx.session.run()

runner = AgentRunner(
    entrypoint=handle_call,
    agent_id="my-agent",
)
runner.start()

import asyncio
from unpod import AsyncClient

async def main():
    async with AsyncClient() as client:  # reads UNPOD_API_KEY
        call = await client.calls.get("cal_...")
        print(call.call_id, call.status, call.direction)
        # status: "pending", "ringing", "active", "completed", "failed", "cancelled"

asyncio.run(main())

import asyncio
from unpod import AsyncClient

async def main():
    async with AsyncClient() as client:  # reads UNPOD_API_KEY
        # All outbound calls for one Speech Pipe row
        calls = await client.calls.list(
            pipe_id="pipe_...",
            status="completed",
        )
        for c in calls:
            print(c.call_id, c.to_number, c.status, c.duration_s)

asyncio.run(main())

import asyncio
from unpod import AsyncClient

async def main():
    async with AsyncClient() as client:  # reads UNPOD_API_KEY
        await client.calls.hangup("cal_...")
        print("Call terminated")

asyncio.run(main())

import asyncio
from unpod import AsyncClient

AGENT_ID = "my-agent"
FROM_NUMBER = "+18005551234"

contacts = [
    {"name": "Alice", "phone": "+14155550101", "appt": "June 10 9:30 AM"},
    {"name": "Bob",   "phone": "+14155550102", "appt": "June 10 2:00 PM"},
    {"name": "Carol", "phone": "+14155550103", "appt": "June 11 10:00 AM"},
]

async def dial_all(contacts: list[dict]) -> None:
    async with AsyncClient() as client:  # reads UNPOD_API_KEY
        for contact in contacts:
            call = await client.calls.create(
                agent_id=AGENT_ID,
                to_number=contact["phone"],
                from_number=FROM_NUMBER,
                instructions=(
                    f"Call {contact['name']} to confirm their appointment "
                    f"on {contact['appt']}."
                ),
                data=contact,
            )
            print(f"Dialing {contact['name']}: {call.call_id}")
            await asyncio.sleep(2)  # pace the campaign

asyncio.run(dial_all(contacts))

import asyncio
from unpod import AsyncClient

async def process_callback_queue(queue: list[dict]) -> None:
    async with AsyncClient() as client:  # reads UNPOD_API_KEY
        for request in queue:
            # Capacity is handled for you: the orchestrator rejects a dispatch
            # when the agent's runners are full
            call = await client.calls.create(
                agent_id="my-agent",
                to_number=request["phone"],
                from_number="+18005551234",
                instructions="This is a callback. The customer requested to be called back.",
                data={"ticket_id": request["ticket_id"], "reason": request["reason"]},
            )
            await db.update_callback_request(
                request["id"],
                call_id=call.call_id,
                status="dialing",
            )

import asyncio
from unpod import AsyncClient

async def get_transcript(session_id: str) -> None:
    async with AsyncClient() as client:  # reads UNPOD_API_KEY
        session = await client.transcripts.get(session_id)
        for entry in session.transcript:
            print(f"[{entry.role}] {entry.content}")

asyncio.run(get_transcript("sess_..."))