Skip to main content

Failover AI Agent

Overview​

Failover routing is a critical feature for production voice applications. It ensures business continuity by automatically redirecting an outbound call to a backup AI agent if the primary agent fails to answer or encounters an error.

When to use Failover​

  • Agent Unavailability: The primary agent worker is offline or overloaded.
  • Specific Network Triggers: The primary call attempt receives specific telephony response codes (e.g., busy, no answer).
  • Graceful Degradation: Switch to a simpler, more robust backup agent during maintenance of the primary agent.

Prerequisites​

  • Primary and backup AI agents created in dashboard
  • Outbound-enabled caller ID/number purchased in dashboard
  • Python 3.10 or later
  • PIOPIY_TOKEN configured in .env

Agent Workers​

primary_phone_agent.py
import asyncio
import os
from dotenv import load_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

load_dotenv()

PRIMARY_AGENT_ID = "primary_agent_id"
PRIMARY_AGENT_TOKEN = "primary_agent_token"


async def create_session(call_id=None, from_number=None, to_number=None, **kwargs):
    voice_agent = VoiceAgent(
        instructions="You are the primary outbound assistant.",
        greeting="Hello, this is our primary outbound assistant.",
    )
    stt = DeepgramSTTService(api_key=os.environ["DEEPGRAM_API_KEY"], model="nova-2")
    llm = OpenAILLMService(api_key=os.environ["OPENAI_API_KEY"], model="gpt-4o-mini")
    tts = CartesiaTTSService(api_key=os.environ["CARTESIA_API_KEY"])
    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=PRIMARY_AGENT_ID,
        agent_token=PRIMARY_AGENT_TOKEN,
        create_session=create_session,
    )
    print("Primary agent online...")
    await agent.connect()


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

Trigger Outbound Call With Failover​

trigger_failover_outbound_call.py
import os
from dotenv import load_dotenv
from piopiy_voice import RestClient

load_dotenv()

client = RestClient(token=os.environ["PIOPIY_TOKEN"])

primary_agent_id = "primary_agent_id"
backup_agent_id = "backup_agent_id"
caller_id = "+15551234567"
to_number = "+15557654321"

response = client.ai.call(
    caller_id=caller_id,
    to_number=to_number,
    agent_id=primary_agent_id,
    failover={
        "trigger": "no_answer",
        "target": {"agent_id": backup_agent_id},
    },
)

print(response)

Run​

Terminal
python primary_phone_agent.py
python backup_phone_agent.py
python trigger_failover_outbound_call.py

Reference​