Observability for Twilio Agent Connect with Opik

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.

Account Setup

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.

Getting Started

Installation

Install the opik and twilio-agent-connect packages:

$
pip install opik twilio-agent-connect

Configuring Opik

Configure the Opik Python SDK for your deployment type. See the Python SDK Configuration guide for detailed instructions on:

• CLI configuration: opik configure
• Code configuration: opik.configure()
• Self-hosted vs Cloud vs Enterprise setup
• Configuration files and environment variables

Logging traces

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:

1
import opik
2
from opik import opik_context
3
from tac import TAC, TACConfig
4
from tac.models.session import ConversationSession
5
from tac.models.tac import TACMemoryResponse
6
7
tac = TAC(TACConfig(
8
    account_sid=TWILIO_ACCOUNT_SID,
9
    auth_token=TWILIO_AUTH_TOKEN,
10
    api_key=TWILIO_API_KEY,
11
    api_secret=TWILIO_API_SECRET,
12
    phone_number=TWILIO_PHONE_NUMBER,
13
))
14
15
16
@opik.track(name="voice-agent-turn", project_name="my-voice-agent")
17
async def handle_message_ready(
18
    message: str,
19
    context: ConversationSession,
20
    memory: TACMemoryResponse | None,
21
) -> str:
22
    # Group every turn of this call into one Opik thread
23
    opik_context.update_current_trace(thread_id=context.conversation_id)
24
25
    response = await run_my_agent(message, context, memory)
26
    return response
27
28
29
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:

1
opik_context.update_current_trace(
2
    thread_id=context.conversation_id,
3
    metadata={
4
        "channel": context.channel,
5
        "profile_id": context.profile_id,
6
    },
7
)

Adding spans to a turn

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.

1
@opik.track(name="knowledge-search", type="general")
2
def search_knowledge_base(query: str) -> list[str]:
3
    return my_vector_store.search(query)
4
5
6
@opik.track(name="account-lookup", type="tool")
7
def account_lookup(phone: str) -> dict:
8
    return billing_api.get_account(phone)
9
10
11
@opik.track(name="generate-reply", type="llm")
12
def generate_reply(messages: list[dict]) -> str:
13
    ...
14
15
16
@opik.track(name="voice-agent-turn")
17
async def handle_message_ready(message, context, memory) -> str:
18
    opik_context.update_current_trace(thread_id=context.conversation_id)
19
20
    # Each call below becomes a span on this turn's trace
21
    docs = search_knowledge_base(message)
22
    account = account_lookup(context.author_info.address)
23
    return generate_reply(build_messages(message, docs, account))

Adding detail to a span

Inside a tracked function, use update_current_span to record anything the arguments and return value do not already capture:

1
from opik import opik_context
2
3
4
@opik.track(name="knowledge-search", type="general")
5
def search_knowledge_base(query: str) -> list[str]:
6
    results = my_vector_store.search(query)
7
8
    opik_context.update_current_span(
9
        metadata={"index": "support-articles", "top_k": 5},
10
        output={"documents": results, "hit_count": len(results)},
11
    )
12
    return results

Tracking LLM calls automatically

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:

1
from opik.integrations.openai import track_openai
2
from openai import OpenAI
3
4
client = track_openai(OpenAI())
5
6
7
@opik.track(name="voice-agent-turn")
8
async def handle_message_ready(message, context, memory) -> str:
9
    opik_context.update_current_trace(thread_id=context.conversation_id)
10
11
    response = client.chat.completions.create(
12
        model="gpt-4o-mini",
13
        messages=build_messages(message, context),
14
    )
15
    return response.choices[0].message.content

See the OpenAI, Anthropic and Gemini integrations, or browse all integrations for the framework you use.

Adding guardrails

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.

1
from opik import exceptions
2
from opik.guardrails import Guardrail, PII, PromptInjection
3
4
inbound = Guardrail(guards=[PromptInjection(threshold=0.5), PII()])
5
6
7
@opik.track(name="voice-agent-turn")
8
async def handle_message_ready(message, context, memory) -> str:
9
    opik_context.update_current_trace(thread_id=context.conversation_id)
10
11
    try:
12
        inbound.validate(message)
13
    except exceptions.GuardrailValidationFailed:
14
        return "Sorry, I can't help with that. Let me put you through to a colleague."
15
16
    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.

Logging call recordings

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.

Step 1: Start the recording

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:

1
from twilio.rest import Client
2
from tac.models.voice import TwiMLOptions, TwiMLRequest
3
4
twilio = Client(TWILIO_API_KEY, TWILIO_API_SECRET, TWILIO_ACCOUNT_SID)
5
6
7
async def start_recording(request: TwiMLRequest) -> TwiMLOptions:
8
    twilio.calls(request.call_sid).recordings.create(
9
        recording_channels="dual",
10
        recording_status_callback=f"{PUBLIC_URL}/recording",
11
    )
12
    return TwiMLOptions()
13
14
15
voice_channel.on_inbound_call_twiml(start_recording)

recording_channels="dual" keeps the caller and the agent on separate audio channels.

Step 2: Attach the recording

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:

1
import httpx
2
import opik
3
from opik import Attachment
4
5
opik_client = opik.Opik(project_name="my-voice-agent")
6
7
8
@app.post("/recording")
9
async def recording_ready(request: Request):
10
    form = await request.form()
11
    call_sid = form["CallSid"]
12
13
    audio = httpx.get(
14
        f"{form['RecordingUrl']}.mp3",
15
        auth=(TWILIO_API_KEY, TWILIO_API_SECRET),
16
    )
17
18
    opik_client.trace(
19
        name="call-recording",
20
        thread_id=call_sid,
21
        output={"duration_s": int(form["RecordingDuration"])},
22
        attachments=[
23
            Attachment(
24
                data=audio.content,
25
                file_name=f"call-{call_sid}.mp3",
26
                content_type="audio/mpeg",
27
            )
28
        ],
29
    )
30
    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.

What gets logged

With this setup, Opik records:

• Threads: one per phone call, containing every turn in order
• Traces: one per caller turn, with the caller’s message and the agent’s reply
• Spans: your retrieval, tools and LLM calls, including token usage and cost
• Guardrail results: the pass or fail result of each check you run on the turn
• Attachments: the call recording, playable in the Opik UI

Next steps

• Log traces — the full span and trace API
• Log media and attachments — attachment limits and supported types
• Guardrails — check inputs and outputs in the request path
• Online evaluation — score every call automatically once it is logged
• All integrations — the LLM provider and framework you use

Further improvements

If you have any questions or suggestions for improving the Twilio Agent Connect integration, please open an issue on our GitHub repository.