Cartesia Line Agents and Voximplant demo - connect to the Cartesia agent and manage call transfer in Voximplant

cartesia-demo-voxengine-scenario.js

// Voximplant VoxEngine scenario:
// - Streams caller audio <-> Cartesia Line agent (Agents connector)
// - Supports:
//   - end_call: hang up the caller leg
//   - call_transfer: place an outbound PSTN consult call and then bridge caller -> consult leg
//
// Configure these keys in Voximplant Application Storage:
// - CARTESIA_API_KEY
// - CARTESIA_AGENT_ID
// - PSTN_CALLER_ID (required for callPSTN; must be a real E.164 number in your Voximplant account)
//
// Call transfer in this demo always dials the public Voximplant demo number shown on the Cartesia Agents product page.

require(Modules.Cartesia);
require(Modules.ApplicationStorage);
require(Modules.ASR);

const CARTESIA_VERSION = "2025-04-16";
const CALL_TRANSFER_NUMBER = "+18339906144";

// Per-session control URL. Any HTTPS request to this URL triggers AppEvents.HttpRequest in this session.
// We'll pass it into Cartesia call metadata so the agent runtime can request telephony actions via HTTP.
let sessionControlUrl = null;

// Current call session state (single-call demo scenario).
let callerLeg = null;
let voiceAIClient = null;
let consultLeg = null;
let transferInProgress = false;
let transferred = false;

VoxEngine.addEventListener(AppEvents.Started, (appEvent) => {
  sessionControlUrl = appEvent.accessSecureURL;
  Logger.write(`===SESSION_CONTROL_URL_READY===>${JSON.stringify({accessSecureURL: sessionControlUrl}) || ""}`);
});

VoxEngine.addEventListener(AppEvents.HttpRequest, (appEvent) => {
  Logger.write(`===HTTP_CONTROL_REQUEST===>${JSON.stringify({method: appEvent.method, path: appEvent.path}) || ""}`);

  // Check for and handle control commands
  const cmd = JSON.parse(appEvent?.content);
  if (cmd.action === "end_call") {
    Logger.write(`===CONTROL_END_CALL===>${JSON.stringify(cmd) || ""}`);
    callerLeg.hangup();
    voiceAIClient?.close();
    VoxEngine.terminate();
  } else if (cmd.action === "call_transfer") {
    Logger.write(`===CONTROL_CALL_TRANSFER===>${JSON.stringify(cmd) || ""}`);
    callTransfer(cmd);
  } else {
    Logger.write(`===CONTROL_COMMAND_UNKNOWN===>${JSON.stringify(cmd) || ""}`);
  }
  return JSON.stringify({ok: true});
});

async function callTransfer(cmd) {
  if (transferInProgress || transferred) return;
  if (!callerLeg) return;

  transferInProgress = true;
  Logger.write(`===CALL_TRANSFER_REQUESTED===>${JSON.stringify(cmd || {}) || ""}`);

  // Detach the agent now for a blind transfer. Delay or conference for a warm transfer
  if (voiceAIClient) {
    VoxEngine.stopMediaBetween(callerLeg, voiceAIClient);
    voiceAIClient.close();
    voiceAIClient = null;
  }

  const currentPstnCallerId = (await ApplicationStorage.get("PSTN_CALLER_ID")).value;
  consultLeg = VoxEngine.callPSTN(CALL_TRANSFER_NUMBER, currentPstnCallerId, {followDiversion: true});

  consultLeg.addEventListener(CallEvents.Failed, () => {
    transferInProgress = false;
    Logger.write(`===CONSULT_CALL_FAILED===>${JSON.stringify({}) || ""}`);
    callerLeg.hangup();
  });

  consultLeg.addEventListener(CallEvents.Disconnected, (event) => {
    Logger.write(`===CONSULT_CALL_DISCONNECTED===>${JSON.stringify(event) || ""}`);
  });

  consultLeg.addEventListener(CallEvents.Connected, () => {
    Logger.write(`===CONSULT_CALL_CONNECTED===>${JSON.stringify({}) || ""}`);
    transferInProgress = false;
    transferred = true;
    VoxEngine.sendMediaBetween(callerLeg, consultLeg);
  });
}

function onWebSocketClose(event) {
  Logger.write(`===ON_WEB_SOCKET_CLOSE===>${JSON.stringify(event) || ""}`);
  // Ignore expected close during transfer
  if (transferInProgress || transferred || event.code === 1000) return;
  // otherwise end the call
  callerLeg.hangup();
  VoxEngine.terminate();
}

VoxEngine.addEventListener(AppEvents.CallAlerting, async ({ call }) => {
  callerLeg = call;

  // Termination functions - add cleanup and logging as needed
  call.addEventListener(CallEvents.Disconnected, () => VoxEngine.terminate());
  call.addEventListener(CallEvents.Failed, () => VoxEngine.terminate());

  try {
    call.answer();
    call.record({ hd_audio: true, stereo: true });   // Optional: record the call

    voiceAIClient = await Cartesia.createAgentsClient({
      apiKey: (await ApplicationStorage.get("CARTESIA_API_KEY")).value,
      agentId: (await ApplicationStorage.get("CARTESIA_AGENT_ID")).value,
      cartesiaVersion: CARTESIA_VERSION,
      onWebSocketClose,
    });

    VoxEngine.sendMediaBetween(call, voiceAIClient);

    voiceAIClient.start({
      // Optional metadata passed into the Cartesia agent
      metadata: {
        from: call.callerid(),
        to: call.number(),
        vox_session_control_url: sessionControlUrl, // Control plane from Cartesia via HTTPS callback
      },
    });

    // "log only" handlers for debugging.
    [
      Cartesia.AgentsEvents.ACK,
      Cartesia.AgentsEvents.Clear,
      Cartesia.AgentsEvents.ConnectorInformation,
      Cartesia.AgentsEvents.DTMF,
      Cartesia.AgentsEvents.Unknown,
      Cartesia.AgentsEvents.WebSocketError,
    ].forEach((eventName) => {
      voiceAIClient.addEventListener(eventName, (event) => {
        Logger.write(`===${event.name}===>${JSON.stringify(event.data) || ""}`);
      });
    });
  } catch (error) {
    Logger.write(`===SOMETHING_WENT_WRONG===>${JSON.stringify(error) || String(error)}`);
    VoxEngine.terminate();
  }
});

cartesia_demo_main.py

import os
import asyncio
from typing import Annotated, Optional

import httpx

from line.llm_agent import LlmAgent
from line.llm_agent.config import LlmConfig
from line.events import AgentEndCall, AgentSendText
from line.llm_agent.tools.decorators import passthrough_tool
from line.llm_agent.tools.utils import ToolEnv
from line.voice_agent_app import AgentEnv, CallRequest, VoiceAgentApp


DEFAULT_SYSTEM_PROMPT = """\
You are a helpful voice agent running on Cartesia Line, connected to a phone call via Voximplant.

This demo has a very simple call transfer:
1) If the caller asks for Voxy or a human, call call_transfer(summary=...).
2) The call_transfer tool will speak the transfer confirmation to the caller.
3) Do not continue the conversation after calling call_transfer.

Ending the call:
- If the caller asks to hang up or says goodbye, say a short goodbye and then call end_call().

Be concise, polite, and ask one question at a time.
"""

DEFAULT_INTRODUCTION = "Hi, this is Voximplant Voice AI powered by Cartesia Line. How can I help?"

async def _post_vox_control(
    url: str,
    payload: dict,
    *,
    timeout_s: float = 3.0,
) -> dict:
    """Send a POST request to the given URL with the provided payload."""
    try:
        async with httpx.AsyncClient(timeout=timeout_s) as client:
            response = await client.post(url, json=payload)
            try:
                return response.json()
            except Exception:
                return {"status_code": response.status_code, "text": response.text}
    except Exception:
        return {"error": "request_failed"}


async def get_agent(env: AgentEnv, request: CallRequest):
    # Extract the Voximplant control URL from the request metadata, if present and valid.
    vox_control_url: Optional[str] = None
    if request.metadata and isinstance(request.metadata, dict):
        raw = request.metadata.get("vox_session_control_url")
        if isinstance(raw, str) and raw.startswith("https://"):
            vox_control_url = raw

    if not vox_control_url:
        return {"error": "missing_vox_control_url", "message": "The call request is missing a valid 'vox_session_control_url' in its metadata."}

    @passthrough_tool
    async def call_transfer(
        ctx: ToolEnv,
        summary: Annotated[
            str,
            "Optional short transfer summary (for logs / analytics). Voximplant will receive it.",
        ] = "",
    ):
        """Request Voximplant to transfer the call to a human (Voximplant performs the telephony actions).
           Speak first so the caller reliably hears the transfer confirmation before Voximplant
           detaches the agent audio and bridges to PSTN.
        """
        yield AgentSendText(text="Sure. One moment, I'm transferring you to a human now.")

        async def _do_transfer():
            # Give TTS time to play before Voximplant starts tearing down the agent bridge.
            # use line.events.AgentTurnEnded to be more precise
            await asyncio.sleep(5.0)
            await _post_vox_control(
                vox_control_url,
                {"action": "call_transfer", "summary": summary},
                timeout_s=10.0,
            )

        asyncio.create_task(_do_transfer())

    @passthrough_tool
    async def end_call(ctx: ToolEnv):
        """End the call."""
        yield AgentEndCall()
        # Give TTS time to play before Voximplant starts tearing down the agent bridge.
        # use line.events.AgentTurnEnded to be more precise
        await asyncio.sleep(3.0)
        await _post_vox_control(vox_control_url, {"action": "end_call"})

    agent = LlmAgent(
        model="gpt-5-nano",
        api_key=os.getenv("OPENAI_API_KEY", ""),
        tools=[end_call, call_transfer],
        config=LlmConfig(
            system_prompt=DEFAULT_SYSTEM_PROMPT,  # use request.agent.system_prompt for the GUI version
            introduction=DEFAULT_INTRODUCTION,    # use request.agent.introduction for the GUI version
        ),
    )

    return agent


voice_agent_app = VoiceAgentApp(get_agent=get_agent)
app = voice_agent_app.fastapi_app


if __name__ == "__main__":
    # Local dev only. Cartesia runs this as a web service in the cloud.
    voice_agent_app.run()