Skip to main content

Hangup via Function Call

Overview​

Allow your AI agent to gracefully end a call session once a task is completed (e.g., after booking a meeting) or if the user requests to terminate. This is implemented by providing the agent with a hangup_call tool.

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 agent can terminate the session programmatically when requested.

agent_with_hangup.py
import asyncio
import os
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
hangup_tool_schema = FunctionSchema(
    name="end_call",
    description="End the current phone call session when the conversation is finished.",
    parameters={"type": "object", "properties": {}}
)

# 2. Define the Tool Handler (Outside create_session)
async def handle_hangup():
    global active_call_id
    print(f"Ending call session: {active_call_id}")
    
    try:
        piopiy_rest.voice.hangup(call_id=active_call_id)
        return "I've helped with everything you need. Thank you for calling Piopiy. Goodbye!"
    except Exception as e:
        print(f"Hangup trigger failed: {e}")
        return "Goodbye! (Session ending)"

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. Once you have helped the user, say goodbye and use end_call to finish.",
        greeting="Hi there! How can I assist you today?"
    )

    # 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(hangup_tool_schema, handle_hangup)
    
    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 calls.")
    await agent.connect()

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