Skip to main content

Function Calling

Overview​

Function calling lets your voice agent execute backend actions (CRM lookup, booking, weather, order status) instead of guessing answers from model text alone.

In Piopiy, you define tool schemas with FunctionSchema, attach async handlers, and the LLM calls them during a live conversation.

Use function calling when you need:

  • Reliable, factual responses from your own systems.
  • Mutating actions like booking, cancellation, ticket creation, or payment steps.
  • Structured outputs from business logic instead of free-form model text.

Where It Fits In Voice Flow​

During a call, Piopiy keeps the full realtime chain:

STT -> Context -> LLM (tool decision) -> Tool Handler -> LLM -> TTS

The LLM decides when to call a tool. Your handler returns structured data, then the LLM explains the result to the caller naturally.

Tool Lifecycle in Piopiy​

  1. Define a tool schema (name, description, properties, required).
  2. Register it with voice_agent.add_tool(schema, handler).
  3. In the handler, read args and return structured output.
  4. The tool result is injected into context and the LLM continues naturally.

Tool Schema Design​

Your schema is the contract the LLM sees before it calls your function.

from piopiy.adapters.schemas.function_schema import FunctionSchema

get_order_status_tool = FunctionSchema(
    name="get_order_status",
    description="Fetch order status by order ID.",
    properties={
        "order_id": {
            "type": "string",
            "description": "Customer order ID, e.g. ORD-10293",
        }
    },
    required=["order_id"],
)

Guidelines:

  • Use action-oriented names (get_order_status, create_ticket).
  • Write short, precise descriptions.
  • Keep inputs minimal and explicit.
  • Make required fields strict.

Handler Contract​

Tool handlers receive FunctionCallParams. In current SDK patterns:

  • Input args are available on params.args (use getattr(..., {}) safely).
  • Return data through await params.result_callback(result_dict).
from piopiy.services.llm_service import FunctionCallParams

async def get_order_status(params: FunctionCallParams):
    args = getattr(params, "args", {}) or {}
    order_id = args.get("order_id")
    if not order_id:
        await params.result_callback({"error": "missing_order_id"})
        return

    # Replace with real backend lookup
    await params.result_callback(
        {
            "order_id": order_id,
            "status": "shipped",
            "eta_date": "2026-03-05",
        }
    )

End-to-End Voice Agent Example​

import os
from piopiy.voice_agent import VoiceAgent
from piopiy.adapters.schemas.function_schema import FunctionSchema
from piopiy.services.llm_service import FunctionCallParams

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


async def get_order_status(params: FunctionCallParams):
    args = getattr(params, "args", {}) or {}
    order_id = args.get("order_id")
    if not order_id:
        await params.result_callback({"error": "missing_order_id"})
        return

    # Replace with your DB/API call
    await params.result_callback(
        {"order_id": order_id, "status": "shipped", "eta_date": "2026-03-05"}
    )


async def create_support_ticket(params: FunctionCallParams):
    args = getattr(params, "args", {}) or {}
    issue = args.get("issue", "").strip()
    if not issue:
        await params.result_callback({"error": "missing_issue"})
        return

    # Replace with CRM/helpdesk API call
    await params.result_callback({"ticket_id": "TCK-8391", "status": "created"})


get_order_status_tool = FunctionSchema(
    name="get_order_status",
    description="Fetch order status by order ID.",
    properties={
        "order_id": {
            "type": "string",
            "description": "Customer order ID, e.g. ORD-10293",
        }
    },
    required=["order_id"],
)

create_support_ticket_tool = FunctionSchema(
    name="create_support_ticket",
    description="Create a support ticket for unresolved customer issues.",
    properties={
        "issue": {
            "type": "string",
            "description": "Short issue summary from the caller.",
        }
    },
    required=["issue"],
)


async def on_new_session(agent_id, call_id, from_number, to_number, metadata=None):
    voice_agent = VoiceAgent(
        instructions=(
            "You are a concise support voice assistant. "
            "Use get_order_status for order queries. "
            "Use create_support_ticket only after collecting enough details "
            "and confirming intent with the caller."
        ),
        greeting="Hello! How can I help you today?",
    )

    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_agent.add_tool(get_order_status_tool, get_order_status)
    voice_agent.add_tool(create_support_ticket_tool, create_support_ticket)

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

Timeout and Error Pattern​

Always bound external calls. Return structured failures the LLM can explain.

import asyncio

async def get_customer_profile(params: FunctionCallParams):
    args = getattr(params, "args", {}) or {}
    customer_id = args.get("customer_id")
    if not customer_id:
        await params.result_callback({"error": "missing_customer_id"})
        return

    try:
        profile = await asyncio.wait_for(fetch_profile(customer_id), timeout=3.0)
        await params.result_callback({"customer_id": customer_id, "profile": profile})
    except asyncio.TimeoutError:
        await params.result_callback({"error": "profile_lookup_timeout"})
    except Exception:
        await params.result_callback({"error": "profile_lookup_failed"})

Best Practices​

  • Keep tool outputs small and JSON-serializable.
  • Validate arguments in handlers and apply safe defaults.
  • Keep read actions and write actions clearly separated.
  • For destructive/mutating actions, instruct the model to confirm first.
  • Keep handlers idempotent where possible.
  • Add timeout/retry logic for external APIs.
  • Write explicit instructions telling the LLM when to call tools.

Common Mistakes​

  • Tool description too vague, causing wrong tool selection.
  • Returning huge payloads that bloat context.
  • Missing validation for required fields.
  • Mutating backend state without caller confirmation.
  • Performing long blocking work in handlers.

Production Checklist​

  • Every tool has a clear purpose and schema.
  • Handlers validate args and handle failures safely.
  • External calls have timeout and retry policies.
  • Observability is in place (latency, error rate, tool call count).
  • Sensitive operations require explicit confirmation in instructions.

References​

What's Next​