Voice Customer Support Agent: Mercury + Tool Calling + Voice

This cookbook builds a customer support agent for a dental clinic on top of Mercury 2, Inception’s diffusion large language model. The agent understands natural-language requests, calls tools to read and write a live appointment database, and holds a real-time voice conversation using ElevenLabs for speech-to-text and text-to-speech. By the end you will have a working agent that can:

Look up a patient and their upcoming appointments
Show open appointment slots for a given day or provider
Book, reschedule, and cancel appointments via real mutations against a database
Do all of the above out loud, over streamed audio

Because Mercury 2 generates tokens with a diffusion process rather than autoregressively, it returns tool calls and final answers faster than comparable models. For a voice customer support agent, where every extra 100 ms is audible, this difference is critical.

Run the Customer Support Agent

Directly via Google Colab or download the .ipynb to run it locally in Jupyter

Architecture

The clinic database is an in-memory SQLite instance seeded with providers, patients, and an availability grid. Two API keys (INCEPTION_API_KEY, ELEVENLABS_API_KEY) are required to run this demo.

1. Setup

Install the Inception SDK, the ElevenLabs SDK, and a few audio helpers.

%pip install -q "inceptionai" "elevenlabs>=2.15.0" \
    "numpy>=1.26.0" "scipy>=1.16.2" "pydub>=0.25.1"

You need two API keys:

INCEPTION_API_KEY — from the Inception dashboard (new accounts get 100M free tokens)
ELEVENLABS_API_KEY — from the ElevenLabs dashboard

The next cell reads them from the environment, and prompts you for any that aren’t already set. Nothing is written to disk.

import os
import json
import getpass
import time
import sqlite3
from datetime import datetime, timedelta

from inceptionai import Inception

for var in ("INCEPTION_API_KEY", "ELEVENLABS_API_KEY"):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")

client = Inception(api_key=os.environ["INCEPTION_API_KEY"])
MERCURY_MODEL = "mercury-2"

print("Inception client ready. Model:", MERCURY_MODEL)

2. A mock clinic database

We spin up an in-memory SQLite database for a fabricated dental clinic: Diffusion Dental. It contains three tables:

providers — the dentists and hygienists
patients — existing patient records
appointments — a grid of 30-minute slots, each available, booked, or cancelled

The availability grid is generated relative to the moment you run the cell. It starts with whatever is left of today and rolls forward over the next five business days, skipping weekends and any hour that has already passed. A couple of slots start out pre-booked so availability looks realistic. This is a placeholder for demo purposes; you can plug in your own custom db here.

CLINIC_HOURS  = (9, 10, 11, 14, 15, 16)   # 30-min slots, morning and afternoon
BUSINESS_DAYS = 5                         # how many open days to put on the calendar


def build_clinic_db(now=None):
    """Create and seed an in-memory clinic database. Returns a live connection.

    The availability grid is built relative to `now` (default: the current date
    and time), so the calendar is always current no matter when you run this.
    Slots start with whatever is left of today and roll forward over the next
    few business days; times that have already passed are never offered.
    """
    now = now or datetime.now()

    conn = sqlite3.connect(":memory:")
    conn.row_factory = sqlite3.Row
    cur = conn.cursor()

    cur.executescript(
        """
        CREATE TABLE providers (
            id        INTEGER PRIMARY KEY,
            name      TEXT NOT NULL,
            specialty TEXT NOT NULL
        );
        CREATE TABLE patients (
            id    INTEGER PRIMARY KEY,
            name  TEXT NOT NULL,
            phone TEXT NOT NULL,
            email TEXT
        );
        CREATE TABLE appointments (
            id           INTEGER PRIMARY KEY,
            patient_id   INTEGER,
            provider_id  INTEGER NOT NULL,
            start_time   TEXT NOT NULL,               -- ISO 8601, e.g. 2026-08-17T14:00
            duration_min INTEGER NOT NULL DEFAULT 30,
            reason       TEXT,
            status       TEXT NOT NULL DEFAULT 'available',  -- available | booked | cancelled
            FOREIGN KEY (patient_id)  REFERENCES patients(id),
            FOREIGN KEY (provider_id) REFERENCES providers(id)
        );
        """
    )

    # Deliberately cartoonish names - nobody should mistake this for real PHI.
    cur.executemany("INSERT INTO providers VALUES (?,?,?)", [
        (1, "Dr. Ada Molar",       "General Dentistry"),
        (2, "Dr. Reed Wyre",       "Orthodontics"),
        (3, "Dr. Pearl Flossmore", "Hygiene & Cleaning"),
    ])

    cur.executemany("INSERT INTO patients VALUES (?,?,?,?)", [
        (1, "Penny Placeholder", "+1-415-555-0101", "[email protected]"),
        (2, "Sam Sampleton",     "+1-415-555-0102", "[email protected]"),
        (3, "Molly Mockford",    "+1-415-555-0103", "[email protected]"),
    ])

    # Walk forward from today, collecting BUSINESS_DAYS weekdays that still have
    # open hours left. Weekends are skipped, and so is the rest of today once the
    # clinic has closed - that day simply doesn't count toward the total.
    rows, appt_id = [], 1
    day = now.replace(hour=0, minute=0, second=0, microsecond=0)
    days_added = 0
    while days_added < BUSINESS_DAYS:
        if day.weekday() < 5:                                    # Mon-Fri
            starts = [day.replace(hour=h, minute=0) for h in CLINIC_HOURS]
            starts = [s for s in starts if s > now]               # never offer the past
            if starts:
                for start in starts:
                    for provider_id in (1, 2, 3):
                        rows.append((appt_id, None, provider_id,
                                     start.strftime("%Y-%m-%dT%H:%M"), 30, None, "available"))
                        appt_id += 1
                days_added += 1
        day += timedelta(days=1)
    cur.executemany("INSERT INTO appointments VALUES (?,?,?,?,?,?,?)", rows)

    # Pre-book a couple of slots so the calendar isn't empty. Picked by position
    # rather than by hard-coded date, so they land on real slots on any run.
    cur.execute("""UPDATE appointments SET patient_id=1, reason='Cleaning', status='booked'
                   WHERE id = (SELECT id FROM appointments
                               WHERE provider_id=3 AND status='available'
                               ORDER BY start_time LIMIT 1)""")
    cur.execute("""UPDATE appointments SET patient_id=2, reason='Braces adjustment', status='booked'
                   WHERE id = (SELECT id FROM appointments
                               WHERE provider_id=2 AND status='available'
                                 AND start_time LIKE '%T14:00'
                               ORDER BY start_time LIMIT 1 OFFSET 1)""")

    conn.commit()
    return conn


conn = build_clinic_db()

# Peek at the seed data.
print("Providers:")
for r in conn.execute("SELECT * FROM providers"):
    print(" ", dict(r))

first, last = conn.execute("SELECT MIN(start_time), MAX(start_time) FROM appointments").fetchone()
n_avail = conn.execute("SELECT COUNT(*) FROM appointments WHERE status='available'").fetchone()[0]
n_booked = conn.execute("SELECT COUNT(*) FROM appointments WHERE status='booked'").fetchone()[0]
print(f"\nCalendar runs {first} -> {last}")
print(f"Slots: {n_avail} available, {n_booked} booked")

3. Define the support tools

Each tool is a plain Python function that reads or writes the database. We keep the functions small and return dictionaries, which then get handed back to Mercury 2 as the tool result. We include a mix of read tools (find_patient, list_available_slots, get_appointments_for_patient) and write tools (book_appointment, reschedule_appointment, cancel_appointment). The write tools validate state first: you can’t book a slot that’s already taken, or cancel an appointment that doesn’t exist. If you have a different database setup, you can customize these functions to match your database-specific query logic. If you do, make sure you update your tool schemas (next section) accordingly.

_TITLES = {"dr", "doctor", "mr", "mrs", "ms", "prof"}


def _provider_id_from_name(conn, provider_name):
    """Resolve a provider from however the caller referred to them.

    Records hold the full name ("Dr. Pearl Flossmore") but a caller on the phone
    will just say a surname, so match on the distinctive words and ignore titles.
    """
    if not provider_name:
        return None
    words = [w for w in provider_name.lower().replace(".", " ").split() if w not in _TITLES]
    if not words:
        return None
    q = "SELECT id FROM providers WHERE " + " AND ".join(["LOWER(name) LIKE ?"] * len(words))
    row = conn.execute(q, [f"%{w}%" for w in words]).fetchone()
    return row["id"] if row else None


def find_patient(conn, name=None, phone=None):
    """Look up a patient by (partial) name and/or phone number."""
    q, params = "SELECT * FROM patients WHERE 1=1", []
    if name:
        q += " AND name LIKE ?";  params.append(f"%{name}%")
    if phone:
        q += " AND phone LIKE ?"; params.append(f"%{phone}%")
    rows = [dict(r) for r in conn.execute(q, params).fetchall()]
    return {"patients": rows, "count": len(rows)}


def list_available_slots(conn, date=None, provider_name=None):
    """List open appointment slots, optionally filtered by date (YYYY-MM-DD)
    and/or provider surname."""
    q = ("SELECT a.id, a.start_time, a.duration_min, p.name AS provider, p.specialty "
         "FROM appointments a JOIN providers p ON p.id = a.provider_id "
         "WHERE a.status = 'available'")
    params = []
    if date:
        q += " AND a.start_time LIKE ?"; params.append(f"{date}%")
    pid = _provider_id_from_name(conn, provider_name)
    if provider_name and pid is None:
        return {"error": f"No provider matching '{provider_name}'."}
    if pid:
        q += " AND a.provider_id = ?"; params.append(pid)
    q += " ORDER BY a.start_time LIMIT 20"
    rows = [dict(r) for r in conn.execute(q, params).fetchall()]
    return {"available_slots": rows, "count": len(rows)}


def book_appointment(conn, slot_id, patient_name, patient_phone, reason=None):
    """Book an available slot for a patient, creating the patient if new."""
    slot = conn.execute("SELECT * FROM appointments WHERE id=?", (slot_id,)).fetchone()
    if slot is None:
        return {"error": f"No slot with id {slot_id}."}
    if slot["status"] != "available":
        return {"error": f"Slot {slot_id} is not available (status: {slot['status']})."}

    existing = conn.execute("SELECT * FROM patients WHERE phone=?", (patient_phone,)).fetchone()
    if existing:
        patient_id = existing["id"]
    else:
        cur = conn.execute("INSERT INTO patients (name, phone) VALUES (?, ?)",
                           (patient_name, patient_phone))
        patient_id = cur.lastrowid

    conn.execute("UPDATE appointments SET patient_id=?, reason=?, status='booked' WHERE id=?",
                 (patient_id, reason, slot_id))
    conn.commit()
    confirmed = conn.execute(
        "SELECT a.id, a.start_time, p.name AS provider "
        "FROM appointments a JOIN providers p ON p.id=a.provider_id WHERE a.id=?",
        (slot_id,)).fetchone()
    return {"status": "booked", "confirmation": dict(confirmed), "patient_id": patient_id}


def reschedule_appointment(conn, appointment_id, new_slot_id):
    """Move a booking to a new available slot, freeing the old one."""
    old = conn.execute("SELECT * FROM appointments WHERE id=?", (appointment_id,)).fetchone()
    if old is None or old["status"] != "booked":
        return {"error": f"Appointment {appointment_id} is not an active booking."}
    new = conn.execute("SELECT * FROM appointments WHERE id=?", (new_slot_id,)).fetchone()
    if new is None or new["status"] != "available":
        return {"error": f"Slot {new_slot_id} is not available."}

    conn.execute("UPDATE appointments SET patient_id=?, reason=?, status='booked' WHERE id=?",
                 (old["patient_id"], old["reason"], new_slot_id))
    conn.execute("UPDATE appointments SET patient_id=NULL, reason=NULL, status='available' WHERE id=?",
                 (appointment_id,))
    conn.commit()
    return {"status": "rescheduled", "from_id": appointment_id, "to_id": new_slot_id}


def cancel_appointment(conn, appointment_id):
    """Cancel a booking and return the slot to the available pool."""
    appt = conn.execute("SELECT * FROM appointments WHERE id=?", (appointment_id,)).fetchone()
    if appt is None or appt["status"] != "booked":
        return {"error": f"Appointment {appointment_id} is not an active booking."}
    conn.execute("UPDATE appointments SET patient_id=NULL, reason=NULL, status='available' WHERE id=?",
                 (appointment_id,))
    conn.commit()
    return {"status": "cancelled", "appointment_id": appointment_id}


def get_appointments_for_patient(conn, name=None, phone=None):
    """List a patient's active (booked) appointments."""
    q = ("SELECT a.id, a.start_time, a.reason, a.status, p.name AS provider "
         "FROM appointments a "
         "JOIN patients pt ON pt.id = a.patient_id "
         "JOIN providers p ON p.id = a.provider_id "
         "WHERE a.status='booked'")
    params = []
    if name:
        q += " AND pt.name LIKE ?";  params.append(f"%{name}%")
    if phone:
        q += " AND pt.phone LIKE ?"; params.append(f"%{phone}%")
    q += " ORDER BY a.start_time"
    rows = [dict(r) for r in conn.execute(q, params).fetchall()]
    return {"appointments": rows, "count": len(rows)}

print("Tool functions defined.")

Tool schemas

Mercury 2 uses the OpenAI-compatible function-calling format. Each schema tells the model the tool’s name, what it does, and its parameters. This is how the model decides which tool to reach for and how to fill in the arguments. You can also plug in custom tool schemas; just make sure they are compatible with the support tools you’ve defined above.

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "find_patient",
            "description": "Look up an existing patient by name and/or phone number.",
            "strict": True,
            "parameters": {
                "type": "object",
                "properties": {
                    "name":  {"type": "string", "description": "Full or partial patient name"},
                    "phone": {"type": "string", "description": "Full or partial phone number"},
                },
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "list_available_slots",
            "description": ("List OPEN appointment slots. Filter by date (format "
                            "YYYY-MM-DD) and/or provider surname. Call this before booking "
                            "so you can offer the patient a real, open slot id."),
            "strict": True,
            "parameters": {
                "type": "object",
                "properties": {
                    "date":          {"type": "string", "description": "Day to search, YYYY-MM-DD"},
                    "provider_name": {"type": "string", "description": "Provider surname, e.g. 'Flossmore'"},
                },
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "book_appointment",
            "description": ("Book a specific available slot for a patient. Requires the "
                            "slot_id (from list_available_slots), the patient's name, and "
                            "their phone number."),
            "strict": True,
            "parameters": {
                "type": "object",
                "properties": {
                    "slot_id":       {"type": "integer", "description": "id of an available slot"},
                    "patient_name":  {"type": "string"},
                    "patient_phone": {"type": "string"},
                    "reason":        {"type": "string", "description": "Reason for the visit"},
                },
                "required": ["slot_id", "patient_name", "patient_phone"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "reschedule_appointment",
            "description": "Move an existing booking to a new available slot.",
            "strict": True,
            "parameters": {
                "type": "object",
                "properties": {
                    "appointment_id": {"type": "integer", "description": "id of the current booking"},
                    "new_slot_id":    {"type": "integer", "description": "id of the new available slot"},
                },
                "required": ["appointment_id", "new_slot_id"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "cancel_appointment",
            "description": "Cancel an existing booking and free the slot.",
            "strict": True,
            "parameters": {
                "type": "object",
                "properties": {
                    "appointment_id": {"type": "integer", "description": "id of the booking to cancel"},
                },
                "required": ["appointment_id"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_appointments_for_patient",
            "description": "List a patient's upcoming (booked) appointments by name and/or phone.",
            "strict": True,
            "parameters": {
                "type": "object",
                "properties": {
                    "name":  {"type": "string"},
                    "phone": {"type": "string"},
                },
            },
        },
    },
]

# Map tool names to the functions that run them (bound to our connection).
TOOL_RUNNERS = {
    "find_patient":                 lambda **kw: find_patient(conn, **kw),
    "list_available_slots":         lambda **kw: list_available_slots(conn, **kw),
    "book_appointment":             lambda **kw: book_appointment(conn, **kw),
    "reschedule_appointment":       lambda **kw: reschedule_appointment(conn, **kw),
    "cancel_appointment":           lambda **kw: cancel_appointment(conn, **kw),
    "get_appointments_for_patient": lambda **kw: get_appointments_for_patient(conn, **kw),
}

print(f"{len(TOOLS)} tools registered.")

4. The agent loop

The loop is the heart of a tool-calling agent:

Send the conversation plus the tool schemas to Mercury 2.
If the reply contains tool_calls, run each one locally and append the result as a role: "tool" message.
Repeat until Mercury returns a normal text answer.

Using this workflow, the loop naturally handles multi-step requests like “reschedule my cleaning to Thursday” (look up the booking → find a Thursday slot → move it). We query mercury-2 on low reasoning effort: real time customer service is a very latency-sensitive use-case where Mercury’s ability to complete complicated tool-calling tasks at ultra-low latencies shines.

TODAY = datetime.now()

SYSTEM_PROMPT = f"""You are the friendly virtual receptionist for Diffusion Dental.
You help callers check availability and book, reschedule, or cancel appointments.

Guidelines:
- Today's date is {TODAY.strftime("%A, %B")} {TODAY.day}, {TODAY.year}. The clinic is open Monday-Friday.
- The clinic's providers go by Molar (general dentistry), Wyre (orthodontics),
  and Flossmore (hygiene and cleaning).
- Always use the tools to look up real data; never invent slot ids, times, or
  patient records.
- Before booking, call list_available_slots and offer the caller a real open slot.
- To book you need the caller's name and phone number - ask if you don't have them.
- Confirm the concrete date, time, and provider back to the caller after any change.
- Keep replies short and conversational - they may be read aloud over the phone."""


def run_agent(user_message, history=None, max_turns=6, verbose=True):
    """Run one user turn through the tool-calling loop. Returns (reply_text, history)."""
    messages = history or [{"role": "system", "content": SYSTEM_PROMPT}]
    messages.append({"role": "user", "content": user_message})

    if verbose:
        print(f"👤 USER: {user_message}\n")

    for _ in range(max_turns):
        t0 = time.time()
        response = client.chat.completions.create(
            model=MERCURY_MODEL,
            reasoning_effort="low",
            messages=messages,
            tools=TOOLS,
            tool_choice="auto",
        )
        dt = time.time() - t0
        msg = response.choices[0].message

        # No tool calls => this is the final answer for this turn.
        if not msg.tool_calls:
            messages.append({"role": "assistant", "content": msg.content})
            if verbose:
                print(f"🦷 AGENT ({dt:.2f}s): {msg.content}")
            return msg.content, messages

        # Otherwise, run each requested tool and feed the results back.
        messages.append(msg.model_dump(exclude_none=True))
        for tool_call in msg.tool_calls:
            name = tool_call.function.name
            args = json.loads(tool_call.function.arguments)
            if verbose:
                print(f"   🔧 tool_call ({dt:.2f}s): {name}({json.dumps(args)})")
            try:
                result = TOOL_RUNNERS[name](**args)
            except Exception as e:                       # never crash the loop
                result = {"error": f"{type(e).__name__}: {e}"}
            if verbose:
                print(f"      ↳ {json.dumps(result)[:200]}")
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result),
            })

    return "Sorry, I couldn't complete that. Let me connect you to the front desk.", messages

print(f"Agent loop ready. Today is {TODAY.strftime('%A, %B')} {TODAY.day}, {TODAY.year}.")

5. Text Customer Service Agent

Let’s walk a caller through a full support flow. We keep history between calls so the agent remembers the conversation.

5a. Check availability

reply, history = run_agent(
    "Hi, what times are open with Dr. Flossmore on Wednesday for a cleaning?"
)

5b. Book a slot (a real database write)

reply, history = run_agent(
    "Let's do the earliest one. My name is Taylor Testcase, phone 415-555-0199.",
    history=history,
)

Let’s confirm the write actually landed in the database. The slot Taylor took should now be booked, and Taylor should exist as a patient.

print("Taylor's bookings:")
print(json.dumps(get_appointments_for_patient(conn, name="Taylor"), indent=2))

5c. Reschedule

reply, history = run_agent(
    "Actually something came up - can you move that to Friday instead?",
    history=history,
)

5d. Cancel

reply, history = run_agent(
    "You know what, I'll just cancel it for now. Thanks!",
    history=history,
)

Every one of those turns was a real read or write against the SQLite database. The agent discovered slot ids and appointment ids purely through tool calls, exactly as it would against a production scheduling system behind an API.

6. Real-time voice with ElevenLabs

Now we give the agent a voice. The pipeline is: caller audio → ElevenLabs STT → Mercury 2 (tool loop) → ElevenLabs TTS → spoken reply We use ElevenLabs scribe_v1 for transcription and eleven_turbo_v2_5 for low-latency speech. To keep the notebook self-contained (no microphone needed), we simulate the caller by synthesizing their line with a different voice, then feed that audio through the real STT → agent → TTS path.

from elevenlabs import ElevenLabs
from IPython.display import Audio, display

eleven = ElevenLabs(api_key=os.environ["ELEVENLABS_API_KEY"],
                    base_url="https://api.elevenlabs.io")

CALLER_VOICE = "21m00Tcm4TlvDq8ikWAM"   # Rachel  - stands in for the caller
AGENT_VOICE  = "IKne3meq5aSn9XLyUdCD"   # Charlie - the clinic receptionist
TTS_MODEL    = "eleven_turbo_v2_5"
STT_MODEL    = "scribe_v1"

def synth(text, voice_id, path):
    """Synthesize `text` to an mp3 file and return the raw bytes."""
    audio = eleven.text_to_speech.convert(
        voice_id=voice_id,
        model_id=TTS_MODEL,
        text=text,
        output_format="mp3_44100_128",
    )
    data = b"".join(audio)
    with open(path, "wb") as f:
        f.write(data)
    return data

print("ElevenLabs client ready.")

6a. Simulate the caller

We synthesize the caller’s spoken request. In a real deployment this audio would come from a phone line or the browser’s microphone.

caller_line = "Hi, I'd like to book a checkup with Dr. Flossmore on Wednesday afternoon."
caller_audio = synth(caller_line, CALLER_VOICE, "caller.mp3")
print(f"Simulated caller said: {caller_line!r}")
display(Audio("caller.mp3"))

6b. Transcribe with `scribe_v1`

Speech-to-text, timed. This is the first hop of the latency budget.

t0 = time.time()
transcript = eleven.speech_to_text.convert(
    model_id=STT_MODEL,
    file=open("caller.mp3", "rb"),
)
stt_latency = time.time() - t0
caller_text = transcript.text
print(f"📝 Transcript ({stt_latency:.2f}s): {caller_text}")

6c. Stream the spoken reply, sentence by sentence

The trick for low perceived latency (borrowed from the voice assistant cookbook): don’t wait for the whole answer. Buffer Mercury’s streamed text until a sentence boundary, then synthesize that sentence while the next one is still generating. Here we run the tool loop for the caller’s request, then stream the final answer into ElevenLabs TTS one sentence at a time.

import re

def speak_streaming(text_chunks, voice_id, out_path):
    """Consume an iterator of text chunks, synthesize each complete sentence as it
    arrives, and concatenate the audio into a single mp3."""
    audio_parts, buffer = [], ""
    sentence_end = re.compile(r"(.+?[.!?])(\s|$)")

    def flush(sentence):
        sentence = sentence.strip()
        if not sentence:
            return
        chunk = eleven.text_to_speech.convert(
            voice_id=voice_id, model_id=TTS_MODEL,
            text=sentence, output_format="mp3_44100_128",
        )
        audio_parts.append(b"".join(chunk))

    for piece in text_chunks:
        buffer += piece
        while True:
            m = sentence_end.match(buffer)
            if not m:
                break
            flush(m.group(1))
            buffer = buffer[m.end():]
    flush(buffer)  # trailing text with no terminal punctuation

    data = b"".join(audio_parts)
    with open(out_path, "wb") as f:
        f.write(data)
    return data

6d. Put it together: one full voice turn

voice_turn takes caller audio and returns the agent’s spoken reply, printing a latency breakdown for each stage.

def voice_turn(audio_path, history=None, agent_voice=AGENT_VOICE, out_path="agent_reply.mp3"):
    timings = {}

    # 1. Speech -> text
    t0 = time.time()
    transcript = eleven.speech_to_text.convert(model_id=STT_MODEL, file=open(audio_path, "rb"))
    caller_text = transcript.text
    timings["stt"] = time.time() - t0

    # 2. Run the tool-calling agent (may make several Mercury calls)
    t0 = time.time()
    reply_text, history = run_agent(caller_text, history=history, verbose=False)
    timings["agent"] = time.time() - t0

    # 3. Text -> streamed speech (synthesize sentence by sentence)
    t0 = time.time()
    speak_streaming(iter([reply_text]), agent_voice, out_path)
    timings["tts"] = time.time() - t0

    print(f"👤 Caller: {caller_text}")
    print(f"🦷 Agent:  {reply_text}\n")
    print("⏱  latency  "
          f"STT {timings['stt']:.2f}s | agent {timings['agent']:.2f}s | "
          f"TTS {timings['tts']:.2f}s | total {sum(timings.values()):.2f}s")
    return history, out_path


history_voice, reply_path = voice_turn("caller.mp3")
display(Audio(reply_path))

6e. Keep the conversation going

Because voice_turn threads history back through, the agent remembers context across spoken turns. Let’s confirm the booking out loud.

follow_up = synth(
    "Yes please book it. I'm Dana Decoy, my number is 415-555-0288.",
    CALLER_VOICE, "caller2.mp3",
)
history_voice, reply_path = voice_turn("caller2.mp3", history=history_voice)
display(Audio(reply_path))

Tips

Getting tool calls right

Put your data’s conventions where Mercury can read it. The model can only match your schema if you tell it what your schema looks like, either in tool descriptions or in the system prompt. For example, the line “The clinic’s providers go by Molar (general dentistry), Wyre (orthodontics), and Flossmore (hygiene and cleaning)” is what makes Mercury pass Flossmore (not Dr. Pearl Flossmore) when a caller asks for “Doctor Flossmore.” The same goes for id formats, enum values, and date formats: if a tool wants 2026-08-19, say so in the tool’s description, not just as {"type": "string"} in the schema. The description fields are prompt text, and they are where the model looks when filling arguments.
Then make the tool tolerant anyway. Prompt guidance reduces malformed arguments; it never eliminates them, and speech-to-text will always run the risk of transcribing speech incorrectly, especially names. That is why _provider_id_from_name strips titles and matches on distinctive words. State the contract in the prompt, enforce it in the tool, and return a structured {"error": ...} the model can read and recover from.

Model sampling

Adjust reasoning effort for your use case. For real-time voice use cases, we recommend reasoning effort low, which runs at ultra-low latencies while still delivering a pleasant conversational experience and the complex tool calls needed for a streamlined customer service agent.

Conversation state

Manage context length as history grows. Every turn appends the full text response and tool call JSON, increasing prompt size and driving up latency for very long-running conversations. However, most of that history goes stale fast. In order to improve latency over very long conversations, consider incorporating a compaction step in your voice loop, using Mercury-2 to summarize your conversation history, maintaining a smooth conversation experience while saving on tokens and latency.
Validate every model-supplied id against the database. The write tools here check state before mutating: you cannot book a taken slot or cancel an appointment that does not exist. Keep that discipline in production: treat slot ids and appointment ids from the model as untrusted input, because a plausible-looking hallucinated id could silently corrupt the loop.

Getting Started

API reference

Capabilities

Resources

Support

Voice Customer Support Agent: Mercury + Tool Calling + Voice

Run the Customer Support Agent

Architecture

1. Setup

2. A mock clinic database

3. Define the support tools

Tool schemas

4. The agent loop

5. Text Customer Service Agent

5a. Check availability

5b. Book a slot (a real database write)

5c. Reschedule

5d. Cancel

6. Real-time voice with ElevenLabs

6a. Simulate the caller

6b. Transcribe with `scribe_v1`

6c. Stream the spoken reply, sentence by sentence

6d. Put it together: one full voice turn

6e. Keep the conversation going

Tips

Getting tool calls right

Model sampling

Conversation state

Resources

Run the Customer Support Agent

​Architecture

​1. Setup

​2. A mock clinic database

​3. Define the support tools

​Tool schemas

​4. The agent loop

​5. Text Customer Service Agent

​5a. Check availability

​5b. Book a slot (a real database write)

​5c. Reschedule

​5d. Cancel

​6. Real-time voice with ElevenLabs

​6a. Simulate the caller

​6b. Transcribe with scribe_v1

​6c. Stream the spoken reply, sentence by sentence

​6d. Put it together: one full voice turn

​6e. Keep the conversation going

​Tips

​Getting tool calls right

​Model sampling

​Conversation state

​Resources

Architecture

1. Setup

2. A mock clinic database

3. Define the support tools

Tool schemas

4. The agent loop

5. Text Customer Service Agent

5a. Check availability

5b. Book a slot (a real database write)

5c. Reschedule

5d. Cancel

6. Real-time voice with ElevenLabs

6a. Simulate the caller

6b. Transcribe with `scribe_v1`

6c. Stream the spoken reply, sentence by sentence

6d. Put it together: one full voice turn

6e. Keep the conversation going

Tips

Getting tool calls right

Model sampling

Conversation state

Resources