Skip to main content

Transfer via Function Call

Overview​

Enable your AI agent to independently transfer a call to a human agent, another department, or a different specialized AI agent during a live conversation. This is achieved by defining a transfer_call tool that the agent can invoke based on user intent.

Prerequisites

Ensure you have completed the Telephony Quickstart and have your .env file configured with API keys for Deepgram, OpenAI, and Cartesia.

Complete Example​

This production-ready script demonstrates how to define a tool handler outside the session scope. We use a global variable to track the active_call_id so the handler can trigger the transfer correctly.

agent_with_transfer.py
import asyncio
import os
import functools
from dotenv import load_dotenv
from piopiy.agent import Agent
from piopiy.voice_agent import VoiceAgent
from piopiy.adapters.schemas.function_schema import FunctionSchema
from piopiy.services.deepgram.stt import DeepgramSTTService
from piopiy.services.openai.llm import OpenAILLMService
from piopiy.services.cartesia.tts import CartesiaTTSService
from piopiy_voice import RestClient

load_dotenv()

# Initialize RestClient
piopiy_rest = RestClient(token=os.getenv("PIOPIY_TOKEN"))

# Global variable to track call state
active_call_id = None

# 1. Define the Tool Schema
transfer_tool_schema = FunctionSchema(
    name="transfer_to_human",
    description="Transfer the call to a human support agent when requested.",
    parameters={
        "type": "object",
        "properties": {
            "reason": {"type": "string", "description": "The reason for the transfer"}
        }
    }
)

# 2. Define the Tool Handler (Outside create_session)
async def handle_transfer(reason):
    global active_call_id
    print(f"Transferring call {active_call_id} due to: {reason}")
    
    pipeline = [
        {
            "action": "connect",
            "params": { 
                "from": os.getenv("PIOPIY_NUMBER"), 
            },
            "endpoints": [
                { "type": "pstn", "number": os.getenv("SUPPORT_NUMBER") }
            ]
        }
    ]
    
    try:
        piopiy_rest.voice.transfer(call_id=active_call_id, pipeline=pipeline)
        return "Of course. I am connecting you to a human representative now. Please stay on the line."
    except Exception as e:
        print(f"Transfer trigger failed: {e}")
        return "I'm sorry, I'm unable to connect you to a person right now."

async def create_session(call_id=None, **kwargs):
    global active_call_id
    active_call_id = call_id
    print(f"New session: {call_id}")

    voice_agent = VoiceAgent(
        instructions="You are a helpful assistant. Transfer to a human if the user asks.",
        greeting="Hello! How can I help you today? Ask for a person if you need specialized help."
    )

    # Setup Services
    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"), model="nova-2")
    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o-mini")
    tts = CartesiaTTSService(api_key=os.getenv("CARTESIA_API_KEY"))

    # 3. Register the outside handler
    voice_agent.add_tool(transfer_tool_schema, handle_transfer)
    
    await voice_agent.Action(stt=stt, llm=llm, tts=tts, vad=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("Agent online. Ready to handle transfers.")
    await agent.connect()

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