Warm Transfers Using Twilio Programmable Voice

A warm transfer briefs a human representative before the caller is connected. In this guide, you'll build a Twilio integration where:

  1. An inbound agent answers the call and collects the caller's reason for calling.
  2. While the inbound agent keeps the caller company, an outbound agent calls a customer service representative and briefs them.
  3. Once the representative is ready, both calls are transferred into a shared Twilio conference.
New to Guava + Twilio? Start with the Twilio Programmable Voice / TwiML guide for the basics of routing Twilio calls to a Guava agent.

Prerequisites

  • Two Guava SIP codes, one for the inbound agent and one for the outbound agent. Create them on the SIP page in the Guava dashboard.
  • A Twilio phone number, and a Twilio API key.
  • A public URL that Twilio can reach for webhooks (e.g. using ngrok during development).

Install dependencies:

pip install guava-sdk twilio "fastapi[standard]"

Configure your Twilio number

In the Twilio Console, go to Phone Numbers > Manage > Active Numbers and select your number. Under Voice Configuration, set A call comes in to Webhook with the URL https://<your-public-url>/inbound-call and method HTTP POST. Then click Save configuration.

Inbound call reaches the inbound agent

When a call comes in, Twilio requests /inbound-call. This returns a Dial command telling Twilio to transfer to the inbound agent's SIP code. The referUrl attribute tells Twilio to send any later transfers from this agent to /refer.

main.py
@app.post("/inbound-call")
def voice():
  return Response(
      f"""<Response>
<Dial referUrl="{PUBLIC_URL}/refer" referMethod="POST">
  <Sip>{INBOUND_AGENT_SIP_CODE}@sip.goguava.ai</Sip>
</Dial>
</Response>""",
      media_type="application/xml",
  )

Inbound agent collects the reason and stalls

The inbound agent collects reason_for_calling. Once that task is complete, the agent is given a new task: keep the caller on the line while a representative is found.

main.py
@agent.on_call_start
def on_call_start(call: guava.Call):
  call.set_task(task_id="collect_reason", checklist=[
      guava.Field(key="reason_for_calling")
  ])

@agent.on_task_complete("collect_reason")
def on_collect_reason(call: guava.Call):
  call.set_task("stall_customer", """Inform the caller that you're checking if a customer service
                  representative is ready to help them. Stay with the caller on the
                  line until the expert notifies you that the representative is ready.""")
  ...

At the same time, the handler generates a random conference ID, stores the inbound call in WARM_TRANSFERS under that ID, and uses the Twilio API to start an outbound call to the representative. When the representative answers, Twilio connects them to the outbound agent's SIP code. The reason for calling and the conference ID are passed through to the outbound agent as X- SIP headers.

main.py
from xml.sax.saxutils import escape
from guava.sip import SipUri

conference_id = secrets.token_hex(10)

WARM_TRANSFERS[conference_id] = call

# Twilio only forwards custom SIP headers prefixed with X-.
outbound_agent_uri = SipUri(
  user=OUTBOUND_AGENT_SIP_CODE,
  host="sip.goguava.ai",
  headers={
      "X-Call-Reason": call.get_field("reason_for_calling"),
      "X-Conference-Id": conference_id,
  },
)

twilio_client.calls.create(
  to=REPRESENTATIVE_NUMBER,
  from_=TWILIO_NUMBER,
  # XML-escape the URI, since the & between headers is invalid in TwiML.
  twiml=f"""<Response>
  <Dial referUrl="{PUBLIC_URL}/refer" referMethod="POST">
  <Sip>{escape(str(outbound_agent_uri))}</Sip>
  </Dial>
  </Response>""",
)
Note: Twilio only forwards custom SIP headers prefixed with X-. SipUri URL-encodes the header values, but the URI still has to be XML-escaped in TwiML.

Outbound agent briefs the representative

The outbound agent reads X-Call-Reason from call.call_info.sip_headers, explains the caller's issue to the representative, and confirms that they're ready to take the call.

main.py
@outbound_agent.on_call_start
def on_outbound_call_start(call: guava.Call):
  assert isinstance(call.call_info, SipCallInfo)
  call_reason = call.call_info.sip_headers.get("X-Call-Reason")

  call.set_task("confirm_ready", f"""A caller is calling about {call_reason}. You are not talking to the caller - you are talking to a customer service representative.""", checklist=[
      "Inform the customer service agent of the caller's reason for calling.",
      "Confirm that the service agent is ready for a transfer."
  ])

Both calls join the conference

Once the representative is ready, the outbound agent looks up the waiting inbound call by X-Conference-Id and transfers both calls to sip:<conference_id>@conference.internal. Each agent says a short message before it transfers.

main.py
@outbound_agent.on_task_complete("confirm_ready")
def on_confirm_ready(call: guava.Call):
  assert isinstance(call.call_info, SipCallInfo)
  conference_id = call.call_info.sip_headers.get("X-Conference-Id")

  assert conference_id
  other_call = WARM_TRANSFERS[conference_id]

  destination = f"sip:{conference_id}@conference.internal"

  other_call.transfer(destination, "The customer service representative is ready.")
  call.transfer(destination, "Transfer to the caller. Tell the rep that you are 'Patching in the caller.'")

Twilio sends each transfer to /refer with the target in ReferTransferTarget. The handler extracts the conference ID and returns <Conference> TwiML, which places the caller and the representative in the same room. endConferenceOnExit="true" ends the conference when either party hangs up.

main.py
@app.post("/refer")
async def refer(request: Request):
  form = await request.form()

  target = str(form["ReferTransferTarget"]).strip("<>")
  conference_id = target.removeprefix("sip:").split("@", 1)[0]

  assert conference_id

  return Response(
      f"""<Response>
<Dial>
  <Conference beep="true" endConferenceOnExit="true">{conference_id}</Conference>
</Dial>
</Response>""",
      media_type="application/xml",
  )
Note: WARM_TRANSFERS is an in-memory dict, so the inbound and outbound agents must run in the same process.

Run the example

Save the full example below as main.py, then set your Twilio credentials and start the server:

export TWILIO_ACCOUNT_SID=ACxxx
export TWILIO_API_KEY_SID=SKxxx
export TWILIO_API_KEY_SECRET=xxx
python main.py

Call your Twilio number. After you give a reason for calling, the representative's phone will ring. Once they've been briefed, you'll both be connected in a conference.

Full example

main.py
import os
import guava
import uvicorn
import threading
import secrets
import logging

from xml.sax.saxutils import escape
from guava import logging_utils
from guava.sip import SipUri
from guava.types.call_info import SipCallInfo
from fastapi import FastAPI, Request, Response
from twilio.rest import Client

INBOUND_AGENT_SIP_CODE = "guavasip-xxx"  # Replace with your inbound SIP code.
OUTBOUND_AGENT_SIP_CODE = "guavasip-yyy"  # Replace with your outbound SIP code.
PUBLIC_URL = "https://your-public-url.example.com"  # Must be reachable by Twilio.
REPRESENTATIVE_NUMBER = "+1..."  # The customer service representative's number.
TWILIO_NUMBER = "+1..."  # Your owned Twilio number.

# Maps conference IDs to the inbound call waiting to be transferred.
WARM_TRANSFERS: dict[str, guava.Call] = {}

logger = logging.getLogger("warm_transfer_example")

twilio_client = Client(
  os.environ["TWILIO_API_KEY_SID"],
  os.environ["TWILIO_API_KEY_SECRET"],
  os.environ["TWILIO_ACCOUNT_SID"],
)

agent = guava.Agent(
  name="Nova",
  organization="Acme Corp",
)

@agent.on_call_start
def on_call_start(call: guava.Call):
  assert isinstance(call.call_info, SipCallInfo)
  logger.info("Inbound call SIP headers: %r", call.call_info.sip_headers)
  call.set_task(task_id="collect_reason", checklist=[
      guava.Field(key="reason_for_calling")
  ])

@agent.on_task_complete("collect_reason")
def on_collect_reason(call: guava.Call):
  call.set_task("stall_customer", """Inform the caller that you're checking if a customer service
                  representative is ready to help them. Stay with the caller on the
                  line until the expert notifies you that the representative is ready.""")

  conference_id = secrets.token_hex(10)

  WARM_TRANSFERS[conference_id] = call

  # Twilio only forwards custom SIP headers prefixed with X-.
  outbound_agent_uri = SipUri(
      user=OUTBOUND_AGENT_SIP_CODE,
      host="sip.goguava.ai",
      headers={
          "X-Call-Reason": call.get_field("reason_for_calling"),
          "X-Conference-Id": conference_id,
      },
  )

  twilio_client.calls.create(
      to=REPRESENTATIVE_NUMBER,
      from_=TWILIO_NUMBER,
      # XML-escape the URI, since the & between headers is invalid in TwiML.
      twiml=f"""<Response>
      <Dial referUrl="{PUBLIC_URL}/refer" referMethod="POST">
      <Sip>{escape(str(outbound_agent_uri))}</Sip>
      </Dial>
      </Response>""",
  )

outbound_agent = guava.Agent(
  name="Nova",
  organization="Acme Corp",
  purpose="You are an outbound calling agent. You talk to customer service representatives on behalf of callers and brief them before a warm transfer.",
)

@outbound_agent.on_call_start
def on_outbound_call_start(call: guava.Call):
  assert isinstance(call.call_info, SipCallInfo)
  logger.info("Outbound call SIP headers: %r", call.call_info.sip_headers)
  call_reason = call.call_info.sip_headers.get("X-Call-Reason")

  call.set_task("confirm_ready", f"""A caller is calling about {call_reason}. You are not talking to the caller - you are talking to a customer service representative.""", checklist=[
      "Inform the customer service agent of the caller's reason for calling.",
      "Confirm that the service agent is ready for a transfer."
  ])

@outbound_agent.on_task_complete("confirm_ready")
def on_confirm_ready(call: guava.Call):
  assert isinstance(call.call_info, SipCallInfo)
  conference_id = call.call_info.sip_headers.get("X-Conference-Id")

  assert conference_id
  other_call = WARM_TRANSFERS[conference_id]

  destination = f"sip:{conference_id}@conference.internal"

  other_call.transfer(destination, "The customer service representative is ready.")
  call.transfer(destination, "Transfer to the caller. Tell the rep that you are 'Patching in the caller.'")

app = FastAPI()

@app.post("/inbound-call")
def voice():
  return Response(
      f"""<Response>
<Dial referUrl="{PUBLIC_URL}/refer" referMethod="POST">
  <Sip>{INBOUND_AGENT_SIP_CODE}@sip.goguava.ai</Sip>
</Dial>
</Response>""",
      media_type="application/xml",
  )

@app.post("/refer")
async def refer(request: Request):
  form = await request.form()
  logger.info("Twilio REFER webhook: %r", dict(form))

  target = str(form["ReferTransferTarget"]).strip("<>")
  conference_id = target.removeprefix("sip:").split("@", 1)[0]

  assert conference_id

  return Response(
      f"""<Response>
<Dial>
  <Conference beep="true" endConferenceOnExit="true">{conference_id}</Conference>
</Dial>
</Response>""",
      media_type="application/xml",
  )

if __name__ == "__main__":
  logging_utils.configure_logging()

  threading.Thread(target=agent.listen_sip, args=(INBOUND_AGENT_SIP_CODE,), daemon=True).start()
  threading.Thread(target=outbound_agent.listen_sip, args=(OUTBOUND_AGENT_SIP_CODE,), daemon=True).start()
  uvicorn.run(app, host="0.0.0.0", port=3000)
→ Next: Cisco CUBE / CUCM

Questions? hi@goguava.ai