Skip to main content

​
Setup Checklist

Before your first call routes to your Speech Pipe, complete these steps once.

​
1. Create a Voice Profile

import asyncio
from unpod import AsyncClient

async def list_profiles():
    client = AsyncClient()
    profiles = await client.voice_profiles.list()
    for p in profiles:
        print(p.name)

asyncio.run(list_profiles())

​
2. Create a Speech Pipe

import asyncio
from unpod import AsyncClient

async def create_pipe():
    client = AsyncClient()
    pipe = await client.pipes.create(
        name="My Agent",
        voice_profile="vp_en_female_hd",   # use a profile_id from step 1
        agent_id="my-agent",           # MUST match AgentRunner(agent_id=...)
    )
    print(pipe.pipe_id)  # UUID - used for REST API calls
    return pipe

asyncio.run(create_pipe())
agent_id vs pipe.pipe_id
  • pipe.pipe_id - UUID assigned by Unpod, used in REST API calls
  • agent_id - string name you choose; must exactly match AgentRunner(agent_id=...)
These are different. Mismatching them is the most common first-run failure.
Replace "vp_en_female_hd" with a profile_id from Step 1. Run Step 1 first to see available profiles.

​
3. Assign a Phone Number

import asyncio
from unpod import AsyncClient

async def assign_number(pipe_id: str) -> None:
    client = AsyncClient()
    numbers = await client.numbers.list()
    if not numbers:
        print("No numbers available. Provision one in the Unpod dashboard first.")
        return
    await client.numbers.attach(numbers[0].id, pipe_id=pipe_id)
    print(f"Assigned {numbers[0].number}")

asyncio.run(assign_number("YOUR_PIPE_UUID"))

​
4. Start Your Runner

from unpod import AgentRunner

AgentRunner(
    entrypoint=entrypoint,
    agent_id="my-agent",   # must match the pipe's agent_id above
).start()

​
Full Setup Script

import asyncio
from unpod import AsyncClient, AgentRunner, CallContext
from unpod.adapters.langchain import LangChainAdapter

async def setup():
    client = AsyncClient()

    # List available voice profiles
    profiles = await client.voice_profiles.list()
    profile = profiles[0]

    # Create Speech Pipe
    pipe = await client.pipes.create(
        name="My Agent",
        voice_profile=profile.profile_id,
        agent_id="my-agent",
    )

    # Assign first available number
    numbers = await client.numbers.list()
    if numbers:
        await client.numbers.attach(numbers[0].id, pipe_id=pipe.pipe_id)

    print(f"Speech Pipe created: {pipe.pipe_id}")
    print(f"Number: {numbers[0].number if numbers else 'none assigned'}")

asyncio.run(setup())
Then start your runner in a separate process: 
async def entrypoint(ctx: CallContext) -> None:
    ctx.session.dialog_machine = LangChainAdapter(your_chain)
    await ctx.session.run()

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

​
Environment Variables

VariableRequiredDescription
UNPOD_API_KEYYesYour Unpod API key
UNPOD_SERVICE_BASE_URLNoDefault: https://api.unpod.io/platform; use http://localhost:8000/platform for local service testing
UNPOD_ORCHESTRATOR_URLNoDefault: wss://api.unpod.io