Function Calling Example

View Source on GitHub

This example demonstrates how to integrate external APIs into your Piopiy Voice Agent by defining custom Python functions as "tools".

The LLM (OpenAI) is instructed to use a get_current_weather tool when the caller asks about the weather. When triggered, the LLM stops speaking, your local Python function executes, and the JSON result is fed back to the LLM to generate a natural voice response.

Requirements

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

Ensure your .env contains:

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"

How to Run

Save the script below as weather.py and run it:

python weather.py

1. Log in to the Piopiy Dashboard.
2. Ensure you have purchased a Piopiy phone number and mapped it to your new AI Agent. (See Dashboard Setup Guide for help).
3. Dial that phone number from your personal phone to interact with your local agent!

When the agent picks up, ask it: "What is the weather like in Chennai right now?"

Full Script

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.audio.vad.silero import SileroVADAnalyzer
from piopiy.audio.interruptions.min_words_interruption_strategy import MinWordsInterruptionStrategy

from piopiy.services.deepgram.stt import DeepgramSTTService
from piopiy.services.openai.llm import OpenAILLMService
from piopiy.services.cartesia.tts import CartesiaTTSService
from piopiy.services.llm_service import FunctionCallParams

load_dotenv()


# ---- 1. Tool handler function ----
async def get_current_weather(params: FunctionCallParams):
    args = getattr(params, "args", {}) or {}
    location = args.get("location", "Chennai")
    
    # Normally you would replace this with a real HTTP request to a Weather API
    print(f"🌦️ Tool Executing: Looking up weather for {location}")
    response_data = {"location": location, "temperature_c": 30}
    
    await params.result_callback(response_data)


# ---- 2. Tool schema definition ----
weather_function = FunctionSchema(
    name="get_current_weather",
    description="Get current weather for a location.",
    properties={
        "location": {
            "type": "string",
            "description": "City name to look up weather for.",
        }
    },
    required=["location"],
)


# ---- 3. Session factory ----
async def create_session(agent_id, call_id, from_number, to_number, metadata=None):
    voice_agent = VoiceAgent(
        instructions="You are a helpful weather assistant. Use your tools to check the weather when asked.",
        greeting="Hello! Try asking me about the weather in Chicago or London.",
    )

    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
    tts = CartesiaTTSService(
        api_key=os.getenv("CARTESIA_API_KEY"),
        voice_id="bdab08ad-4137-4548-b9db-6142854c7525", # British Lady
    )

    # Register the tool!
    voice_agent.add_tool(weather_function, get_current_weather)

    await voice_agent.Action(
        stt=stt,
        llm=llm,
        tts=tts,
        vad=True,
        allow_interruptions=True,
    )

    await voice_agent.start()


# ---- 4. Entrypoint ----
async def main():
    agent = Agent(
        agent_id=os.getenv("AGENT_ID"),
        agent_token=os.getenv("AGENT_TOKEN"),
        create_session=create_session,
    )
    
    print("🚀 Weather Agent starting...")
    await agent.connect()


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