feat: openai#1760 Add SIP support for realtime agent runner (openai#1993)

Author: seratch

examples/realtime/twilio_sip/README.md

@@ -0,0 +1,55 @@
+# Twilio SIP Realtime Example
+
+This example shows how to handle OpenAI Realtime SIP calls with the Agents SDK. Incoming calls are accepted through the Realtime Calls API, a triage agent answers with a fixed greeting, and handoffs route the caller to specialist agents (FAQ lookup and record updates) similar to the realtime UI demo.
+
+## Prerequisites
+
+- Python 3.9+
+- An OpenAI API key with Realtime API access
+- A configured webhook secret for your OpenAI project
+- A Twilio account with a phone number and Elastic SIP Trunking enabled
+- A public HTTPS endpoint for local development (for example, [ngrok](https://ngrok.com/))
+
+## Configure OpenAI
+
+1. In [platform settings](https://platform.openai.com/settings) select your project.
+2. Create a webhook pointing to `https://<your-public-host>/openai/webhook` with "realtime.call.incoming" event type and note the signing secret. The example verifies each webhook with `OPENAI_WEBHOOK_SECRET`.
+
+## Configure Twilio Elastic SIP Trunking
+
+1. Create (or edit) an Elastic SIP trunk.
+2. On the **Origination** tab, add an origination SIP URI of `sip:proj_<your_project_id>@sip.api.openai.com;transport=tls` so Twilio sends inbound calls to OpenAI. (The Termination tab always ends with `.pstn.twilio.com`, so leave it unchanged.)
+3. Add at least one phone number to the trunk so inbound calls are forwarded to OpenAI.
+
+## Setup
+
+1. Install dependencies:
+ ```bash
+ uv pip install -r examples/realtime/twilio-sip/requirements.txt
+ ```
+2. Export required environment variables:
+ ```bash
+ export OPENAI_API_KEY="sk-..."
+ export OPENAI_WEBHOOK_SECRET="whsec_..."
+ ```
+3. (Optional) Adjust the multi-agent logic in `examples/realtime/twilio_sip/agents.py` if you want
+ to change the specialist agents or tools.
+4. Run the FastAPI server:
+ ```bash
+ uv run uvicorn examples.realtime.twilio_sip.server:app --host 0.0.0.0 --port 8000
+ ```
+5. Expose the server publicly (example with ngrok):
+ ```bash
+ ngrok http 8000
+ ```
+
+## Test a Call
+
+1. Place a call to the Twilio number attached to the SIP trunk.
+2. Twilio sends the call to `sip.api.openai.com`; OpenAI fires `realtime.call.incoming`, which this example accepts.
+3. The triage agent greets the caller, then either keeps the conversation or hands off to:
+ - **FAQ Agent** – answers common questions via `faq_lookup_tool`.
+ - **Records Agent** – writes short notes using `update_customer_record`.
+4. The background task attaches to the call and logs transcripts plus basic events in the console.
+
+You can edit `server.py` to change instructions, add tools, or integrate with internal systems once the SIP session is active.

examples/realtime/twilio_sip/__init__.py

@@ -0,0 +1 @@
+"""OpenAI Realtime SIP example package."""

examples/realtime/twilio_sip/agents.py

@@ -0,0 +1,87 @@
+"""Realtime agent definitions shared by the Twilio SIP example."""
+
+from __future__ import annotations
+
+import asyncio
+
+from agents import function_tool
+from agents.extensions.handoff_prompt import RECOMMENDED_PROMPT_PREFIX
+from agents.realtime import RealtimeAgent, realtime_handoff
+
+# --- Tools -----------------------------------------------------------------
+
+
+WELCOME_MESSAGE = "Hello, this is ABC customer service. How can I help you today?"
+
+
+@function_tool(
+ name_override="faq_lookup_tool", description_override="Lookup frequently asked questions."
+)
+async def faq_lookup_tool(question: str) -> str:
+ """Fetch FAQ answers for the caller."""
+
+ await asyncio.sleep(3)
+
+ q = question.lower()
+ if "plan" in q or "wifi" in q or "wi-fi" in q:
+ return "We provide complimentary Wi-Fi. Join the ABC-Customer network." # demo data
+ if "billing" in q or "invoice" in q:
+ return "Your latest invoice is available in the ABC portal under Billing > History."
+ if "hours" in q or "support" in q:
+ return "Human support agents are available 24/7; transfer to the specialist if needed."
+ return "I'm not sure about that. Let me transfer you back to the triage agent."
+
+
+@function_tool
+async def update_customer_record(customer_id: str, note: str) -> str:
+ """Record a short note about the caller."""
+
+ await asyncio.sleep(1)
+ return f"Recorded note for {customer_id}: {note}"
+
+
+# --- Agents ----------------------------------------------------------------
+
+
+faq_agent = RealtimeAgent(
+ name="FAQ Agent",
+ handoff_description="Handles frequently asked questions and general account inquiries.",
+ instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
+ You are an FAQ specialist. Always rely on the faq_lookup_tool for answers and keep replies
+ concise. If the caller needs hands-on help, transfer back to the triage agent.
+ """,
+ tools=[faq_lookup_tool],
+)
+
+records_agent = RealtimeAgent(
+ name="Records Agent",
+ handoff_description="Updates customer records with brief notes and confirmation numbers.",
+ instructions=f"""{RECOMMENDED_PROMPT_PREFIX}
+ You handle structured updates. Confirm the customer's ID, capture their request in a short
+ note, and use the update_customer_record tool. For anything outside data updates, return to the
+ triage agent.
+ """,
+ tools=[update_customer_record],
+)
+
+triage_agent = RealtimeAgent(
+ name="Triage Agent",
+ handoff_description="Greets callers and routes them to the most appropriate specialist.",
+ instructions=(
+ f"{RECOMMENDED_PROMPT_PREFIX} "
+ "Always begin the call by saying exactly: '"
+ f"{WELCOME_MESSAGE}' "
+ "before collecting details. Once the greeting is complete, gather context and hand off to "
+ "the FAQ or Records agents when appropriate."
+ ),
+ handoffs=[faq_agent, realtime_handoff(records_agent)],
+)
+
+faq_agent.handoffs.append(triage_agent)
+records_agent.handoffs.append(triage_agent)
+
+
+def get_starting_agent() -> RealtimeAgent:
+ """Return the agent used to start each realtime call."""
+
+ return triage_agent

examples/realtime/twilio_sip/requirements.txt

@@ -0,0 +1,3 @@
+fastapi>=0.120.0
+openai>=2.2,<3
+uvicorn[standard]>=0.38.0

examples/realtime/twilio_sip/server.py

@@ -0,0 +1,211 @@
+"""Minimal FastAPI server for handling OpenAI Realtime SIP calls with Twilio."""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+
+import websockets
+from fastapi import FastAPI, HTTPException, Request, Response
+from openai import APIStatusError, AsyncOpenAI, InvalidWebhookSignatureError
+
+from agents.realtime.config import RealtimeSessionModelSettings
+from agents.realtime.items import (
+ AssistantAudio,
+ AssistantMessageItem,
+ AssistantText,
+ InputText,
+ UserMessageItem,
+)
+from agents.realtime.model_inputs import RealtimeModelSendRawMessage
+from agents.realtime.openai_realtime import OpenAIRealtimeSIPModel
+from agents.realtime.runner import RealtimeRunner
+
+from .agents import WELCOME_MESSAGE, get_starting_agent
+
+logging.basicConfig(level=logging.INFO)
+
+logger = logging.getLogger("twilio_sip_example")
+
+
+def _get_env(name: str) -> str:
+ value = os.getenv(name)
+ if not value:
+ raise RuntimeError(f"Missing environment variable: {name}")
+ return value
+
+
+OPENAI_API_KEY = _get_env("OPENAI_API_KEY")
+OPENAI_WEBHOOK_SECRET = _get_env("OPENAI_WEBHOOK_SECRET")
+
+client = AsyncOpenAI(api_key=OPENAI_API_KEY, webhook_secret=OPENAI_WEBHOOK_SECRET)
+
+# Build the multi-agent graph (triage + specialist agents) from agents.py.
+assistant_agent = get_starting_agent()
+
+app = FastAPI()
+
+# Track background tasks so repeated webhooks do not spawn duplicates.
+active_call_tasks: dict[str, asyncio.Task[None]] = {}
+
+
+async def accept_call(call_id: str) -> None:
+ """Accept the incoming SIP call and configure the realtime session."""
+
+ # The starting agent uses static instructions, so we can forward them directly to the accept
+ # call payload. If someone swaps in a dynamic prompt, fall back to a sensible default.
+ instructions_payload = (
+ assistant_agent.instructions
+ if isinstance(assistant_agent.instructions, str)
+ else "You are a helpful triage agent for ABC customer service."
+ )
+
+ try:
+ # AsyncOpenAI does not yet expose high-level helpers like client.realtime.calls.accept, so
+ # we call the REST endpoint directly via client.post(). Keep this until the SDK grows an
+ # async helper.
+ await client.post(
+ f"/realtime/calls/{call_id}/accept",
+ body={
+ "type": "realtime",
+ "model": "gpt-realtime",
+ "instructions": instructions_payload,
+ },
+ cast_to=dict,
+ )
+ except APIStatusError as exc:
+ if exc.status_code == 404:
+ # Twilio occasionally retries webhooks after the caller hangs up; treat as a no-op so
+ # the webhook still returns 200.
+ logger.warning(
+ "Call %s no longer exists when attempting accept (404). Skipping.", call_id
+ )
+ return
+
+ detail = exc.message
+ if exc.response is not None:
+ try:
+ detail = exc.response.text
+ except Exception: # noqa: BLE001
+ detail = str(exc.response)
+
+ logger.error("Failed to accept call %s: %s %s", call_id, exc.status_code, detail)
+ raise HTTPException(status_code=500, detail="Failed to accept call") from exc
+
+ logger.info("Accepted call %s", call_id)
+
+
+async def observe_call(call_id: str) -> None:
+ """Attach to the realtime session and log conversation events."""
+
+ runner = RealtimeRunner(assistant_agent, model=OpenAIRealtimeSIPModel())
+
+ try:
+ initial_model_settings: RealtimeSessionModelSettings = {
+ "turn_detection": {
+ "type": "semantic_vad",
+ "interrupt_response": True,
+ }
+ }
+ async with await runner.run(
+ model_config={
+ "call_id": call_id,
+ "initial_model_settings": initial_model_settings,
+ }
+ ) as session:
+ # Trigger an initial greeting so callers hear the agent right away.
+ # Issue a response.create immediately after the WebSocket attaches so the model speaks
+ # before the caller says anything. Using the raw client message ensures zero latency
+ # and avoids threading the greeting through history.
+ await session.model.send_event(
+ RealtimeModelSendRawMessage(
+ message={
+ "type": "response.create",
+ "other_data": {
+ "response": {
+ "instructions": (
+ "Say exactly '"
+ f"{WELCOME_MESSAGE}"
+ "' now before continuing the conversation."
+ )
+ }
+ },
+ }
+ )
+ )
+
+ async for event in session:
+ if event.type == "history_added":
+ item = event.item
+ if isinstance(item, UserMessageItem):
+ for user_content in item.content:
+ if isinstance(user_content, InputText) and user_content.text:
+ logger.info("Caller: %s", user_content.text)
+ elif isinstance(item, AssistantMessageItem):
+ for assistant_content in item.content:
+ if (
+ isinstance(assistant_content, AssistantText)
+ and assistant_content.text
+ ):
+ logger.info("Assistant (text): %s", assistant_content.text)
+ elif (
+ isinstance(assistant_content, AssistantAudio)
+ and assistant_content.transcript
+ ):
+ logger.info(
+ "Assistant (audio transcript): %s",
+ assistant_content.transcript,
+ )
+ elif event.type == "error":
+ logger.error("Realtime session error: %s", event.error)
+
+ except websockets.exceptions.ConnectionClosedError:
+ # Callers hanging up causes the WebSocket to close without a frame; log at info level so it
+ # does not surface as an error.
+ logger.info("Realtime WebSocket closed for call %s", call_id)
+ except Exception as exc: # noqa: BLE001 - demo logging only
+ logger.exception("Error while observing call %s", call_id, exc_info=exc)
+ finally:
+ logger.info("Call %s ended", call_id)
+ active_call_tasks.pop(call_id, None)
+
+
+def _track_call_task(call_id: str) -> None:
+ existing = active_call_tasks.get(call_id)
+ if existing:
+ if not existing.done():
+ logger.info(
+ "Call %s already has an active observer; ignoring duplicate webhook delivery.",
+ call_id,
+ )
+ return
+ # Remove completed tasks so a new observer can start for a fresh call.
+ active_call_tasks.pop(call_id, None)
+
+ task = asyncio.create_task(observe_call(call_id))
+ active_call_tasks[call_id] = task
+
+
+@app.post("/openai/webhook")
+async def openai_webhook(request: Request) -> Response:
+ body = await request.body()
+
+ try:
+ event = client.webhooks.unwrap(body, request.headers)
+ except InvalidWebhookSignatureError as exc:
+ raise HTTPException(status_code=400, detail="Invalid webhook signature") from exc
+
+ if event.type == "realtime.call.incoming":
+ call_id = event.data.call_id
+ await accept_call(call_id)
+ _track_call_task(call_id)
+ return Response(status_code=200)
+
+ # Ignore other webhook event types for brevity.
+ return Response(status_code=200)
+
+
+@app.get("/")
+async def healthcheck() -> dict[str, str]:
+ return {"status": "ok"}

src/agents/realtime/model.py

@@ -139,6 +139,13 @@ class RealtimeModelConfig(TypedDict):
  is played to the user.
  """
 
+ call_id: NotRequired[str]
+ """Attach to an existing realtime call instead of creating a new session.
+
+ When provided, the transport connects using the `call_id` query string parameter rather than a
+ model name. This is used for SIP-originated calls that are accepted via the Realtime Calls API.
+ """
+
 
 class RealtimeModel(abc.ABC):
  """Interface for connecting to a realtime model and sending/receiving events."""