apps/opik-documentation/documentation/fern/docs-v2/integrations/twilio-agent-connect.mdx
Twilio Agent Connect (TAC) is a framework for building AI agents that answer real phone calls. It handles the telephony, speech-to-text and text-to-speech, and calls your handler with the caller's transcribed message.
This guide explains how to log your voice agent to Opik. You will trace every caller turn, group the turns of a call into a single conversation, and attach the call recording so you can listen to what the caller heard.
Comet provides a hosted version of the Opik platform, simply create an account and grab your API Key.
You can also run the Opik platform locally, see the installation guide for more information.
Install the opik and twilio-agent-connect packages:
pip install opik twilio-agent-connect
Configure the Opik Python SDK for your deployment type. See the Python SDK Configuration guide for detailed instructions on:
opik configureopik.configure()Twilio Agent Connect calls your handler once for every caller turn. Add the @opik.track decorator to that handler and each turn becomes a trace.
Set thread_id to the conversation id so that every turn of the same phone call is grouped into a single Opik thread:
import opik
from opik import opik_context
from tac import TAC, TACConfig
from tac.models.session import ConversationSession
from tac.models.tac import TACMemoryResponse
tac = TAC(TACConfig(
account_sid=TWILIO_ACCOUNT_SID,
auth_token=TWILIO_AUTH_TOKEN,
api_key=TWILIO_API_KEY,
api_secret=TWILIO_API_SECRET,
phone_number=TWILIO_PHONE_NUMBER,
))
@opik.track(name="voice-agent-turn", project_name="my-voice-agent")
async def handle_message_ready(
message: str,
context: ConversationSession,
memory: TACMemoryResponse | None,
) -> str:
# Group every turn of this call into one Opik thread
opik_context.update_current_trace(thread_id=context.conversation_id)
response = await run_my_agent(message, context, memory)
return response
tac.on_message_ready(handle_message_ready)
That is the whole tracing setup. One caller turn is one trace, and one phone call is one thread.
You can also log useful call details as metadata:
opik_context.update_current_trace(
thread_id=context.conversation_id,
metadata={
"channel": context.channel,
"profile_id": context.profile_id,
},
)
A voice agent usually does several things before it answers: it looks up the caller, searches a knowledge base, calls a few tools, then calls an LLM. Add @opik.track to each of those functions and they appear as nested spans under the turn's trace, so you can see where the time went and what each step returned.
Use the type argument to tell Opik what kind of step it is. Opik shows llm and tool spans differently, and llm spans are where token usage and cost appear.
@opik.track(name="knowledge-search", type="general")
def search_knowledge_base(query: str) -> list[str]:
return my_vector_store.search(query)
@opik.track(name="account-lookup", type="tool")
def account_lookup(phone: str) -> dict:
return billing_api.get_account(phone)
@opik.track(name="generate-reply", type="llm")
def generate_reply(messages: list[dict]) -> str:
...
@opik.track(name="voice-agent-turn")
async def handle_message_ready(message, context, memory) -> str:
opik_context.update_current_trace(thread_id=context.conversation_id)
# Each call below becomes a span on this turn's trace
docs = search_knowledge_base(message)
account = account_lookup(context.author_info.address)
return generate_reply(build_messages(message, docs, account))
Inside a tracked function, use update_current_span to record anything the arguments and return value do not already capture:
from opik import opik_context
@opik.track(name="knowledge-search", type="general")
def search_knowledge_base(query: str) -> list[str]:
results = my_vector_store.search(query)
opik_context.update_current_span(
metadata={"index": "support-articles", "top_k": 5},
output={"documents": results, "hit_count": len(results)},
)
return results
If you call an LLM provider directly, use the matching Opik integration instead of writing your own llm span. Token usage, cost and model parameters are then logged for you, and the spans still nest under the current turn:
from opik.integrations.openai import track_openai
from openai import OpenAI
client = track_openai(OpenAI())
@opik.track(name="voice-agent-turn")
async def handle_message_ready(message, context, memory) -> str:
opik_context.update_current_trace(thread_id=context.conversation_id)
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=build_messages(message, context),
)
return response.choices[0].message.content
See the OpenAI, Anthropic and Gemini integrations, or browse all integrations for the framework you use.
A phone call is a live, unattended channel, so it is a good place to check what the caller says and what the agent is about to say. Opik guardrails run in the same request, and each check appears on the turn's trace with its own pass or fail result.
from opik import exceptions
from opik.guardrails import Guardrail, PII, PromptInjection
inbound = Guardrail(guards=[PromptInjection(threshold=0.5), PII()])
@opik.track(name="voice-agent-turn")
async def handle_message_ready(message, context, memory) -> str:
opik_context.update_current_trace(thread_id=context.conversation_id)
try:
inbound.validate(message)
except exceptions.GuardrailValidationFailed:
return "Sorry, I can't help with that. Let me put you through to a colleague."
return await run_my_agent(message, context, memory)
See the Guardrails overview for the available checks, and Custom guardrails if you need one trained on your own policy.
Your handler receives text, because Twilio performs the speech-to-text and text-to-speech. To listen to a conversation in Opik, record the call with Twilio and attach the audio to your thread. Opik plays audio attachments inline.
Twilio Agent Connect lets you run code when an inbound call arrives. Use that hook to start a recording, and tell Twilio where to notify you when the audio is ready:
from twilio.rest import Client
from tac.models.voice import TwiMLOptions, TwiMLRequest
twilio = Client(TWILIO_API_KEY, TWILIO_API_SECRET, TWILIO_ACCOUNT_SID)
async def start_recording(request: TwiMLRequest) -> TwiMLOptions:
twilio.calls(request.call_sid).recordings.create(
recording_channels="dual",
recording_status_callback=f"{PUBLIC_URL}/recording",
)
return TwiMLOptions()
voice_channel.on_inbound_call_twiml(start_recording)
recording_channels="dual" keeps the caller and the agent on separate audio channels.
When the call ends, Twilio calls your webhook with a link to the audio. Download it and attach it to a trace in the same thread:
import httpx
import opik
from opik import Attachment
opik_client = opik.Opik(project_name="my-voice-agent")
@app.post("/recording")
async def recording_ready(request: Request):
form = await request.form()
call_sid = form["CallSid"]
audio = httpx.get(
f"{form['RecordingUrl']}.mp3",
auth=(TWILIO_API_KEY, TWILIO_API_SECRET),
)
opik_client.trace(
name="call-recording",
thread_id=call_sid,
output={"duration_s": int(form["RecordingDuration"])},
attachments=[
Attachment(
data=audio.content,
file_name=f"call-{call_sid}.mp3",
content_type="audio/mpeg",
)
],
)
opik_client.flush()
Twilio uses the call SID as the conversation id, so the recording lands in the same Opik thread as the turns of that call. Open the thread in Opik and you can read the transcript and play the audio side by side.
<Note> Call recording is subject to local laws on consent. Check the rules that apply where your callers are before you enable it. </Note>With this setup, Opik records:
If you have any questions or suggestions for improving the Twilio Agent Connect integration, please open an issue on our GitHub repository.