Call Transfer
Overview
Call transfer allows you to route an active call session to a different destination. This is essential for escalating support calls, routing to specialized departments, or connecting to external partners.
Prerequisites
Before running these examples, ensure you have completed the Telephony Quickstart to set up your environment and credentials.
1. Phone Agent Worker
Every transfer example requires a running agent worker to manage the active call stream.
phone_agent.py
import asyncio
import os
from dotenv import load_dotenv
from piopiy.agent import Agent
from piopiy.voice_agent import VoiceAgent
# ... (Standard imports from Quickstart)
load_dotenv()
async def create_session(call_id=None, **kwargs):
# This worker stays connected while you trigger the transfer externally
voice_agent = VoiceAgent(
instructions="You are a helpful assistant.",
greeting="Please wait while I connect your call."
)
# Configure your STT, LLM, TTS here...
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
)
await agent.connect()
if __name__ == "__main__":
asyncio.run(main())
2. Triggering the Transfer
Use the following scripts to transfer an active call. Each tab shows a different transfer destination.
- Mobile/PSTN
- SIP Endpoint
- AI Agent
Transfer to a standard mobile or landline number.
trigger_pstn_transfer.py
import os
from dotenv import load_dotenv
from piopiy_voice import RestClient
load_dotenv()
client = RestClient(token=os.environ["PIOPIY_TOKEN"])
pipeline = [
{
"action": "connect",
"params": { "from": "+15551234567" }, # Your caller ID
"endpoints": [
{ "type": "pstn", "number": "+15559876543" }
]
}
]
response = client.voice.transfer(call_id="active_call_id", pipeline=pipeline)
print(response)
Transfer to a specific SIP URI (VoIP systems, softphones).
trigger_sip_transfer.py
import os
from dotenv import load_dotenv
from piopiy_voice import RestClient
load_dotenv()
client = RestClient(token=os.environ["PIOPIY_TOKEN"])
pipeline = [
{
"action": "connect",
"params": { "from": "+15551234567" },
"endpoints": [
{ "type": "sip", "uri": "sip:[email protected]" }
]
}
]
response = client.voice.transfer(call_id="active_call_id", pipeline=pipeline)
print(response)
Transfer to another specialized AI agent.
trigger_agent_transfer.py
import os
from dotenv import load_dotenv
from piopiy_voice import RestClient
load_dotenv()
client = RestClient(token=os.environ["PIOPIY_TOKEN"])
pipeline = [
{
"action": "connect",
"params": { "from": "+15551234567" },
"endpoints": [
{ "type": "agent", "id": "target_agent_id" }
]
}
]
response = client.voice.transfer(call_id="active_call_id", pipeline=pipeline)
print(response)
Execution Flow
- Start the worker:
python phone_agent.py - Accept a call: Dial your Piopiy number.
- Trigger transfer: In a new terminal, run your chosen trigger script (e.g.,
python trigger_pstn_transfer.py).