Phone Numbers

import asyncio
from unpod import AsyncClient

async def main():
    async with AsyncClient() as client:  # reads UNPOD_API_KEY
        numbers = await client.numbers.list()
        for n in numbers:
            print(n.number_id, n.number, n.status, n.pipe_id or "unattached")

asyncio.run(main())

from unpod import AsyncClient
from unpod.models import TrunkCreate, ByoConfigCreate

client = AsyncClient()

trunk = await client.trunks.create(TrunkCreate(
    name="tata-byo",
    type="byo",
    byo_config=ByoConfigCreate(
        provider="tata",
        sip_domain="sip.tata.in",
        auth_username="user",
        auth_password="secret",
        transport="tls",   # default
    ),
))

trunks = await client.trunks.list()
await client.trunks.delete(trunk.trunk_id)

result = await client.numbers.sync()   # {"synced": int, "new": int}

import asyncio
from unpod import AsyncClient

async def main():
    async with AsyncClient() as client:  # reads UNPOD_API_KEY
        # All numbers
        numbers = await client.numbers.list()

        # Filter by status or country
        active = await client.numbers.list(status="active")
        us_numbers = await client.numbers.list(country="US")
        byon = await client.numbers.list(trunk_type="byo")

        for n in numbers:
            print(n.number_id, n.number, n.status, n.pipe_id or "unattached")

asyncio.run(main())

import asyncio
from unpod import AsyncClient

async def main():
    async with AsyncClient() as client:  # reads UNPOD_API_KEY
        number = await client.numbers.attach(
            number_id="num_...",
            pipe_id="pipe_...",
        )
        print("Attached:", number.number, "-> pipe", number.pipe_id)

asyncio.run(main())

import asyncio
from unpod import AsyncClient

async def main():
    async with AsyncClient() as client:  # reads UNPOD_API_KEY
        number = await client.numbers.detach("num_...")
        print("Detached:", number.number)

asyncio.run(main())

import asyncio
from unpod import AsyncClient

async def attach_first_free(pipe_id: str) -> None:
    async with AsyncClient() as client:  # reads UNPOD_API_KEY
        numbers = await client.numbers.list(status="active")
        free = [n for n in numbers if n.pipe_id is None]
        if not free:
            raise RuntimeError("No free numbers available")
        await client.numbers.attach(number_id=free[0].number_id, pipe_id=pipe_id)
        print("Attached", free[0].number)

asyncio.run(attach_first_free("pipe_..."))

import asyncio
from unpod import AsyncClient

async def rotate(numbers: list[str], pipes: list[str]) -> None:
    async with AsyncClient() as client:  # reads UNPOD_API_KEY
        for num_id, pipe_id in zip(numbers, pipes):
            await client.numbers.attach(number_id=num_id, pipe_id=pipe_id)
            print(f"{num_id} -> {pipe_id}")

asyncio.run(rotate(["num_a", "num_b"], ["pipe_x", "pipe_y"]))