Hand off from AI to a human agent

VoiceAI-to-human handoff with context

Store the AI agent's notes by call ID, transfer the caller to a queue, and load those notes for the human agent.

Also called AI escalation to a live agent, warm transfer from AI

queuetransfer

The claim

Three documented pieces, joined by one id. The SWAIG tool webhook, documented in the REST spec as ai-swaig-tool-webhook, carries the call’s call_id, so the handler can key its notes by it. enter_queue puts the caller in a named queue. The bundled schema requires queue_name and transfer_after_bridge on it. It describes the verb as placing the call where “it will wait to be connected to an available agent or resource”. The vendored REST spec’s GET /api/relay/rest/queues/{queue_id}/members/next “Retrieves the next member in the queue without dequeuing”, and the member carries call_id, position and wait_time. The human takes the call with a connect whose to is queue:support, one of the forms the schema lists for connect.to.

How it works

def hand_off(self, args, raw_data):
    call_id = raw_data["call_id"]
    save_note(call_id, {"caller_name": args["caller_name"], "issue": args["issue"],
                        "from": raw_data.get("caller_id_num")})
    result = FunctionResult("Thanks. I am putting you through to a person now.")
    result.action.append({"SWML": ENQUEUE, "transfer": "true"})   # the documented shape
    return result

def brief(queue_id):
    member = client.queues.get_next_member(queue_id)
    return {"call_id": member["call_id"], ..., "notes": load_notes().get(member["call_id"])}

What the platform receives from the tool, then what the human’s phone runs:

{"response": "Thanks. I am putting you through to a person now.",
 "action": [{"SWML": {"version": "1.0.0", "sections": {"main": [
               {"enter_queue": {"queue_name": "support", "transfer_after_bridge": "false"}}]}},
             "transfer": "true"}]}

{"version": "1.0.0", "sections": {"main": [{"answer": {}}, {"connect": {"to": "queue:support"}}]}}

transfer_after_bridge is a string the schema requires; "false" means carry on in this document after the bridge, and there is nothing after it. The transfer: "true" beside the SWML is what the tool webhook documents for a call that leaves the agent. FunctionResult.execute_swml(transfer=True) would put that flag inside the document, so the action is built by hand. The notes go to a JSON file at NOTES_PATH. The agent and the screen are two processes, so a dictionary in the agent would be empty in the shell that runs brief. Swap the two functions for your database. The screen asks for the next member, takes its call_id, and shows the notes filed under it.

Limitations

You prove the documents, the tool result and the requests. Who the platform bridges to whom, and how long the caller waits, are the platform’s side of a live call. The next-member read does not dequeue; the bridge does. The two are not one step. Between the read and the bridge the front of the queue can change, and the screen would then describe a different caller than the one bridged. Correlate by call_id on the bridged call before you act on the notes.

The notes are a local JSON file the two processes share, written whole and swapped in. It is not a store for concurrent hand-offs or for production; put them in a database keyed by call_id before anything depends on it.

What to change first

Key the note by args["caller_name"] instead of call_id and run the verifier. The brief assertion fails, because the queue member carries a call id and no name. The id is the only field both sides hold.

Where this sits

Before this

Route a call to an AI agent Queue a call until an agent is free

Write call dispositions from data your handlers recorded

python/app.py

"""Hand off from AI to a human agent.

Two halves and one key. The agent's `hand_off` tool writes what it learned
under the call's id, then returns a SWML action whose one verb is
`enter_queue`. The bundled schema requires `queue_name` and
`transfer_after_bridge` on it. It describes the verb as placing the call in a
named queue "where it will wait to be connected to an available agent or
resource". The human's side asks the vendored REST spec's
`GET /api/relay/rest/queues/{queue_id}/members/next` for the next member.
Its documented fields include `call_id`, and the notes are looked up by it.
The human takes the call with a `connect` whose `to` is `queue:<name>`, a form
the bundled schema lists for `connect.to`.

The call id is the join. The SWAIG tool webhook carries it as `call_id`, the
queue member carries it as `call_id`, and nothing else has to agree.

Written against signalwire-sdk 3.0.1 (AgentBase, FunctionResult,
RestClient.queues, SWMLService).
"""
import json
import os
from pathlib import Path

from dotenv import load_dotenv
from signalwire import AgentBase, FunctionResult, SWMLService
from signalwire.rest import RestClient

# the SDK does not read .env for you
load_dotenv()

# RestClient() reads SIGNALWIRE_PROJECT_ID / SIGNALWIRE_API_TOKEN /
# SIGNALWIRE_SPACE from the environment (signalwire/rest/client.py).
client = RestClient()

QUEUE = os.getenv("QUEUE_NAME", "support")
# the queue's id, for the human's screen: the REST list of queues carries it
QUEUE_ID = os.getenv("QUEUE_ID")

# call_id -> what the agent learned before it handed off. The agent and the
# human's screen are two processes (python app.py, then python app.py brief),
# so the notes live in a file both can open; swap the two functions for your
# database. A dictionary here would be empty in the second process.
NOTES_PATH = Path(os.getenv("NOTES_PATH", "handoff-notes.json"))


def save_note(call_id, note):
    notes = load_notes()
    notes[call_id] = note
    # write whole, then swap in: a reader never sees a half-written file
    tmp = NOTES_PATH.with_suffix(".tmp")
    tmp.write_text(json.dumps(notes, indent=2), encoding="utf-8")
    os.replace(tmp, NOTES_PATH)


def load_notes():
    if not NOTES_PATH.exists():
        return {}
    return json.loads(NOTES_PATH.read_text(encoding="utf-8"))

# enter_queue: the caller waits here; "false" means carry on in this document
# after the bridge, and there is nothing after it, so the call ends with the bridge
ENQUEUE = {"version": "1.0.0", "sections": {"main": [
    {"enter_queue": {"queue_name": QUEUE, "transfer_after_bridge": "false"}}]}}


class TriageAgent(AgentBase):
    def __init__(self):
        super().__init__(name="triage", route="/triage")
        self.prompt_add_section(
            "Role",
            "You are the front desk at Ridgeline Cycles. Find out who is calling and "
            "what the problem is in one or two questions. When you have both, or when "
            "the caller asks for a person, call hand_off with what you learned.")

    @AgentBase.tool(
        name="hand_off",
        description=("Hand the caller to a person, with a short note of who they are "
                     "and why they called."),
        parameters={
            "type": "object",
            "properties": {
                "caller_name": {"type": "string",
                                "description": "The caller's name as they gave it."},
                "issue": {"type": "string",
                          "description": "One sentence on why they called."},
            },
            "required": ["caller_name", "issue"],
        },
    )
    def hand_off(self, args, raw_data):
        call_id = raw_data["call_id"]
        save_note(call_id, {
            "caller_name": args["caller_name"],
            "issue": args["issue"],
            "from": raw_data.get("caller_id_num"),
        })
        result = FunctionResult("Thanks. I am putting you through to a person now.")
        # the documented action: the SWML, and a sibling transfer flag so the
        # call leaves the agent for the queue (execute_swml(transfer=True) puts
        # the flag inside the document instead)
        result.action.append({"SWML": ENQUEUE, "transfer": "true"})
        return result


def brief(queue_id=None):
    """What the human's screen shows: the next caller and the agent's notes."""
    queue_id = queue_id or QUEUE_ID
    if not queue_id:
        raise SystemExit("QUEUE_ID is required for brief; see .env.example")
    member = client.queues.get_next_member(queue_id)
    call_id = member["call_id"]
    return {"call_id": call_id, "position": member.get("position"),
            "waiting_seconds": member.get("wait_time"),
            "notes": load_notes().get(call_id)}


def take(service=None):
    """The document the human's phone runs to answer the front of the queue."""
    service = service or SWMLService(name="take", route="/take")
    service.add_verb("answer", {})
    service.add_verb("connect", {"to": f"queue:{QUEUE}"})
    return service


agent = TriageAgent()

if __name__ == "__main__":
    import sys

    if len(sys.argv) == 2 and sys.argv[1] == "brief":
        print(json.dumps(brief(), indent=2))
    elif len(sys.argv) == 2 and sys.argv[1] == "take":
        # the human's document, served behind the same basic-auth pair
        take().serve(port=int(os.getenv("TAKE_PORT", "3001")))
    else:
        agent.run()

Run it

cd python
pip install -r requirements.txt
cp ../.env.example .env          # then edit .env: credentials, the basic-auth pair, QUEUE_NAME, QUEUE_ID
python app.py                    # the agent, on port 3000
python app.py take               # in another shell: the human's document, on port 3001
python app.py brief              # in a third: who is next, and what the agent learned

QUEUE_ID is the id of the queue named in QUEUE_NAME. enter_queue creates the queue by name on first use, and GET /api/relay/rest/queues lists it with its id. Point the human’s number at the take document, behind the same basic-auth pair.

The webhook needs a public HTTPS URL. For a local run, expose port 3000 with a tunnel such as ngrok and use that hostname. Point the shop’s number at https://<user>:<password>@<your-host>/triage/. The schema says a queue that does not exist is created on first use, so the first hand-off creates support. The take command serves the human’s document on port 3001, behind the same basic-auth pair. Expose it the same way and point the human’s number at it. When the human dials in, the platform bridges the front of the queue.

Verify it

No network, no account.

cd ..                     # back to the recipe folder
python verify.py

The verifier executes the tool offline, swaps the SDK’s HTTP layer for a recorder, and asserts the following.

the rendered agent has one function, hand_off, requiring caller_name and issue
executing it with a tool webhook body that carries call_id and caller_id_num returns the exact response and one action
that action’s SWML validates, holds exactly one enter_queue with queue_name: support and transfer_after_bridge: "false", and carries transfer: "true"
the notes sit under the call id with the name, the issue and the number, and nothing else
the notes land in the file at NOTES_PATH, and brief, run after the module is reloaded as a second process would load it, reads them back from there
brief makes one GET of the documented next-member path for the configured QUEUE_ID, with no body or query, and returns the member’s position, wait and the notes for its call_id
the spec documents call_id, position and wait_time on the member and describes the read as “without dequeuing”
the human’s document validates and connects to queue:support; the schema requires queue_name and transfer_after_bridge and lists the queue: form for connect.to

Report an issue