How to Orchestrate Multi-Call Conversations with an LLM and Twilio Conversation Memory in Python

Time to read:

September 14, 2026

Written by

Twilion

How to Orchestrate Multi-Call Conversations with an LLM and Twilio Conversation Memory in Python

Have you ever been on the phone with an AI voice agent and gotten frustrated with its lack of memory? Maybe your agent hung up on you, or it got disconnected, forcing you to start a conversation all over again. This kind of interruption can waste time for you and your users, and cause a lot of frustration.

Twilio Conversation Memory is your solution. Conversation Memory allows context to be persisted between calls. This means that if you call the Twilio agent back, it won’t lose the context of what you were talking about when you hung up, and can pick up right where you left off. This can save you a lot of frustration and help you get things done better and faster when you’re talking to an agent.

In this tutorial, you will make a Python FastAPI service that retains caller context, preferences and action history across multiple separate inbound calls.

Prerequisites

To complete this tutorial you will need:

A free Twilio account with a voice-capable phone number
Python 3.11 or later
An OpenAI API key
ngrok to expose local webhooks to Twilio
An IDE or text editor such as Visual Studio Code or PyCharm

Building the app

Step 1 - Set up the FastAPI project

To get started, create a new project directory and a virtual environment:

mkdir twilio-multi-call-memory
cd twilio-multi-call-memory
python3 -m venv .venv
source .venv/bin/activate

Step 2 - Install required dependencies

Install FastAPI, Uvicorn, the Twilio, OpenAI, httpx, and python-dotenv packages via pip.

pip install fastapi uvicorn twilio openai httpx python-dotenv

The Twilio package will allow your application to interface with Twilio’s services. The python-dotenv package allows you to import your environment variables into your solution using a .env file. You will add those variables in the next step. The OpenAI package will be used to connect your solution to OpenAI, and httpx will be used to make async calls to Twilio’s Conversation Memory REST API.

Step 3 - Create a Twilio Memory Store

For this tutorial, you will need a Conversation Memory Store. Go into your Twilio console and look for Memory Stores. You can use the console search, or look for Data > Conversation Memory > Memory stores.

Memory Stores use machine learning, and you may have to agree to a warning before proceeding. Keep in mind that Conversation Memory is not intended for use with sensitive information. Conversation products are only available on the new Twilio Console, so make sure your account has been migrated. For more information about Conversation Memory, you may want to read the documentation, including the Getting Started Guide.

Once you have found the correct tab, click on Create New Store.

Now follow the steps to set up your memory store.

The console gives you a setup checklist to get you started. Click on Connect Conversation Orchestrator, and give it a friendly name. You write a short description, then can move on to Messaging and Chat Traffic. For the remainder of the items in this checklist, you can select the default values for now.

You don’t have any customer profiles yet, so you can skip the rest of the checklist. However, you will need your memory store ID, which is at the top left of the memory store screen. There should be a convenient button to copy-paste that ID. Keep that ID for the next step. When you connect the Conversation Orchestrator, the console will also show you a configuration ID (prefixed cnv_config_) — copy that value too, as you’ll need it in the next step as well.

Step 4 - Configure environment variables

Now that you have a memory store created, you will need to be able to access that from your application. For this, you will need to get your Memory Store ID and paste that into your secrets file. Create a .env file in the root directory of your project. Add the following values, replacing the placeholders.

OPENAI_API_KEY=sk-...
TWILIO_API_KEY=SK...
TWILIO_API_SECRET=...
TWILIO_CONFIGURATION_ID=cnv_config_...
TWILIO_MEMORY_STORE_ID=mem_store_…
TWILIO_PHONE_NUMBER=+15551234567

Get your API key from your Twilio console, created under Settings > Account Settings > API Keys & Auth Tokens. Because other types of API keys do not have access to the Conversation Memory features, creating a Main API key is required for this tutorial. Keep in mind that the secret key will only be shown once, so be sure you save it. You get your memory store key from the previous step and paste it in here. Your OpenAI API Key is generated from OpenAI’s dashboard. You will also need your Twilio voice-capable phone number, which is in 10DLC format.

Save the file, and move on to the next step, creating your services.

Step 5 - Set up the OpenAI service

This demonstration uses the fiction of an auto repair shop as the agent that you are calling. However, Conversation Memory would be useful in lots of different scenarios, such as tech support, travel, and more. Feel free to adjust the audio prompts as you see fit for your own personal projects.

Create an openai_service.py module to handle interaction with gpt-4o-mini.

Paste the following into your new module:

import json
import os
from fastapi import WebSocket
from openai import AsyncOpenAI
MODEL = "gpt-4o-mini"
BASE_SYSTEM_PROMPT = """You are the phone assistant for Owlbert's Auto Repair. You are friendly, concise, and speak naturally as if on the phone.
Do not use lists, bullet points, or emojis — respond in plain sentences.
If the caller mentions their name, vehicle, or a problem with their car, remember it and refer back to it naturally as the conversation continues.
If you do not know something, say so honestly rather than guessing."""
SUMMARIZER_PROMPT = """You are summarizing a phone call between a caller and Owlbert's Auto Repair.
Write a concise summary in 4-8 sentences that captures: the caller's name if given, their vehicle if mentioned, the reason for the call, any symptoms or diagnostic detail discussed, any prices or estimates mentioned, and any commitments or next steps agreed upon.
Write in plain prose (no lists or bullets). If the caller did not share meaningful information (e.g. the call was very short or unclear), respond with a single sentence noting that.
Maximum 4000 characters."""
client = AsyncOpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
if not os.environ.get("OPENAI_API_KEY"):
    raise RuntimeError("OPENAI_API_KEY is not set.")
async def send_json(ws: WebSocket, payload: dict) -> None:
    await ws.send_text(json.dumps(payload))
async def stream_response(
    ws: WebSocket,
    call_sid: str | None,
    memory_context: str | None,
    messages: list[dict],
) -> str:
    system_prompt = (
        BASE_SYSTEM_PROMPT
        if not memory_context or not memory_context.strip()
        else f"{BASE_SYSTEM_PROMPT}\n\nPrior context on this customer (from previous calls):\n{memory_context}"
    )
    openai_messages = [{"role": "system", "content": system_prompt}] + messages
    full_response = ""
    try:
        stream = await client.chat.completions.create(
            model=MODEL,
            max_tokens=400,
            messages=openai_messages,
            stream=True,
        )
        async for chunk in stream:
            delta = chunk.choices[0].delta
            if delta.content:
                full_response += delta.content
                await send_json(ws, {"type": "text", "token": delta.content, "last": False})
    finally:
        await send_json(ws, {"type": "text", "token": "", "last": True})
    print(f"[{call_sid}] Assistant: {full_response}")
    return full_response
async def summarize_conversation(messages: list[dict]) -> str:
    if not messages:
        return ""
    transcript = "\n".join(f"{m['role'].upper()}: {m['content']}" for m in messages)
    response = await client.chat.completions.create(
        model=MODEL,
        max_tokens=500,
        messages=[
            {"role": "system", "content": SUMMARIZER_PROMPT},
            {"role": "user", "content": f"Transcript:\n{transcript}\n\nSummary:"},
        ],
    )
    summary = response.choices[0].message.content or ""
    print(f"Call summary ({len(summary)} chars): {summary}")
    return summary

This code handles your initial connection to OpenAI. Its function is to parse the information from a caller and stream it to the OpenAI API. Notice the system prompt here, which explains the functionality of the agent. It contains some useful instructions for the agent, such as to avoid bullet points and emojis when speaking on the phone.

This code also has an additional call to OpenAI to summarize the call itself. This will parse the conversation into a quick summary that will be stored in Twilio’s Conversation Memory. You can adjust this prompt according to your application’s needs. Keep in mind that if you don’t say anything meaningful, nothing will be stored.

Next, create the webhook for Twilio’s connection.

Step 6 - Build the Twilio webhook and Conversation Memory pipeline

Create another new file called conversation_relay_handler.py. Paste in this code:

from fastapi import WebSocket, WebSocketDisconnect
import openai_service
from conversation_memory_service import ConversationMemoryService
async def handle(
    ws: WebSocket,
    memory: ConversationMemoryService,
) -> None:
    call_sid: str | None = None
    caller_phone = ""
    memory_context: str | None = None
    conversation_id: str | None = None
    profile_id: str | None = None
    messages: list[dict] = []
    try:
        while True:
            data = await ws.receive_json()
            msg_type = data.get("type")
            if msg_type == "setup":
                call_sid = data.get("callSid")
                caller_phone = data.get("from", "")
                print(f"[{call_sid}] Call connected from {caller_phone}")
                try:
                    call_context = await memory.start_call(caller_phone)
                    if call_context is not None:
                        conversation_id = call_context.conversation_id
                        profile_id = call_context.profile_id
                        memory_context = call_context.memory_context
                        if memory_context and memory_context.strip():
                            print(f"[{call_sid}] Memory context loaded ({len(memory_context)} chars)")
                except Exception as exc:
                    print(f"[{call_sid}] Memory recall failed — continuing without context: {exc}")
                    memory_context = None
            elif msg_type == "prompt":
                if not data.get("last"):
                    continue
                user_text = data.get("voicePrompt", "")
                print(f"[{call_sid}] Caller: {user_text}")
                messages.append({"role": "user", "content": user_text})
                response_text = await openai_service.stream_response(
                    ws, call_sid, memory_context, messages
                )
                messages.append({"role": "assistant", "content": response_text})
            elif msg_type == "interrupt":
                spoken = data.get("utteranceUntilInterrupt", "")
                print(f"[{call_sid}] Interrupted after: '{spoken}'")
                if messages and messages[-1]["role"] == "assistant":
                    messages.pop()
            elif msg_type == "error":
                description = data.get("description", "")
                print(f"[{call_sid}] Conversation Relay error: {description}")
    except WebSocketDisconnect:
        print(f"[{call_sid}] Call ended")
    finally:
        if profile_id and conversation_id:
            summary = await openai_service.summarize_conversation(messages)
            await memory.finish_call(profile_id, conversation_id, summary)

conversation_relay_handler is the bridge between Twilio’s WebSocket and the rest of the app. When Twilio opens the socket after a <ConversationRelay> TwiML directive, handle loops reading frames one at a time. FastAPI’s receive_json assembles each frame and parses it as JSON for you. A prompt frame carries the caller’s transcribed speech. It appends the text to an in-memory conversation history and hands the whole history plus the memory context off to openai_service.stream_response, which streams tokens back out through the same socket.

The code also contains interruption handling for your agent. If the agent is interrupted during a conversation, it pops the last message off of the history so the agent realizes the full message was not sent and was incomplete. This will allow the customer to continue talking and handle the interruption in a more human way, without your user missing context.

When the call ends, the finally block hands the full conversation history off to openai_service.summarize_conversation and saves the resulting summary to Twilio via memory.finish_call, so the next call from this caller can pick up right where this one left off.

Step 7 - Process multi-turn historical context

Now you will create one more module to handle the context and memory processing. You’ll call this file conversation_memory_service.py. Paste the following into the file:

import base64
import json
import os
import re
from dataclasses import dataclass
from datetime import datetime, timezone
import httpx
MEMORY_BASE = "https://memory.twilio.com"
CONVERSATIONS_BASE = "https://conversations.twilio.com"
@dataclass
class CallContext:
    conversation_id: str
    profile_id: str
    memory_context: str
class ConversationMemoryService:
    def __init__(self) -> None:
        api_key = os.environ.get("TWILIO_API_KEY")
        api_secret = os.environ.get("TWILIO_API_SECRET")
        self.store_id = os.environ.get("TWILIO_MEMORY_STORE_ID")
        self.configuration_id = os.environ.get("TWILIO_CONFIGURATION_ID")
        self.twilio_number = os.environ.get("TWILIO_PHONE_NUMBER")
        if not api_key:
            raise RuntimeError("TWILIO_API_KEY not set.")
        if not api_secret:
            raise RuntimeError("TWILIO_API_SECRET not set.")
        if not self.store_id:
            raise RuntimeError("TWILIO_MEMORY_STORE_ID not set.")
        if not self.configuration_id:
            raise RuntimeError("TWILIO_CONFIGURATION_ID not set.")
        if not self.twilio_number:
            raise RuntimeError("TWILIO_PHONE_NUMBER not set.")
        basic = base64.b64encode(f"{api_key}:{api_secret}".encode()).decode()
        self._client = httpx.AsyncClient(
            headers={"Authorization": f"Basic {basic}"}
        )
    async def start_call(self, caller_phone: str) -> CallContext | None:
        if not caller_phone or not caller_phone.strip():
            return None
        conversation_id = await self._create_conversation(caller_phone)
        if conversation_id is None:
            return None
        profile_id = await self._ensure_profile(caller_phone)
        if profile_id is None:
            return CallContext(conversation_id, "", "")
        memory_context = await self._recall(profile_id)
        return CallContext(conversation_id, profile_id, memory_context)
    async def finish_call(self, profile_id: str, conversation_id: str, summary: str) -> None:
        if not profile_id or not conversation_id or not summary or not summary.strip():
            return
        resp = await self._client.post(
            f"{MEMORY_BASE}/v1/Stores/{self.store_id}/Profiles/{profile_id}/ConversationSummaries",
            json={
                "summaries": [
                    {
                        "conversationId": conversation_id,
                        "content": summary[:4096],
                        "occurredAt": datetime.now(timezone.utc).isoformat(),
                        "source": "voice-agent",
                    }
                ]
            },
        )
        if resp.status_code >= 400:
            print(f"ConversationSummaries write failed ({resp.status_code}): {resp.text}")
            return
        print(
            f"Saved conversation summary for profile {profile_id} "
            f"(conv {conversation_id}, {len(summary)} chars)"
        )
    async def _create_conversation(self, caller_phone: str) -> str | None:
        resp = await self._client.post(
            f"{CONVERSATIONS_BASE}/v2/Conversations",
            json={
                "configurationId": self.configuration_id,
                "name": f"Voice call {datetime.now(timezone.utc).isoformat()}",
                "participants": [
                    {
                        "name": "Caller",
                        "type": "CUSTOMER",
                        "addresses": [{"channel": "VOICE", "address": caller_phone}],
                    },
                    {
                        "name": "Owlbert Agent",
                        "type": "AI_AGENT",
                        "addresses": [{"channel": "VOICE", "address": self.twilio_number}],
                    },
                ],
            },
        )
        if resp.status_code == 409:
            existing = self._extract_conversation_id_from_conflict(resp.text)
            if existing is not None:
                print(f"Reusing existing conversation {existing} (previous call still open)")
                return existing
            print(f"Create Conversation 409 but could not extract existing id: {resp.text}")
            return None
        if resp.status_code >= 400:
            print(f"Create Conversation failed ({resp.status_code}): {resp.text}")
            return None
        try:
            body = resp.json()
        except json.JSONDecodeError:
            print(f"Create Conversation: could not extract id from response: {resp.text}")
            return None
        return body.get("id") or body.get("conversationId")
    @staticmethod
    def _extract_conversation_id_from_conflict(body: str) -> str | None:
        match = re.search(r"conv_conversation_[0-9a-z]+", body)
        return match.group(0) if match else None
    async def _ensure_profile(self, caller_phone: str) -> str | None:
        existing = await self._lookup_profile_id(caller_phone)
        if existing is not None:
            return existing
        resp = await self._client.post(
            f"{MEMORY_BASE}/v1/Stores/{self.store_id}/Profiles",
            json={"traits": {"Contact": {"phone": caller_phone}}},
        )
        if resp.status_code >= 400:
            print(f"Create Profile failed ({resp.status_code}): {resp.text}")
            return None
        try:
            body = resp.json()
        except json.JSONDecodeError:
            print(f"Create Profile: could not extract id from response: {resp.text}")
            return None
        return body.get("id")
    async def _lookup_profile_id(self, phone: str) -> str | None:
        resp = await self._client.post(
            f"{MEMORY_BASE}/v1/Stores/{self.store_id}/Profiles/Lookup",
            json={"idType": "phone", "value": phone},
        )
        if resp.status_code == 404:
            return None
        if resp.status_code >= 400:
            print(f"Profile lookup failed ({resp.status_code}): {resp.text}")
            return None
        try:
            body = resp.json()
        except json.JSONDecodeError:
            print(f"Could not parse profile lookup response: {resp.text}")
            return None
        profiles = body.get("profiles")
        if isinstance(profiles, list) and profiles:
            first = profiles[0]
            return first.get("profileId") or first.get("id")
        return body.get("profileId")
    async def _recall(self, profile_id: str) -> str:
        resp = await self._client.post(
            f"{MEMORY_BASE}/v1/Stores/{self.store_id}/Profiles/{profile_id}/Recall",
            json={
                "observationsLimit": 20,
                "summariesLimit": 5,
                "communicationsLimit": 0,
            },
        )
        if resp.status_code >= 400:
            print(f"Recall failed ({resp.status_code}): {resp.text}")
            return ""
        print(f"Recall raw response: {resp.text}")
        return self._format_recall(resp.text)
    @staticmethod
    def _format_recall(raw: str) -> str:
        lines: list[str] = []
        try:
            body = json.loads(raw)
        except json.JSONDecodeError:
            return ""
        for observation in body.get("observations", []) or []:
            text = ConversationMemoryService._extract_text(
                observation, "text", "content", "observation", "value"
            )
            if text and text.strip():
                lines.append(f"- {text}")
        for summary in body.get("summaries", []) or []:
            text = ConversationMemoryService._extract_text(
                summary, "text", "summary", "content", "value"
            )
            if text and text.strip():
                lines.append(f"Summary: {text}")
        return "\n".join(lines)
    @staticmethod
    def _extract_text(el, *candidates: str) -> str:
        if isinstance(el, str):
            return el
        if not isinstance(el, dict):
            return ""
        for name in candidates:
            value = el.get(name)
            if isinstance(value, str):
                return value
        return ""

This part of the code is what will handle your multi-turn conversation.

The first thing this code does is bring in all your environment variables from .env. It then creates a CallContext using conversation_id, profile_id, and memory_context to store some information about the call. Making an API call to Conversation Summaries, it stores a Customer ID to the Memory Store that you created earlier. It also creates a profile for your caller that stores their phone number. When the call is disconnected, the summary of the call generated by OpenAI will be stored to Twilio.

If the caller calls the number back too soon, the previous call memory may still be active and trying to log. This code accounts for that by checking to see if a call is finished with finish_call. If the old call has closed out, but it recognizes the caller’s profile, the API will retrieve the summary and context of the previous call from the Memory Store. The conversation can then resume on the same topic right where it left off!

The public entry point start_call first calls _create_conversation to create (or, on a 409 conflict, reuse) a Twilio Conversation for the call, then _ensure_profile to find or create a Memory Store profile for the caller’s phone number. _ensure_profile runs a two-step lookup: first _lookup_profile_id normalizes the number, matches it against a canonical profile, and returns a profile ID (or a 404, if no previous caller with that number was found); if none exists, it creates one. If a profile was found or created, _recall posts, asking for up to 20 observations and 5 summaries. The response is logged raw for inspection and then passed to _format_recall, which parses the JSON.

The application tries several plausible field names (text, content, observation, summary, value) via _extract_text and stitches whatever it finds into a bulleted string. That string is what eventually gets prepended to the OpenAI system prompt as “Prior context on this customer,” making the caller’s history part of the model’s instructions before they’ve even spoken.

Step 8 - Finalize your application

To complete your project you will need to create a main.py file to wire up the services that you’ve created. Paste the code below into the file:

import os
from dotenv import load_dotenv
from fastapi import FastAPI, Request, WebSocket
from fastapi.responses import Response
from conversation_memory_service import ConversationMemoryService
import conversation_relay_handler
load_dotenv()
app = FastAPI()
memory = ConversationMemoryService()
@app.post("/voice")
async def voice(request: Request) -> Response:
    host = request.url.hostname
    twiml = f"""<?xml version="1.0" encoding="UTF-8"?>
<Response>
    <Connect>
        <ConversationRelay url="wss://{host}/ws" welcomeGreeting="Thanks for calling Owlbert's Auto Repair. How can I help you today?" />
    </Connect>
</Response>"""
    return Response(content=twiml, media_type="application/xml")
@app.websocket("/ws")
async def ws_endpoint(websocket: WebSocket) -> None:
    await websocket.accept()
    await conversation_relay_handler.handle(websocket, memory)
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=int(os.environ.get("PORT", 8000)))

This sets up the websocket route to call your conversation_relay_handler, and includes the initial greeting for your user. Feel free to change the greeting according to your needs.

Testing, troubleshooting, or product demonstration

It is now time to test your voice application. Save your files and run the project with:

uvicorn main:app --reload --port 8000

Once your webhook is running, you will need to expose it to the internet by using ngrok or another tunneling service.

ngrok http 8000

Replace 8000 with whatever port your application is running on if you have a different port shown.

Now ngrok will provide you with a url for utilizing in your Twilio console. Go into your Twilio console and find the Twilio phone number that you prepared. Under the option A Call Comes In, choose Webhook, and fill in your ngrok URL followed by /voice, as shown in the graphic below:

configuration for webhooks in the Twilio console

To put your AI to the test, you’ll have to make two phone calls and check the conversation memory.

Make call #1: Talk to the agent about a car repair issue, including some details such as make and model.
Hang up and make call #2 from the same phone number.
Verify the agent greets you and remembers the details of your first call without any prompting.

If your Conversation Memory feature is working properly, you should also see details and a summary of the conversation saved to your Twilio dashboard. Check your Memory Store and it should show you the logged number you called from, as well as a stored conversation under Summaries:

Conclusion

Today you have learned how Twilio Conversation Memory simplifies maintaining state across separate voice calls in Python. This should provide value to any phone AI agent, storing information that keeps conversations feeling more convenient and human.

Do you want to do more with Twilio Conversations? Explore the possibilities by checking out the conversations documentation, where you can find blueprints for Conversational Agents, AI-to-Human handoff, and more.

Related Resources

Twilio Docs

From APIs to SDKs to sample apps

API reference documentation, SDKs, helper libraries, quickstarts, and tutorials for your language and platform.

Resource Center

The latest ebooks, industry reports, and webinars

Learn from customer engagement experts to improve your own communication.

Ahoy

Twilio's developer community hub

Best practices, code samples, and inspiration to build communications and digital engagement experiences.

How to Orchestrate Multi-Call Conversations with an LLM and Twilio Conversation Memory in Python

Related Posts

Related Resources