Building a Compliant BFSI Voice Agent

Quick links.
Code: https://github.com/murf-ai/murf-cookbook/tree/main/examples/agents/payment-reminder
Video Walkthrough:

Financial services is one of the most heavily regulated spaces to put a voice AI agent into. Whether it's a payment reminder, a KYC verification call or a fraud alert, the conversation is bound by rules.

The common thread across all of them is what your agent says, when it says it, and to whom — and these are no longer just a UX choice. Reveal an account detail before verifying who's on the line, apply a note of pressure to someone who just disclosed a hardship, say the wrong thing at the wrong moment, and you've crossed a line that carries real consequences.

Most people reach for a better prompt to fix this, which works most of the time. But a phone call is a live, irreversible conversation, and "most of the time" isn't good enough when it comes to compliance.

So the question isn't "how do I write a better prompt?" — it's "how do I build a voice agent that stays compliant even when the model ignores the prompt?"

In this tutorial, we'll build an outbound payment-reminder agent that calls a customer, verifies identity and offers a payment link, while handling disputes, hardship and wrong numbers the way the rules demand.

---

Architecture

The basic voice architecture for most agents remains the same: wire up the telephony, connect the LLM and write a system prompt telling the agent to "be a polite debt collector".

In financial services, a single prompt left to its own devices is a compliance violation waiting to happen. LLMs are probabilistic — they want to please the user, answer questions, and keep the conversation going. But in a highly regulated space, you need the agent to be deterministic.

To achieve this, we aren't just writing a prompt; we are building a 3-Layer Architecture.

Layer 1: The Foundation (Basic Outbound Call Agent + System Prompt)

This is our baseline. In this layer, we wire up Twilio, LiveKit, Murf, and the LLM, and give the agent its identity, its voice, and its core context. This layer handles the mechanics of listening, thinking, and speaking — but we can't trust it to govern the flow of the call.

Layer 2: The Enforcer (The State Machine)

This is the secret sauce of a compliant BFSI agent. Instead of hoping the LLM remembers to verify a user's identity before talking about debt, we hardcode a state machine. The conversation is broken into strict phases:

• Greeting
• Verification
• Payment_Discussion
• Outcome

The agent is physically locked out of the context required to discuss a payment until the Verification state is cleared. If the user tries to skip ahead, the state machine yanks the LLM back to the current required task, turning an unpredictable AI into a strict, compliant workflow.

Layer 3: The Safety Net (Guardrails & Human Escalation)

Even with a state machine, the real world is messy. Callers get angry, they mention bankruptcy, or they threaten legal action. Layer 3 acts as our emergency brake. We implement semantic guardrails that constantly monitor the user's intent. If a trigger word or high-stress emotion is detected, the agent immediately stops generating responses and triggers an escalation protocol, gracefully terminating the call.

---

Setup & Requirements

Twilio

Twilio is a cloud communications platform that turns the global telecom network into simple software APIs, acting as a digital bridge between the internet and phone networks.

Setup:

1. Create an account at console.twilio.com or 1console.twilio.com. Copy your account SID, auth token and phone number, and paste them into the .env file.
2. On the console, go to TwiML Bins and create a bin.
3. Give it a name and put this in the TwiML section:

<?xml version="1.0" encoding="UTF-8"?>
<Response>
  <Dial>
    <Sip>sip:<phone_number>@<sip_uri>;transport=tcp</Sip>
  </Dial>
</Response>

Replace <phone_number> with your Twilio phone number and <sip_uri> with your SIP URI. For example, if your phone number is +1234567 and your SIP URI is sip:abc123.sip.livekit.cloud, your TwiML section would look like this:

<?xml version="1.0" encoding="UTF-8"?>
<Response>
  <Dial>
    <Sip>sip:[e-mail];transport=tcp</Sip>
  </Dial>
</Response>

1. On 1console.twilio.com, go to Products & Services → Numbers & Senders, click your phone number, and in the Voice and emergency calling section click Edit configuration details.
2. Select your configuration method as the one with TwiML Bins in it, set your primary method as TwiML Bins, and select the bin you just created.
3. Go to Products & Services → Elastic SIP Trunking → Trunks and create a new trunk.
4. In the Termination tab of your trunk, set the Termination SIP URI and paste it into your .env file.
5. Scroll down to Credential Lists and create a new one. Copy the username and password into your .env file as well.

LiveKit

LiveKit is streaming infrastructure built on WebRTC that manages audio buffering and handles network drops gracefully.

Setup:

1. Create an account at cloud.livekit.io and create a project.
2. From your project settings, copy the URL, API key, API secret and SIP URI into your .env file.
3. On the console, go to Telephony → SIP Trunks and create a new trunk.
4. Give the trunk a name, select outbound, and put the Termination SIP URI of your Twilio trunk under addresses.
5. Add your Twilio phone number to Numbers, and under Optional Settings enter the username and password from the credential list you created in the Twilio trunk.

If you switch to the JSON editor, your JSON should look something like this:

{
  "name": "payment reminder",
  "address": "payment-reminder.pstn.twilio.com",
  "transport": "SIP_TRANSPORT_TCP",
  "numbers": [
    "[phone]"
  ],
  "authUsername": "payment-reminder",
  "authPassword": "********"
}

1. Go to Telephony → Dispatch Rules and create a new dispatch rule.
2. Give your rule a name, a prefix and an agent name. Remember your agent name — it must match the one in your code, or the call will connect but all you'll hear is silence.

If you switch to the JSON editor, your dispatch rule should look something like this:

{
  "sipDispatchRuleId": "SDR_abc123",
  "rule": {
    "dispatchRuleIndividual": {
      "roomPrefix": "payment-"
    }
  },
  "trunkIds": [
    "ST_abc123"
  ],
  "name": "payment",
  "roomConfig": {
    "agents": [
      {
        "agentName": "payment-reminder"
      }
    ]
  }
}

Copy this and save it as dispatch-rule.json in your project root directory.

Speech-To-Text (STT) — Deepgram Nova-3

We use Deepgram's Nova-3 as the STT for this project. Deepgram offers $200 in free credits when your account is created, which is more than enough to test and run this project. You can get your API key at console.deepgram.com.

LLM

You will require a Large Language Model to act as the central reasoning engine for your voice agent. This project is configured to support two major providers. To choose yours, set the LLM_PROVIDER variable in your .env file to one of:

• openai (requires a paid API key)
• gemini (an excellent choice to test with — Google offers a free tier)

If you wish to use a different provider, the code is structured so you can wire in any custom LLM.

Text-To-Speech (TTS) — Murf Falcon

Once our LLM decides exactly what to say, we need to convert that text back into natural human speech to stream down the phone line.

For this project, we are using Murf Falcon, the consistently fastest TTS model (130ms time to first audio) built specifically for real-time conversational applications. Falcon is optimized to generate lifelike, natural-sounding human speech on the fly.

It achieves this by supporting continuous chunked audio streaming — the moment our LLM streams its first few words, Falcon immediately starts converting those tokens into audio packets and pushing them back into the LiveKit media stream. This ensures the caller hears a seamless, expressive response without awkward pauses.

Beyond just speed, Falcon also addresses a critical banking compliance requirement by offering localized data residency in India, ensuring that sensitive customer data never leaves the country.

Go to murf.ai/api, create an account and get your API key.

Project setup

Create your project directory and set up a standard Python virtual environment to keep your dependencies isolated:

mkdir payment-reminder-agent
cd payment-reminder-agent
python -m venv venv

# Activate it (Mac/Linux)
source venv/bin/activate

# Or on Windows(Powershell):
# venv\Scripts\Activate.ps1

Create a requirements.txt file in your project root:

livekit-agents>=1.0.0
livekit-plugins-deepgram>=0.7.0
livekit-plugins-openai>=0.10.0
livekit-plugins-google>=0.6.0
livekit-plugins-silero>=0.7.0
livekit-murf>=0.1.0
livekit-api>=0.7.0
python-dotenv>=1.0.0
twilio>=9.0.0
psutil>=5.9.0

Then run:

pip install -r requirements.txt

Customer data

We'll set up two ways to store customer information.

For a single customer, create scenario_config.json in your project root:

{
  "companyName": "NovaFin",
  "agentName": "Asha",
  "agentVoice": "en-IN-anisha",
  "useCase": "payment_reminder",
  "language": "en",
  "customerName": "Aria",
  "accountEnding": "4321",
  "registeredMobileLastFour": "1234",
  "amountDue": "10000",
  "amountDueFormatted": "₹10,000",
  "dueDate": "June 21, 2026",
  "daysPastDue": 4,
  "scenario": "normal_reminder",
  "requireIdentityVerification": true,
  "recordingDisclosureRequired": true,
  "paymentLinkEnabled": true,
  "humanHandoffEnabled": true
}

For multiple customers, create a CSV file in your project root:

name,phone,amount_due,due_date,account_ending,registered_mobile_last_four
Aria Sharma,+910000000001,10000,"June 21, 2026",4321,1234
Rahul Verma,+910000000002,5000,"July 1, 2026",8765,5678
Priya Patel,+910000000003,15000,"June 28, 2026",2468,1357

Now that we have everything in place, let's dive into building this agent.

---

Layer 1: The Foundation

In this layer, we wire up the core telephony, give our agent a strict system prompt, and define the tools it can use. We'll split this across a few dedicated files to keep our logic completely decoupled from our telephony infrastructure.

Managing state and data (data.py)

In a real-world BFSI setup, you need to track exactly what happens on a call for auditing purposes. Did the user verify their identity? Did they promise to pay? Was a dispute raised?

Instead of parsing a massive text transcript after the fact, we use structured data logging. Create a data.py file. This file handles our environment variables, loads our config, and defines an OutcomeLog dataclass to strictly track the state of the call.

import contextvars
import json
import logging
import os
import re
import time
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone


# Context variables allow us to handle multiple simultaneous calls
# without their customer details bleeding into each other.
_call_ctx: contextvars.ContextVar[dict | None] = contextvars.ContextVar("call_cfg", default=None)


def get_config() -> dict:
    """Read the customer/scenario config fresh from disk."""
    with open(_CONFIG_PATH, encoding="utf-8") as f:
        return json.load(f)


def set_call_context_config(cfg: dict) -> None:
    """Bind this call's config to the current context so parallel calls stay isolated."""
    _call_ctx.set(cfg)


def _effective_config() -> dict:
    """Fetches the specific caller's config for the current active call."""
    ctx_cfg = _call_ctx.get()
    return ctx_cfg if ctx_cfg is not None else get_config()


def is_identity_match(mobile_last_four: str, account_last_four: str) -> bool:
    """Identity is confirmed only when both the mobile and account last four match."""
    return (
        re.sub(r"\D", "", mobile_last_four) == str(_effective_config()["registeredMobileLastFour"]) and
        re.sub(r"\D", "", account_last_four) == str(_effective_config()["accountEnding"])
    )


@dataclass
class OutcomeLog:
    """Tracks the definitive outcome of the call for compliance and CRM syncing."""
    scenario: str = ""
    call_started: bool = False
    recording_disclosure_played: bool = False
    identity_verified: bool = False
    amount_disclosed: bool = False
    payment_link_sent: bool = False
    promise_to_pay_date: str | None = None
    dispute_detected: bool = False
    hardship_detected: bool = False
    outcome: str = "unknown"

    def save_to_file(self) -> str:
        os.makedirs("logs", exist_ok=True)
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = f"logs/{self.scenario}_{timestamp}.json"
        with open(filename, "w", encoding="utf-8") as f:
            json.dump(asdict(self), f, indent=2)
        return filename

Giving the agent some tools (tools.py)

LLMs are great at talking, but they can't natively do anything. To bridge the gap between conversation and action, we use function calling.

When the user says, "Yes, my account ends in 4321 and my phone ends in 1234," we don't want the LLM to just guess if that's correct. We want it to call a Python function to check against our data.

Create tools.py and define the actions the agent can take:

import logging
from typing import Annotated
from livekit.agents import RunContext, function_tool, get_job_context
from pydantic import Field
import data


logger = logging.getLogger(__name__)


def _get_outcome_log() -> data.OutcomeLog | None:
    job_ctx = get_job_context(required=False)
    return job_ctx.proc.userdata.get("outcome_log") if job_ctx else None


@function_tool()
async def verify_borrower_identity(
    mobile_last_four: Annotated[str, Field(description="Last four digits of registered mobile")],
    account_last_four: Annotated[str, Field(description="Last four digits of account number")],
    context: RunContext,
) -> str:
    """Verify the borrower's identity before disclosing financial details."""
    outcome_log = _get_outcome_log()

    if data.is_identity_match(mobile_last_four, account_last_four):
        if outcome_log:
            outcome_log.identity_verified = True
        return "Thank you, your identity has been confirmed."

    return (
        "I'm sorry, those details don't match what we have on file. "
        "Could you please give me the last four digits of your registered mobile number "
        "and your account number again?"
    )


@function_tool()
async def flag_hardship(
    reason: Annotated[str, Field(description="Brief reason the borrower gave for their hardship.")],
    context: RunContext,
) -> str:
    """Flag the account for hardship and arrange a human callback."""
    outcome_log = _get_outcome_log()
    if outcome_log:
        outcome_log.hardship_detected = True
        outcome_log.outcome = "hardship_detected"

    return (
        "I understand, and I'm truly sorry to hear that. "
        "I've flagged your account and a member of our team will call you back to discuss your options."
    )


@function_tool()
async def send_payment_link(
    context: RunContext,
) -> str:
    """Send the official payment link to the borrower's registered mobile number."""
    outcome_log = _get_outcome_log()
    # Compliance backstop: never send the link before identity is verified.
    if outcome_log and not outcome_log.identity_verified:
        return "I can't send the payment link until your identity is verified."
    if outcome_log:
        outcome_log.payment_link_sent = True
    return (
        "I've just sent the payment link to your registered mobile number. "
        "Please check your messages."
    )

The brain (prompt.py)

Now we define the system prompt. In Layer 1, this is our primary defense mechanism. Notice how explicit and strict the instructions are. We don't just tell it what to do; we explicitly command it on what it must not do (for example: "Do not reveal the customer's full name, account number, amount due or overdue status until their identity has been verified").

def build_payment_prompt(config: dict) -> str:
    return (
        f"You are {config['agentName']}, an automated payment assistance voice agent from {config['companyName']}.\n\n"
        "CORE RULES — NEVER BREAK THESE\n"
        "- Do not reveal the customer's full name, account number, amount due, or overdue status\n"
        "  to anyone until identity has been verified.\n"
        "- If the borrower mentions job loss, inability to pay, or medical emergency,\n"
        "  call flag_hardship immediately. Do not pressure them.\n"
        "- Never make threats, use intimidating language, or shame the borrower.\n\n"
        "Follow this exact order:\n"
        f"1. Introduce yourself. State the call may be recorded. Ask if you are speaking with {config['customerName']}.\n"
        "2. If confirmed, ask for the last four digits of their mobile and account number.\n"
        "3. Call verify_borrower_identity.\n"
        f"4. Only after verification: mention the amount due ({config['amountDueFormatted']}) and due date ({config['dueDate']}).\n"
        "5. Offer to send the official payment link.\n"
        "6. Close the call politely.\n\n"
        "LANGUAGE AND TONE\n"
        "Keep every response to 1-2 sentences. Speak clearly and calmly. Do not use filler "
        'phrases like "Certainly!" or "Of course!".'
    )

Wiring it all together (agent.py)

This is where it all comes together. We spin up a LiveKit worker that manages the real-time media streams. It handles voice activity detection (VAD), pipes audio to our STT, passes text to the LLM, and streams Murf Falcon's audio back into the phone call.

One crucial detail: we must wait for the media track. When SIP calls connect, the network often takes half a second to start flowing RTP packets, so if your agent speaks the instant the line connects, the first two words of your greeting will be chopped off. We handle this by specifically waiting for the audio track to subscribe before speaking.

import asyncio
import os
import time
from livekit import rtc
from livekit.agents import Agent, AgentSession, JobContext, JobProcess, WorkerOptions, cli
from livekit.plugins import deepgram, google, openai, silero, murf
import data
from prompt import build_payment_prompt
from tools import verify_borrower_identity, flag_hardship, send_payment_link


def prewarm(proc: JobProcess) -> None:
    """Load the VAD once per worker process, before any call comes in."""
    proc.userdata["vad"] = silero.VAD.load()


async def entrypoint(ctx: JobContext) -> None:
    # 1. Connect to the LiveKit Room
    await ctx.connect(auto_subscribe=rtc.AutoSubscribe.AUDIO_ONLY)

    # 2. Setup call context and logging
    call_cfg = data.get_config()
    data.set_call_context_config(call_cfg)
    ctx.proc.userdata["outcome_log"] = data.OutcomeLog(scenario=call_cfg["scenario"])

    # 3. Initialize the Pipeline Models
    tts_instance = murf.TTS(voice=call_cfg["agentVoice"], streaming=True)

    session = AgentSession(
        stt=openai.STT(model="gpt-realtime-whisper", use_realtime=True),
        llm=openai.LLM(model="gpt-4o-mini"),
        tts=tts_instance,
        vad=ctx.proc.userdata["vad"],
    )

    prompt = build_payment_prompt(call_cfg)
    opening_line = (
        f"Hello, this is {call_cfg['agentName']}, an automated payment assistance agent from "
        f"{call_cfg['companyName']}. This call may be recorded for quality and compliance. "
        f"Am I speaking with {call_cfg['customerName']}?"
    )

    # 4. Start the Agent Pipeline
    agent = Agent(
        instructions=prompt,
        tools=[verify_borrower_identity, flag_hardship, send_payment_link],
    )
    await session.start(agent, room=ctx.room)

    # 5. Execute the Outbound SIP Dial
    phone_number = "[phone]"  # Retrieved dynamically in full code
    await _dial_and_greet(ctx, session, opening_line, phone_number)


async def _dial_and_greet(ctx, session, opening_line, phone_number):
    """Dials via Twilio SIP trunk and handles the initial greeting."""
    # (SIP connection logic omitted for brevity...)

    # Wait for the audio track so the greeting doesn't clip!
    track_ready = asyncio.Event()
    # Event listener logic here to set track_ready when RTP flows...
    await asyncio.wait_for(track_ready.wait(), timeout=3.0)
    await asyncio.sleep(0.7)

    # Agent speaks before activating STT so it doesn't transcribe itself
    handle = session.say(opening_line, allow_interruptions=False)
    await handle.wait_for_playout()

    # Now start listening to the user
    session.room_io.set_participant("phone-user")


if __name__ == "__main__":
    cli.run_app(
        WorkerOptions(
            entrypoint_fnc=entrypoint,
            prewarm_fnc=prewarm,
            agent_name="payment-agent",
        )
    )

Running it (run.py)

Finally, to orchestrate everything, we create a run.py script. This script triggers the LiveKit dispatches, effectively telling our agent: "Hey, call this number, and here is the customer data payload to inject into agent.py." Because LiveKit abstracts the infrastructure, we can run this sequentially for a single user, or dispatch thousands of calls in parallel reading from a CSV file.

First, we handle our imports and define strictly how we parse phone numbers and dates. In telephony, formatting is everything. If you don't enforce E.164 formatting ([phone]), your SIP trunk will reject the call:

import argparse
import asyncio
import csv
import json
import logging
import os
import re
import subprocess
import sys
import time
import uuid
from datetime import date, datetime

from dotenv import load_dotenv


_ROOT = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, _ROOT)
load_dotenv(os.path.join(_ROOT, ".env"), override=False)

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("run")

_CONFIG_PATH = os.path.join(_ROOT, "scenario_config.json")
_E164_RE = re.compile(r"^\+\d{7,15}$")


def _clean_phone(raw: str) -> str:
    """Return the phone in E.164 format or raise with a helpful message."""
    phone = raw.strip()
    if not _E164_RE.match(phone):
        raise ValueError(
            f"Invalid phone number {phone!r}.\n"
            "  Phone must be in E.164 format: +CountryCodeNumber (e.g. [phone]).\n"
            "  If you edited the CSV in Excel, it likely converted the number to scientific\n"
            "  notation. Open the file in Notepad/VS Code instead."
        )
    return phone


# (Helper functions like _parse_due_date, _days_past_due, _format_inr, and _row_to_meta
# go here to sanitize data before passing it to the agent.)

The LiveKit dispatcher is the bridge between our Python script and LiveKit's cloud infrastructure. When we call _dispatch, we are telling LiveKit to spin up a new room and attach our payment-agent to it, injecting the specific customer's metadata (name, amount due, phone number) straight into the room context.

async def _dispatch(lk, room_name: str, meta: dict) -> None:
    from livekit import api as lk_api
    await lk.room.create_room(lk_api.CreateRoomRequest(name=room_name))
    dispatch = await lk.agent_dispatch.create_dispatch(
        lk_api.CreateAgentDispatchRequest(
            agent_name="payment-agent",
            room=room_name,
            metadata=json.dumps(meta),
        )
    )
    logger.info("Dispatched → room=%s  dispatch=%s", room_name, dispatch.id)


async def _wait_for_room_close(
    lk, room_name: str, timeout_s: int = 600, agent_proc: "subprocess.Popen | None" = None
) -> bool:
    """Poll every 5s until the room disappears (call ended) or timeout."""
    from livekit import api as lk_api
    await asyncio.sleep(15)  # initial wait — dial + ring + answer takes ~10s
    deadline = time.monotonic() + timeout_s
    while time.monotonic() < deadline:
        # Check if agent crashed
        if agent_proc is not None and agent_proc.poll() is not None:
            logger.warning("Agent process exited — deleting room %s and moving on", room_name)
            try:
                await lk.room.delete_room(lk_api.DeleteRoomRequest(room=room_name))
            except Exception:
                pass
            return False

        # Check if room still exists
        try:
            result = await lk.room.list_rooms(lk_api.ListRoomsRequest(names=[room_name]))
            if not result.rooms:
                return True
        except Exception as exc:
            logger.warning("Room poll error: %s", exc)
        await asyncio.sleep(5)
    return False

We'll add three different execution modes:

1. Single phone call
2. Sequential phone calls from a CSV file
3. Parallel phone calls from a CSV file

async def _run_single(phone: str) -> None:
    from livekit import api as lk_api
    phone = _clean_phone(phone)
    with open(_CONFIG_PATH, encoding="utf-8") as f:
        cfg = json.load(f)
    amount   = str(cfg["amountDue"])
    due_date = cfg["dueDate"]
    meta = {
        "phone_number":                phone,
        "customer_name":               cfg["customerName"],
        "amount_due":                  amount,
        "amount_due_formatted":        _format_inr(amount),
        "due_date":                    due_date,
        "days_past_due":               _days_past_due(due_date),
        "account_ending":              str(cfg["accountEnding"]),
        "registered_mobile_last_four": str(cfg["registeredMobileLastFour"]),
    }
    lk = lk_api.LiveKitAPI(
        url=os.environ["LIVEKIT_URL"],
        api_key=os.environ["LIVEKIT_API_KEY"],
        api_secret=os.environ["LIVEKIT_API_SECRET"],
    )
    try:
        room_name = f"payment-{uuid.uuid4().hex[:8]}"
        print(f"\nCalling {phone} (room: {room_name}) …")
        await _dispatch(lk, room_name, meta)
        print("Call dispatched — your phone will ring in ~5 s.")
    finally:
        await lk.aclose()


async def _run_sequential(rows: list[dict], proc: list | None = None) -> None:
    from livekit import api as lk_api
    lk = lk_api.LiveKitAPI(
        url=os.environ["LIVEKIT_URL"],
        api_key=os.environ["LIVEKIT_API_KEY"],
        api_secret=os.environ["LIVEKIT_API_SECRET"],
    )
    try:
        for i, row in enumerate(rows, 1):
            name = row["name"].strip()
            phone = row["phone"].strip()

            # (Agent restart recovery logic goes here...)

            print(f"\n[{i}/{len(rows)}] Calling {name} ({phone}) …")
            room_name = f"payment-{uuid.uuid4().hex[:8]}"
            meta = _row_to_meta(row)
            await _dispatch(lk, room_name, meta)

            print(f"        Waiting for call to finish (room: {room_name}) …")
            agent_p = proc[0] if proc else None
            closed = await _wait_for_room_close(lk, room_name, agent_proc=agent_p)
            status = "done" if closed else "timed out"
            print(f"        {name}: {status}.")
    finally:
        await lk.aclose()


async def _run_parallel(rows: list[dict]) -> None:
    from livekit import api as lk_api
    lk = lk_api.LiveKitAPI(
        url=os.environ["LIVEKIT_URL"],
        api_key=os.environ["LIVEKIT_API_KEY"],
        api_secret=os.environ["LIVEKIT_API_SECRET"],
    )
    try:
        print(f"\nDispatching {len(rows)} calls simultaneously …")
        tasks = []
        for row in rows:
            room_name = f"payment-{uuid.uuid4().hex[:8]}"
            tasks.append(_dispatch(lk, room_name, _row_to_meta(row)))
        await asyncio.gather(*tasks)
        print(f"All {len(rows)} calls dispatched.")
    finally:
        await lk.aclose()

Finally, we wrap it all in an easy-to-use command line interface. We spin up the agent worker process in the background, wait for it to connect to LiveKit, and then execute our chosen mode.

def _start_agent() -> subprocess.Popen:
    """Start the LiveKit agent worker in the background (same terminal output)."""
    return subprocess.Popen(
        [sys.executable, os.path.join(_ROOT, "agent.py"), "start"],
    )


async def main() -> None:
    parser = argparse.ArgumentParser(description="Payment reminder system")
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--to", metavar="PHONE", help="Single call: E.164 phone number")
    group.add_argument("--csv", metavar="FILE", help="Campaign CSV file")
    parser.add_argument("--mode", choices=["sequential", "parallel"], default="sequential")
    args = parser.parse_args()

    print("\nStarting agent worker …")
    proc = [_start_agent()]

    from livekit import api as lk_api
    lk = lk_api.LiveKitAPI(
        url=os.environ["LIVEKIT_URL"],
        api_key=os.environ["LIVEKIT_API_KEY"],
        api_secret=os.environ["LIVEKIT_API_SECRET"],
    )

    try:
        await _wait_for_agent_ready(lk)  # Ensures LiveKit is listening
        print("Agent ready.\n")
        await lk.aclose()

        if args.csv:
            rows = _load_csv(args.csv)
            if args.mode == "parallel":
                await _run_parallel(rows)
            else:
                await _run_sequential(rows, proc=proc)
        else:
            await _run_single(args.to)

        print("\nAgent is running — press Ctrl+C when done.")
        while True:
            if proc[0].poll() is not None:
                break
            await asyncio.sleep(2)

    except KeyboardInterrupt:
        print("\nStopping …")
    finally:
        if proc[0].poll() is None:
            proc[0].terminate()


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

To run the agent, use one of the three commands below:

python run.py --to [phone]
python run.py --csv reminders.csv --mode sequential
python run.py --csv reminders.csv --mode parallel

Note: Don't use Excel to edit the CSV files — phone numbers get converted to scientific notation, which throws an error. Edit them in Notepad or your code editor.

---

Layer 2: The Enforcer

In this layer, we implement a state machine. Instead of hoping the model follows the required flow, we take away its autonomy. The core mechanism here is data starvation — the LLM is physically locked out of the financial data until the state machine confirms the caller's identity.

Defining the rules (state_machine.py)

First, we map the entire lifecycle of a compliance-bound phone call into discrete, locked phases. Create a state_machine.py file. Instead of letting the LLM wander, we define exact states, the only actions allowed in those states, and the strict paths it can take to move forward.

from __future__ import annotations

import logging
from enum import Enum

logger = logging.getLogger(__name__)


class CallState(str, Enum):
    PRE_CALL_CHECK = "PRE_CALL_CHECK"
    OPENING_DISCLOSURE = "OPENING_DISCLOSURE"
    IDENTITY_VERIFICATION = "IDENTITY_VERIFICATION"
    PAYMENT_CONTEXT = "PAYMENT_CONTEXT"
    INTENT_CLASSIFICATION = "INTENT_CLASSIFICATION"
    SEND_PAYMENT_LINK = "SEND_PAYMENT_LINK"
    PROMISE_TO_PAY = "PROMISE_TO_PAY"
    DISPUTE_INTAKE = "DISPUTE_INTAKE"
    HARDSHIP_ESCALATION = "HARDSHIP_ESCALATION"
    WRONG_PERSON_END = "WRONG_PERSON_END"
    HUMAN_HANDOFF = "HUMAN_HANDOFF"
    CALL_SUMMARY = "CALL_SUMMARY"


# Once we reach one of these, the call is winding down — no more payment flow.
TERMINAL_STATES: frozenset[CallState] = frozenset(
    {CallState.WRONG_PERSON_END, CallState.HUMAN_HANDOFF, CallState.CALL_SUMMARY}
)


# What the agent is allowed to do in each state. The prompt is rebuilt from this
# every time we transition, so the LLM only ever sees the actions for the current phase.
ALLOWED_ACTIONS: dict[CallState, list[str]] = {
    CallState.PRE_CALL_CHECK: [],
    CallState.OPENING_DISCLOSURE: [
        "introduce yourself and the company",
        "state the call may be recorded",
        "ask if speaking with the customer by name",
    ],
    CallState.IDENTITY_VERIFICATION: [
        "ask for the last four digits of the registered mobile number",
        "ask for the last four digits of the account number",
        "call verify_borrower_identity",
    ],
    CallState.PAYMENT_CONTEXT: [
        "state amount due and due date",
        "offer to send the payment link",
    ],
    CallState.INTENT_CLASSIFICATION: [
        "listen for payment intent, dispute, hardship, or stop-calling",
        "route to the appropriate next state",
    ],
    CallState.SEND_PAYMENT_LINK: [
        "call send_payment_link",
        "confirm the link was sent to the registered number",
    ],
    CallState.PROMISE_TO_PAY: [
        "ask for a commitment date",
        "call log_promise_to_pay",
    ],
    CallState.DISPUTE_INTAKE: [
        "call create_dispute_ticket",
        "stop all payment reminder language",
    ],
    CallState.HARDSHIP_ESCALATION: [
        "call flag_hardship",
        "do not pressure the borrower",
    ],
    CallState.WRONG_PERSON_END: [
        "call end_call_wrong_person",
        "apologise and end the call",
    ],
    CallState.HUMAN_HANDOFF: [
        "call transfer_to_human",
        "inform the borrower they are being transferred",
    ],
    CallState.CALL_SUMMARY: [
        "summarise the outcome",
        "thank the borrower and close the call politely",
    ],
}


# Strict one-way streets. You cannot jump from Opening straight to Payment Context —
# the only path to money talk runs through IDENTITY_VERIFICATION.
VALID_TRANSITIONS: dict[CallState, set[CallState]] = {
    CallState.PRE_CALL_CHECK: {
        CallState.OPENING_DISCLOSURE,
    },
    CallState.OPENING_DISCLOSURE: {
        CallState.IDENTITY_VERIFICATION,
        CallState.WRONG_PERSON_END,
        CallState.DISPUTE_INTAKE,
        CallState.HARDSHIP_ESCALATION,
        CallState.HUMAN_HANDOFF,
    },
    CallState.IDENTITY_VERIFICATION: {
        CallState.PAYMENT_CONTEXT,
        CallState.WRONG_PERSON_END,
        CallState.DISPUTE_INTAKE,
        CallState.HARDSHIP_ESCALATION,
        CallState.HUMAN_HANDOFF,
    },
    CallState.PAYMENT_CONTEXT: {
        CallState.INTENT_CLASSIFICATION,
        CallState.DISPUTE_INTAKE,
        CallState.HARDSHIP_ESCALATION,
        CallState.HUMAN_HANDOFF,
    },
    CallState.INTENT_CLASSIFICATION: {
        CallState.SEND_PAYMENT_LINK,
        CallState.PROMISE_TO_PAY,
        CallState.DISPUTE_INTAKE,
        CallState.HARDSHIP_ESCALATION,
        CallState.HUMAN_HANDOFF,
        CallState.CALL_SUMMARY,
    },
    CallState.SEND_PAYMENT_LINK: {
        CallState.PROMISE_TO_PAY,
        CallState.CALL_SUMMARY,
        CallState.DISPUTE_INTAKE,
        CallState.HARDSHIP_ESCALATION,
        CallState.HUMAN_HANDOFF,
    },
    CallState.PROMISE_TO_PAY: {
        CallState.CALL_SUMMARY,
        CallState.HUMAN_HANDOFF,
    },
    CallState.DISPUTE_INTAKE: {
        CallState.HUMAN_HANDOFF,
        CallState.CALL_SUMMARY,
    },
    CallState.HARDSHIP_ESCALATION: {
        CallState.HUMAN_HANDOFF,
        CallState.CALL_SUMMARY,
    },
    CallState.WRONG_PERSON_END: set(),
    CallState.HUMAN_HANDOFF: set(),
    CallState.CALL_SUMMARY: set(),
}


class CallStateMachine:
    def __init__(self) -> None:
        self._state: CallState = CallState.PRE_CALL_CHECK
        self._history: list[str] = [CallState.PRE_CALL_CHECK.value]

    @property
    def current_state(self) -> CallState:
        return self._state

    @property
    def allowed_actions(self) -> list[str]:
        return list(ALLOWED_ACTIONS[self._state])

    def transition(self, new_state: CallState) -> bool:
        if new_state not in VALID_TRANSITIONS.get(self._state, set()):
            logger.warning(
                "Invalid transition %s -> %s — staying in %s",
                self._state.value,
                new_state.value,
                self._state.value,
            )
            return False
        logger.info("State: %s -> %s", self._state.value, new_state.value)
        self._state = new_state
        self._history.append(new_state.value)
        return True

    def is_terminal(self) -> bool:
        return self._state in TERMINAL_STATES

    def history(self) -> list[str]:
        return list(self._history)

Changes to the prompt (prompt.py)

Now we rewrite our build_payment_prompt function to accept the current state and a boolean flag: identity_verified. Notice the if identity_verified: block. If the state machine hasn't cleared the user, the LLM is literally not given the amount or due date variables, making it nearly impossible for the LLM to hallucinate or leak the debt amount.

Replace the entire build_payment_prompt function from Layer 1 with this:

def build_payment_prompt(
    config: dict,
    identity_verified: bool,
    current_state: str,
    allowed_actions: list[str],
) -> str:
    agent_name = config["agentName"]
    company = config["companyName"]
    customer = config["customerName"]
    amount = config["amountDueFormatted"]
    due = config["dueDate"]

    if identity_verified:
        flow_step4 = f"4. Only after verification: mention the amount due ({amount}) and due date ({due})."
        identity_block = (
            "--- IDENTITY STATUS ---\n"
            "{\n"
            '  "identity_verified": true,\n'
            f'  "customer_name": "{customer}",\n'
            f'  "amount_due": "{amount}",\n'
            f'  "due_date": "{due}"\n'
            "}"
        )
    else:
        # The agent is physically locked out of the financial data.
        flow_step4 = "4. Only after verification: mention the amount due (it will appear in the status block once verified)."
        identity_block = (
            "--- IDENTITY STATUS ---\n"
            "{\n"
            '  "identity_verified": false,\n'
            '  "instruction": "Do not reveal any account details until identity is verified."\n'
            "}"
        )

    actions_list = "\n".join(f"- {a}" for a in allowed_actions) or "- (no actions available in this state)"
    state_block = (
        "---\n"
        f"CURRENT STATE: {current_state}\n\n"
        "IN THIS STATE YOU MAY ONLY:\n"
        f"{actions_list}\n\n"
        "Do not perform actions from other states. Do not skip states.\n"
        "---"
    )

    return (
        f"You are {agent_name}, an automated payment assistance voice agent from {company}.\n\n"
        "CORE RULES — NEVER BREAK THESE\n"
        "- Do not reveal the customer's name, account number, amount due, or overdue status\n"
        "  to anyone until identity has been verified.\n"
        "- If the borrower mentions job loss, inability to pay, or a medical emergency,\n"
        "  call flag_hardship immediately. Do not pressure them.\n"
        "- Never make threats, use intimidating language, or shame the borrower.\n"
        "- Never mention the borrower's family, employer, or references.\n\n"
        "CONVERSATION FLOW\n"
        f"1. Introduce yourself and the company. State the call may be recorded. Ask if you are speaking with {customer}.\n"
        "2. If confirmed, ask for the last four digits of their registered mobile number and account number.\n"
        "3. Call verify_borrower_identity with both sets of digits.\n"
        f"{flow_step4}\n"
        "5. Offer to send the official payment link with send_payment_link.\n"
        "6. Close the call politely.\n\n"
        f"{identity_block}\n\n"
        f"{state_block}\n\n"
        "LANGUAGE AND TONE\n"
        "Keep every response to 1-2 sentences. Speak clearly and calmly. Do not use filler\n"
        'phrases like "Certainly!" or "Of course!".'
    )

Wiring the changes (agent.py)

To make this work live on a call, we intercept the user's audio transcripts before the LLM generates a response. We use LiveKit's event listeners to track the conversation, evaluate the state, and use update_instructions() to seamlessly rewrite the LLM's system prompt.

Add an import:

from state_machine import CallState, CallStateMachine

Add this helper function outside entrypoint, after _dial_and_greet:

def _update_agent_instructions(session, call_cfg, identity_verified, sm) -> None:
    """Rebuild the prompt for the machine's current state and push it live."""
    new_prompt = build_payment_prompt(
        call_cfg, identity_verified, sm.current_state.value, sm.allowed_actions
    )
    # Hot-swaps the LLM's system prompt mid-call without dropping audio.
    asyncio.create_task(session.current_agent.update_instructions(new_prompt))

Inside entrypoint, replace the outcome_log line:

outcome_log = data.OutcomeLog(scenario=call_cfg["scenario"])
ctx.proc.userdata["outcome_log"] = outcome_log

# Layer 2: the state machine governs the flow of the call.
sm = CallStateMachine()
identity_verified = False
sm.transition(CallState.OPENING_DISCLOSURE)

Change the prompt = build_payment_prompt line to pass the new arguments:

prompt = build_payment_prompt(
    call_cfg, identity_verified, sm.current_state.value, sm.allowed_actions
)

Add the turn processor and transcript listener before await:

def _process_user_turn(utterance: str) -> None:
    nonlocal identity_verified
    if not utterance:
        return

    # Advance from greeting to verification once the user replies.
    if sm.current_state == CallState.OPENING_DISCLOSURE:
        sm.transition(CallState.IDENTITY_VERIFICATION)
        _update_agent_instructions(session, call_cfg, identity_verified, sm)

    # verify_borrower_identity succeeded → unlock the financial data.
    if (
        outcome_log.identity_verified
        and not identity_verified
        and sm.current_state == CallState.IDENTITY_VERIFICATION
    ):
        identity_verified = True
        outcome_log.amount_disclosed = True
        sm.transition(CallState.PAYMENT_CONTEXT)
        _update_agent_instructions(session, call_cfg, identity_verified, sm)


@session.on("user_input_transcribed")
def on_user_input_transcribed(event) -> None:
    if event.is_final and event.transcript:
        _process_user_turn(event.transcript)

---

Layer 3: The Safety Net

Now we have a proper prompt and an established flow for the call. But what happens if the LLM misinterprets the user? What if the caller says, "I lost my job, I can't pay right now," and the LLM simply responds with, "I understand, but your payment is still due on the 21st" instead of calling the hardship tool? In debt collection, ignoring a hardship disclosure is a major violation.

We cannot leave compliance entirely up to probabilistic tool-calling. We need guardrails.

Layer 3 introduces a rules engine that listens to every single word spoken on the call. If it detects high-risk language, it physically forces the state machine onto a safe path, regardless of what the LLM decides to do.

The rules (guardrails.py)

from __future__ import annotations


class GuardrailEngine:

    @staticmethod
    def check_pre_call(scenario: str) -> tuple[bool, str]:
        """Returns (can_proceed, block_reason). Blocks the grievance_pending scenario."""
        if scenario == "grievance_pending":
            return (False, "Active grievance ticket on file — call blocked until resolved.")
        return (True, "")

    @staticmethod
    def should_stop_payment_flow(utterance: str) -> tuple[bool, str]:
        """Checks the utterance for triggers that stop the normal payment flow.

        Returns (True, reason) where reason is one of:
        'dispute', 'hardship', 'human_requested', 'stop_calling'.
        """
        text = utterance.lower()

        dispute_phrases = [
            "already paid", "paid already", "paid this", "dispute",
            "wrong amount", "incorrect amount", "i didn't borrow",
            "i did not borrow", "don't owe", "do not owe",
        ]
        for phrase in dispute_phrases:
            if phrase in text:
                return (True, "dispute")

        hardship_phrases = [
            "lost my job", "lost job", "cannot pay", "can't pay", "cant pay",
            "unable to pay", "medical emergency", "death in family",
            "in the hospital", "in hospital", "unemployed", "no income",
            "financial hardship",
        ]
        for phrase in hardship_phrases:
            if phrase in text:
                return (True, "hardship")

        human_phrases = [
            "speak to a human", "talk to a human", "speak to a person",
            "talk to a person", "real person", "real agent", "human agent",
            "escalate",
        ]
        for phrase in human_phrases:
            if phrase in text:
                return (True, "human_requested")

        stop_phrases = [
            "stop calling", "stop calling me", "remove me", "do not call",
            "don't call", "opt out", "take me off",
        ]
        for phrase in stop_phrases:
            if phrase in text:
                return (True, "stop_calling")

        return (False, "")

    @staticmethod
    def is_prohibited_language(agent_text: str) -> tuple[bool, str]:
        """Monitors the AI's own output for illegal collection threats."""
        text = agent_text.lower()
        prohibited = [
            "legal action", "contact your family", "contact your employer",
            "contact your references", "tell your family", "tell your employer",
        ]
        for phrase in prohibited:
            if phrase in text:
                return (True, phrase)
        return (False, "")

    @staticmethod
    def check_wrong_person(utterance: str, expected_name: str) -> bool:
        """Returns True if the utterance indicates the caller is not the expected person."""
        text = utterance.lower()
        name = expected_name.lower()
        wrong_person_phrases = [
            "wrong number", "wrong person", "you have the wrong",
            f"not {name}", f"no {name} here", f"no {name}",
            "different person", "nobody by that name",
        ]
        for phrase in wrong_person_phrases:
            if phrase in text:
                return True
        return False

Enforcing the guardrails (agent.py)

Import the guardrails we just created at the top of the file:

from guardrails import GuardrailEngine

In the entrypoint() function, right after the sm = CallStateMachine() block, create the guardrail engine and run the pre-call check:

guardrails = GuardrailEngine()

# Layer 3: refuse to dial a blocked account.
can_proceed, block_reason = guardrails.check_pre_call(call_cfg["scenario"])
if not can_proceed:
    logger.warning("Call blocked: %s", block_reason)
    outcome_log.outcome = "blocked_pre_call"
    outcome_log.save_to_file()
    return  # entrypoint returns before dialing

Replace the _process_user_turn function with this:

def _process_user_turn(utterance: str) -> None:
    nonlocal identity_verified
    if not utterance:
        return

    # >>> LAYER 3: deterministic safety net on the caller's words. Overrides the LLM.
    # Wrong-person check while we're still in the opening / verification phase.
    if sm.current_state in (CallState.OPENING_DISCLOSURE, CallState.IDENTITY_VERIFICATION):
        if guardrails.check_wrong_person(utterance, call_cfg["customerName"]):
            sm.transition(CallState.WRONG_PERSON_END)
            _update_agent_instructions(session, call_cfg, identity_verified, sm)

    # Stop-flow triggers: dispute, hardship, or a request for a human / to stop calling.
    if not sm.is_terminal():
        stop, reason = guardrails.should_stop_payment_flow(utterance)
        if stop:
            if reason == "dispute":
                outcome_log.dispute_detected = True
                outcome_log.outcome = "payment_dispute"
                sm.transition(CallState.DISPUTE_INTAKE)
            elif reason == "hardship":
                outcome_log.hardship_detected = True
                outcome_log.outcome = "hardship_detected"
                sm.transition(CallState.HARDSHIP_ESCALATION)
            elif reason in ("human_requested", "stop_calling"):
                sm.transition(CallState.HUMAN_HANDOFF)
            _update_agent_instructions(session, call_cfg, identity_verified, sm)

    # >>> LAYER 2: normal state advancement (only runs if no guardrail fired above).
    if sm.current_state == CallState.OPENING_DISCLOSURE:
        sm.transition(CallState.IDENTITY_VERIFICATION)
        _update_agent_instructions(session, call_cfg, identity_verified, sm)

    if (
        outcome_log.identity_verified
        and not identity_verified
        and sm.current_state == CallState.IDENTITY_VERIFICATION
    ):
        identity_verified = True
        outcome_log.amount_disclosed = True
        sm.transition(CallState.PAYMENT_CONTEXT)
        _update_agent_instructions(session, call_cfg, identity_verified, sm)

And finally, add a listener to monitor the agent's output:

@session.on("conversation_item_added")
def on_conversation_item_added(event) -> None:
    item = event.item
    if item.type == "message" and item.role == "assistant" and item.text_content:
        # >>> LAYER 3: monitor the AI's own output for prohibited language.
        is_prohibited, phrase = guardrails.is_prohibited_language(item.text_content)
        if is_prohibited:
            logger.warning("Prohibited language in agent output: %r", phrase)
            outcome_log.prohibited_language_detected = True

By layering strict keyword interception over our state machine, we've built an architecture that doesn't just ask the AI to behave — it forces it to. Your agent is no longer a fragile prompt wrapper; it is a governed, audited, compliance-first communication system.

---

Common Errors

Error	Cause	Fix
Required environment variable 'X' is not set	Missing .env value	Copy .env.example to .env and fill in the variable
Agent answers but stays silent	Dispatch rule has no agents block	Edit the rule in LiveKit Cloud — add agentName to roomConfig.agents
DuplexClosed in logs, call drops mid-greeting	dev mode restarts on file save	Use python agent.py start for all phone testing
Call drops immediately	TwiML Bin not reachable, or <Sip> URI missing ;transport=tcp	Check the URI in the TwiML Bin and add ;transport=tcp at the end
401 or 403 from Murf or Deepgram	Wrong or expired API key	Re-check MURF_API_KEY and DEEPGRAM_API_KEY in .env

---

By separating the stack into a baseline conversationalist (Layer 1), a strict state machine (Layer 2), and an unyielding set of semantic guardrails (Layer 3), we've transformed a probabilistic AI into a compliant communication tool.

We've only scratched the surface of BFSI compliance in this tutorial. As your use cases grow and change, so will your regulatory requirements. But by shifting control from the LLM's prompt to your deterministic Python code, you can handle any edge case simply by adding a state or a guardrail.

You can find the complete source code for this project on GitHub: https://github.com/murf-ai/murf-cookbook/tree/main/examples/agents/payment-reminder

Now take this foundation, tune the guardrails to match your requirements, and build your own compliant voice agent.