Phone Agent (Outbound Calling)

View Source on GitHub

This example demonstrates how to implement a dedicated Phone Agent built for programmatic outbound calling.

You will use the RestClient to dispatch a phone call, pass custom parameters (like user demographics) into the create_session method dynamically via the variables tag, and handle the real-time AI conversation via the active phone agent pipeline!

Requirements

pip install "piopiy-ai[cartesia,deepgram,openai,silero]"
pip install piopiy

Ensure your .env contains:

# Agent Variables
AGENT_ID="your_agent_id"
AGENT_TOKEN="your_agent_token"
OPENAI_API_KEY="your_openai_key"
DEEPGRAM_API_KEY="your_deepgram_key"
CARTESIA_API_KEY="your_cartesia_key"

# Outbound Trigger Variables
PIOPIY_TOKEN="your_rest_api_token"
PIOPIY_NUMBER="your_purchased_number"
CUSTOMER_NUMBER="destination_phone_number"

How to Run

Save the first script below as phone_agent.py and run it in your primary terminal. It will securely connect to the Piopiy infrastructure and wait for inbound or outbound trigger events.

python phone_agent.py

While the server is running, save the second script as outbound_call.py and run it in a second terminal:

python outbound_call.py

The REST API will dial the CUSTOMER_NUMBER and dynamically send "Alice" to the Agent process. This modifies the agent's LLM prompt and greeting to personally address the recipient!

Full Scripts

phone_agent.py

This is the background process that streams the audio utilizing the VoiceAgent. Look closely at metadata in the create_session function to see how outbound variables are parsed.

import asyncio
import os
import dotenv

from piopiy.agent import Agent
from piopiy.voice_agent import VoiceAgent
from piopiy.services.deepgram.stt import DeepgramSTTService
from piopiy.services.openai.llm import OpenAILLMService
from piopiy.services.cartesia.tts import CartesiaTTSService

dotenv.load_dotenv()

async def create_session(agent_id: str, call_id: str, from_number: str, to_number: str, **kwargs):
    # Retrieve dynamic data passed from the outbound RestClient.ai.call()
    metadata = kwargs.get("metadata", {})
    customer_name = metadata.get("customer_name", "there")
    
    print(f"📞 New Phone Agent Session: {call_id} from {from_number}")
    if customer_name != "there":
        print(f"   Context: Outbound call to {customer_name}")
    
    # 1. VoiceAssistant Configuration
    voice_agent = VoiceAgent(
        instructions=f\"\"\"
            You are a helpful outbound sales representative.
            Your name is 'Piopiy Agent'.
            You are speaking to {customer_name}.
            Keep your responses extremely short and conversational.
        \"\"\",
        greeting=f"Hello {customer_name}! This is the Piopiy Phone Agent calling. Do you have a quick minute?",
    )

    # 2. Speech-to-Text Setup
    stt = DeepgramSTTService(
        api_key=os.getenv("DEEPGRAM_API_KEY"),
        model="nova-2", 
        language="en-US"
    )

    # 3. LLM Setup
    llm = OpenAILLMService(
        api_key=os.getenv("OPENAI_API_KEY"),
        model="gpt-4o-mini",
    )

    # 4. Text-to-Speech Setup
    tts = CartesiaTTSService(
        api_key=os.getenv("CARTESIA_API_KEY"),
        voice_id="694f9ed8-38bb-4e94-91ce-f831ae3f3fc0", # Delivery Man Voice
    )

    # 5. Start the pipeline
    await voice_agent.Action(
        stt=stt,
        llm=llm,
        tts=tts,
        vad=True,
        allow_interruptions=True,
    )
    
    await voice_agent.start()

async def main():
    agent = Agent(
        agent_id=os.getenv("AGENT_ID"),
        agent_token=os.getenv("AGENT_TOKEN"),
        create_session=create_session,
    )
    
    print("🚀 Phone Agent starting...")
    print("   Waiting for outbound triggers...")
    
    await agent.connect()

if __name__ == "__main__":
    asyncio.run(main())

---

outbound_call.py

This script fires the underlying REST API mapping the user's phone number to your AI Agent ID.

import os
import dotenv
from piopiy_voice import RestClient

dotenv.load_dotenv()

def trigger_outbound_call():
    token = os.getenv("PIOPIY_TOKEN")
    agent_id = os.getenv("AGENT_ID")
    caller_id = os.getenv("PIOPIY_NUMBER")
    to_number = os.getenv("CUSTOMER_NUMBER")

    if not all([token, agent_id, caller_id, to_number]):
        print("❌ Error: Missing required environment variables.")
        return

    # Initialize the Piopiy REST Client
    client = RestClient(token=token)

    print(f"📞 Triggering outbound call to {to_number} from {caller_id}...")
    
    try:
        # Trigger an Outbound AI Agent Call
        response = client.ai.call(
            caller_id=caller_id,
            to_number=to_number,
            agent_id=agent_id,
            variables={
                "customer_name": "Alice", # This is passed to create_session as metadata
            }
        )
        print("✅ Call successfully initiated!")
        print(response)
    except Exception as e:
        print(f"❌ Failed to initiate call: {e}")

if __name__ == "__main__":
    trigger_outbound_call()