DEV Community: Sonam Gupta

Build a Schema-First AI Call QA Pipeline in Python

Sonam Gupta — Wed, 21 Oct 2026 20:26:29 +0000

Free-text call summaries are useful for people, but awkward for software. A dashboard cannot reliably group a paragraph by sentiment, count missing disclosures, or identify why callers escalated without another parsing step.

This tutorial builds a schema-first AI call QA pipeline in Python. Each completed conversation becomes a set of typed records for sentiment, compliance, and escalation. Those records can move directly into a scorecard because the expected fields, enums, and required values are declared before analysis begins.

The complete sample is available in the Telnyx Code Examples repository.

What the pipeline does

The example has four layers:

JSON-schema insight templates
        |
        v
Insight group: qa-scorecard-v1
        |
        v
AI Assistant insight_settings
        |
        v
Conversation insights -> validation -> QA scorecard

It creates three versioned insight templates:

sentiment-v1 returns positivity, frustration, overall sentiment, and key emotions.
compliance-v1 returns an identity-verification result and a constrained list of compliance flags.
escalation-v1 returns a controlled escalation reason and whether the agent handled it well.

The templates are assigned to one insight group. That group is attached to an AI Assistant once, so future assistant conversations can be analyzed against all three contracts.

Why start with JSON Schema?

Without a contract, an analysis model might describe the same issue as "caller asked for a manager," "requested escalation," or "wanted a supervisor." Those phrases may be equivalent to a reader, but they create inconsistent analytics.

The escalation template constrains the result instead:

ESCALATION_SCHEMA = {
    "type": "object",
    "properties": {
        "escalation_reason": {
            "type": "string",
            "enum": [
                "caller_request",
                "compliance_violation",
                "poor_call_quality",
                "system_error",
                "other",
            ],
        },
        "agent_performed_well": {"type": "boolean"},
    },
    "required": ["escalation_reason", "agent_performed_well"],
    "additionalProperties": False,
}

The sample sends json_schema as a JSON string, which is the canonical wire format used by create_templates.py:

payload = {
    "name": "escalation-v1",
    "instructions": "Determine whether the caller escalated and why.",
    "json_schema": json.dumps(ESCALATION_SCHEMA),
}

requests.post(
    f"{API_BASE}/ai/conversations/insights",
    headers=headers,
    json=payload,
    timeout=30,
)

The API also accepts the schema as an object. See the structured insights documentation for the concept and request shape.

Create and verify the insight group

create_group.py finds or creates qa-scorecard-v1, assigns the three templates, and then reads the group back. The script exits if any expected template is missing from the returned membership.

This read-after-write check is important. A successful request only proves that the API accepted it; reading the resource back proves that the configuration the assistant will use is present.

for name in TEMPLATE_NAMES:
    _assign(headers, group_id, templates[name])

membership = _read_membership(headers, group_id)
assigned_names = {item.get("name") for item in membership}

missing = [name for name in TEMPLATE_NAMES if name not in assigned_names]
if missing:
    sys.exit(f"Assignment verification failed - missing {missing}")

Telnyx insight groups let you apply a reusable bundle of analysis templates to an assistant instead of managing each template separately.

Attach the group to an AI Assistant

The assistant is connected to the group through insight_settings:

payload = {
    "insight_settings": {
        "insight_group_id": group_id,
    }
}

requests.post(
    f"{API_BASE}/ai/assistants/{assistant_id}",
    headers=headers,
    json=payload,
    timeout=30,
)

wire_assistant.py follows that update with a GET request and verifies that the returned insight_group_id matches the expected group. If ASSISTANT_ID is not set, the script first creates an assistant using the platform-default model and voice.

Receive completed insights without depending on one delivery path

The sample supports two ways to discover completed analysis:

A call.conversation.insights_generated webhook provides a completion signal for real-time workflows.
GET /ai/conversations/{conversation_id}/conversations-insights remains the polling source of truth and works even when the webhook receiver is not running.

The Flask receiver verifies Telnyx's Ed25519 signature before accepting an event. TELNYX_PUBLIC_KEY is required, and there is no unverified fallback path. Missing or invalid signature data returns 400.

Polling also makes the lifecycle explicit:

Status	Pipeline behavior
`pending`	Count the record as pending.
`in_progress`	Keep waiting; analysis is not complete.
`completed`	Validate the result against its declared schema.
`failed`	Report the failed record and stop the live-call workflow with a nonzero exit.

That distinction prevents a slow record from being mislabeled as bad data.

Validate before aggregating

scorecard.py maps each returned template ID back to its template name, parses the result, and validates required fields and enum values. Only valid completed records contribute to the scorecard.

The output aggregates:

sentiment distribution;
compliance-flag counts; and
escalation reasons.

A completed result that violates its schema is printed with its conversation ID and excluded from the totals. You can test this behavior locally with --strict, which injects a synthetic sentiment record missing positivity_score:

python scorecard.py --strict

This gives the pipeline two separate failure categories: records that are not ready yet and records that are complete but structurally invalid.

Run the sample

Clone the repository and install the Python dependencies:

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/insights-pipeline

python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

cp .env.example .env

Configure .env with your Telnyx API key and placeholders for the assistant and destination numbers:

TELNYX_API_KEY=YOUR_TELNYX_API_KEY
ASSISTANT_PHONE_NUMBER=<E164_ASSISTANT_NUMBER>
DESTINATION_NUMBER=<E164_TEST_NUMBER>
ASSISTANT_ID=<OPTIONAL_ASSISTANT_ID>
PUBLIC_WEBHOOK_BASE_URL=<OPTIONAL_HTTPS_BASE_URL>
TELNYX_PUBLIC_KEY=<REQUIRED_ONLY_FOR_WEBHOOK_RECEIVER>
DEMO_MODE=true

Keep API keys and private credentials in environment variables or a secret manager. Do not place them in source control, prompts, or screenshots.

Run the scripts in order:

python create_templates.py
python create_group.py
python wire_assistant.py
python place_call.py
python scorecard.py --date YYYY-MM-DD

DEMO_MODE=true is the default. In that mode, place_call.py prints the scheduled-event payload but does not dial a real number or write a fake conversation ID. Set DEMO_MODE=false only when you are ready to place the configured test call.

The webhook receiver is optional because the polling path can complete the workflow without it:

python webhook_receiver.py

For a manual scorecard run, you can bypass the local conversation store:

python scorecard.py --conversation-id <CONVERSATION_ID>

Where to take it next

The schema-first pattern extends beyond support QA. Add templates for appointment outcomes, sales qualification, cancellation reasons, accessibility needs, or any other result that downstream software must count and filter reliably.

For production, replace the local conversations.json ledger with durable storage, send invalid records to an operational review queue, and persist the template version beside each result. The central contract remains the same: define the record first, analyze the conversation second, and aggregate only data that passes validation.

Resources

Build an Auditable AI Phone Workflow with Telnyx Missions

Sonam Gupta — Wed, 21 Oct 2026 18:26:34 +0000

An AI phone workflow can return a successful response and still leave a difficult question unanswered: what actually happened?

If a clinic says an assistant called a patient, confirmed an appointment, and completed the outreach plan, a reviewer may need more than application logs. They need the objective, the approved plan, every observed state transition, the assistant associated with the run, and the final result in one durable record.

The open-source missions-audit sample shows one way to build that record with Python and Telnyx AI Missions. It creates a mission, starts a run, submits a plan, advances step statuses, logs typed events, optionally schedules a real AI-assisted phone call, and renders the resulting ledger as a timestamped audit report.

The important idea is simple: the audit trail is part of the workflow's API contract, not a story reconstructed later from process logs.

What is an AI mission audit trail?

A mission is the durable objective. A run is one execution of that objective. Each run has an ordered plan, and each plan contains steps whose statuses change as work progresses.

The sample combines five related resources:

Resource	Purpose
Mission	Stores the durable objective
Run	Represents one execution and its final result
Plan and steps	Describe the intended work and its progress
Typed events	Record the chronological audit timeline
Agent link	Identifies the Telnyx AI assistant associated with the run

This produces a queryable record of what was planned, what the worker observed, and how the run ended.

The five-script workflow

The project separates the workflow into five small programs:

create_mission.py creates the durable mission.
start_run.py starts one execution of that mission.
plan_steps.py submits an ordered three-step plan.
execute.py advances the run, steps, and event timeline.
audit.py reads fresh server state and renders the report.

The default plan is:

provision_assistant
schedule_outreach_call
verify_outcome

Those names describe the live workflow, but the project is safe by default. More on that distinction shortly.

Creating a mission, run, and plan

The sample uses raw HTTP requests against https://api.telnyx.com/v2. Missions responses are wrapped in a data envelope and use identifiers such as mission_id and run_id.

Creating the durable objective starts with POST /ai/missions:

mission = post("/ai/missions", {
    "name": "Clinic outreach audit",
    "description": "Verify clinic contacts can be reached",
    "instructions": (
        "Verify the contact is reachable, confirm the appointment "
        "slot briefly, and stay polite."
    ),
    "execution_mode": "external",
})

The worker is external, so the application controls execution while the Missions API keeps the durable record.

One execution is created with:

run = post(f"/ai/missions/{mission_id}/runs", {})

The run begins in pending. The worker can advance it through running to succeeded or failed. The API also supports paused and cancelled states for operator intervention.

The plan uses client-selected step_id values, descriptions, and one-based sequence numbers:

steps = [
    {
        "step_id": "provision_assistant",
        "description": "Create or select the outreach assistant.",
        "sequence": 1,
    },
    {
        "step_id": "schedule_outreach_call",
        "description": "Create one scheduled phone-call event.",
        "sequence": 2,
    },
    {
        "step_id": "verify_outcome",
        "description": "Confirm the event reached a terminal status.",
        "sequence": 3,
    },
]

post(f"/ai/missions/{mission_id}/runs/{run_id}/plan", {"steps": steps})

Steps move from pending to in_progress, then to completed or failed. A skipped state is available for intentional no-ops.

The worker's honesty contract

The most useful pattern in the sample is not an endpoint. It is a rule enforced by execute.py:

Record every state change the worker observes, and never invent a platform outcome.

When a step starts, the worker updates the step and writes a step_started event. When it completes, the worker writes the corresponding status and step_completed event. Failures become both a failed step and a step_failed event.

def run_step(step_id, work):
    update_step(step_id, "in_progress")
    log_event("step_started", f"Step {step_id} started", step_id)

    try:
        result = work()
    except Exception as exc:
        update_step(step_id, "failed")
        log_event(
            "step_failed",
            f"Step {step_id} failed: {exc}",
            step_id,
        )
        raise

    update_step(step_id, "completed")
    log_event("step_completed", f"Step {step_id} completed", step_id)
    return result

In live mode, the worker polls the scheduled-event API every five seconds, with a hard ten-minute cap. Each newly observed event status becomes a status_change mission event. A request failure or timeout becomes an explicit error event and causes the verification step to fail.

That means the ledger never reports that a call completed merely because the worker expected it to complete.

Demo mode and live-call mode are deliberately different

DEMO_MODE=true is the default.

In demo mode, the sample exercises the mission, run, plan, step, and event APIs without creating an assistant, creating a scheduled event, placing a call, or linking an agent. It records a custom event with this meaning:

{
  "demo_mode": true,
  "note": "no assistant created, no call placed"
}

The run's result summary says the same thing. The simulated walk is useful for learning the ledger mechanics, but it cannot be mistaken for evidence of a real phone call.

Set DEMO_MODE=false to exercise the live path. Live mode requires placeholder-backed environment variables for a destination you control and a phone number attached to the assistant:

TELNYX_API_KEY=YOUR_TELNYX_API_KEY
DEMO_MODE=false
DESTINATION_NUMBER=<E164_TEST_NUMBER>
ASSISTANT_PHONE_NUMBER=<E164_ASSISTANT_NUMBER>

The live worker can create or reuse an assistant, schedule the phone-call event, poll its status, record the latest call attempt, and link the assistant to the mission run. The Assistants and scheduled-events APIs return bare JSON objects rather than the Missions API's data envelope, so the sample handles those response shapes separately.

Rendering the report from fresh state

audit.py waits for the run to reach succeeded, failed, or cancelled, with the same ten-minute safety cap. It then performs fresh reads of the mission, run, plan, events, and linked agents.

The report is therefore built from the server's current view rather than from the worker's in-memory state:

MISSIONS AUDIT REPORT

Mission
  mission_id: ...
  name: Clinic outreach audit

Run
  status: succeeded
  result_summary: ...

Steps
  provision_assistant     completed
  schedule_outreach_call  completed
  verify_outcome          completed

Event timeline
  step_started
  step_completed
  status_change
  custom

Linked Telnyx agents
  run_id=... telnyx_agent_id=...

If the report reaches its polling cap before the run becomes terminal, it prints a warning, performs a final fresh read, renders the current state, and exits unsuccessfully. It does not silently present an unfinished run as complete.

Run the sample

Clone the repository and install its dependencies:

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/missions-audit

python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

cp .env.example .env

Add your Telnyx API key to .env. Keep DEMO_MODE=true for the side-effect-free walkthrough.

Run the included smoke test first:

pytest smoke_test.py -q

Then run the workflow, passing the identifiers printed by each step:

python create_mission.py
python start_run.py --mission-id <MISSION_ID>
python plan_steps.py --mission-id <MISSION_ID> --run-id <RUN_ID>
python execute.py --mission-id <MISSION_ID> --run-id <RUN_ID>
python audit.py --mission-id <MISSION_ID> --run-id <RUN_ID>

For live mode, use only test numbers you control and configure the required assistant phone number before running execute.py.

Where this pattern fits

The clinic example is only one use case. The same structure works anywhere an automated workflow must be replayable:

Collections outreach where each contact attempt needs a durable record
Appointment reminders that require proof of execution and outcome
Incident-response agents whose plans and tool calls need review
Sales workflows where a human must approve or audit each stage
Operational automations that can pause, resume, fail, or be cancelled

The reusable design is a durable objective, an explicit plan, typed events, independently queryable state, and a final report assembled from fresh reads.

Resources

An audit trail is most useful when it is designed before the first action runs. With Missions, plans, typed events, and agent linkage treated as API resources, "what happened?" becomes a query instead of an investigation.

Build Observable AI Model Fallbacks with Telnyx AI Gateway

Sonam Gupta — Tue, 20 Oct 2026 02:08:30 +0000

An AI model fallback is only useful if your application can answer two questions afterward:

Which model actually served the request?
How many attempts did the gateway make?

A client-side retry loop can hide both answers. The application eventually receives a successful response, but the path to that response is difficult to reconstruct. SDK retries can also stack on top of gateway retries, creating extra requests and misleading attempt counts.

The gateway-fallbacks Python sample moves that resilience policy into Telnyx AI Gateway. It provisions a primary and fallback model, forces a timeout on the primary, and proves the resulting failover twice:

response headers show the model and attempt count on the live request
the spend ledger records the requested model, serving model, and attempts together

The architecture

The example separates management, inference, and observation:

Management plane                         Inference plane
api.telnyx.com/v2/llm_token_gateway     llm.telnyx.com/v1
Telnyx account API key                  scoped token key

POST /token_groups                      GET /models
POST /token_keys                        POST /chat/completions
GET  /spend/events                             |
GET  /usage/summary                            v
        |                               gateway retries and fallbacks
        +----------------------+----------------+
                               |
                               v
Observation: x-ltg-served-model, x-ltg-attempts,
             requested_model, model, attempts, cache_hit

The management plane owns token groups, token keys, budgets, and usage reporting. It uses the Telnyx account API key.

The inference plane is OpenAI-compatible. It uses only the token key issued for the group, so application code does not need the account credential.

Put the fallback policy in the token group

The sample provisions two allowed model aliases and one fallback relationship:

allowed_models = ["Kimi-K3", "GLM-5.3-Flash"]
fallbacks = {
    "Kimi-K3": ["GLM-5.3-Flash"],
}
retries = {
    "max_attempts": 2,
    "retry_delay_ms": 250,
    "backoff": "exponential",
}

group = client.llm_token_gateway.token_groups.create(
    name="gateway-fallbacks-demo",
    allowed_models=allowed_models,
    fallbacks=fallbacks,
    retries=retries,
    max_budget=10,
    budget_duration="1d",
    idempotency_key=f"gateway-fallbacks-group-{uuid.uuid4()}",
).data

The group constrains which models its key can call and defines how failures move through the chain. The sample validates the same policy limits locally before making the API request:

fallback-chain depth cannot exceed three
max_attempts must be between one and five
a chain cannot reference its own primary or contain duplicates
every fallback must appear in allowed_models
max_attempts × (1 + longest chain) cannot exceed eight

The gateway also enforces these rules. policy.py deliberately sends an invalid max_attempts=9 update and expects an HTTP 400 without mutating the group.

Keep retries in one layer

The gateway owns retries and model fallbacks, so the OpenAI client disables its own retry behavior:

llm = OpenAI(
    base_url="https://llm.telnyx.com/v1",
    api_key=token_key,
    max_retries=0,
)

This matters because an SDK retry is a new request layered on top of the gateway's attempts. Disabling it keeps the response headers, ledger records, and billing aligned around one retry policy.

How the gateway treats failures

The sample distinguishes failures where no model work occurred from failures where processing may already have started.

Definite no-work failures such as pre-dispatch errors, 429, and 529 can retry the same model.

Uncertain failures such as timeouts, 408, 500, 502, 503, and 504 move to the next fallback because the original request may already have reached the provider.

A context-length rejection is terminal because sending the same oversized context to another attempt does not fix the input.

Force a model fallback for the demo

Waiting for a provider incident is not a useful test plan. The sample creates a controlled race instead.

inference.py sends a streaming request to Kimi-K3 with a one-second first-event deadline:

result = chat_completion(
    token_key,
    "Kimi-K3",
    build_forced_failure_prompt(),
    extra_headers={
        "x-ltg-request-timeout": "1000",
        "x-ltg-max-attempts": "1",
    },
    stream=True,
    max_tokens=64,
)

The generated input is intentionally heavy, approximately 22,000 tokens, and includes a unique nonce so a repeated run cannot rely on provider KV caching to shorten prefill. The goal is for the primary's first event to miss the one-second deadline while GLM-5.3-Flash completes as the fallback.

Provider latency changes over time, so this is not presented as an absolute guarantee. The script tries up to three times and reports honestly if the primary answers within the deadline or if both attempts time out.

Verify the live request through headers

The client reads two headers from the raw response:

served_model = raw.headers.get("x-ltg-served-model")
attempts = raw.headers.get("x-ltg-attempts")

A successful failover should show:

requested model:   Kimi-K3
served model:      GLM-5.3-Flash
x-ltg-attempts:    2 or more

The SDK response model is useful, but the gateway headers make the routing decision explicit on the live path.

Verify the same event in the spend ledger

The next step reads the gateway spend events for the issued token key:

events = client.llm_token_gateway.spend.retrieve_events(
    start_date=start_date,
    end_date=end_date,
    token_key_id=token_key_id,
    page_number=1,
    page_size=50,
)

for event in events.data:
    print(
        event.requested_model,
        "->",
        event.model,
        "attempts:",
        event.attempts,
        "status:",
        event.status,
        "cache_hit:",
        event.cache_hit,
    )

When requested_model and model differ, the ledger provides the durable counterpart to the response header. The sample uses a half-open UTC date range and snapshot pagination so the report remains reproducible as the event set changes.

GET /usage/summary adds totals and per-model rollups when the account has the llm_token_gateway.usage.read permission. A 403 from that endpoint is treated as a nonfatal warning because the spend-events ledger still contains the request-level proof.

Per-request controls, caching, and mutation safety

The gateway supports request-level overlays for attempts, delay, backoff, and timeout:

extra_headers = {
    "x-ltg-max-attempts": "1",
    "x-ltg-retry-delay": "1000",
    "x-ltg-backoff": "exponential",
}

These controls adjust one request without rewriting the token-group policy.

The sample also updates cache_ttl to 60 seconds and sends the same prompt twice. Gateway caching is best effort, so the script does not invent a cache hit: it reports x-ltg-attempts: 0 when the repeat is served from cache and otherwise directs you to the ledger's cache_hit field.

Every update and delete follows the mutation-safety pattern:

retrieve a fresh ETag
send it through If-Match
include an Idempotency-Key

A stale ETag returns 412, preventing one process from overwriting another process's newer configuration.

Run the sample

Clone the repository and install the pinned dependencies:

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/gateway-fallbacks

python -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt

cp .env.example .env

Add your Telnyx account API key to .env:

TELNYX_API_KEY=your_telnyx_api_key_here
PRIMARY_MODEL=Kimi-K3
FALLBACK_MODEL=GLM-5.3-Flash

Run the local smoke tests before making network requests:

python -m pytest smoke_test.py -v

Then run the five-step demonstration:

# Creates the group and key. The key secret is printed once.
python provision.py

export GATEWAY_TOKEN_KEY="<paste the secret>"

python inference.py
python ledger.py
python policy.py
python teardown.py

The live demonstration incurs real inference spend. The sample configures a maximum budget of USD 10 over one day, but you should review the policy before running it in your account.

provision.py stores only the group and key IDs in .gateway_state.json; it never writes the token secret to disk. teardown.py revokes the token key, deletes the group, verifies that the revoked key receives HTTP 401, and removes the local state file.

The reusable pattern

The useful pattern is not limited to these two models:

Put model access, retry rules, fallbacks, and budgets in gateway policy.
Give applications a scoped token key instead of the account credential.
Disable overlapping client retries.
Read routing evidence from the response headers.
Confirm the same event in the spend ledger.

That turns model failover from an assumption into something operators can inspect during the request and audit afterward.

Resources

Build a WhatsApp Coexistence Sync Monitor in Python

Sonam Gupta — Fri, 16 Oct 2026 23:30:28 +0000

WhatsApp coexistence lets a team use the WhatsApp Business app and the WhatsApp Business API on the same phone number.

That is useful when human agents handle sensitive or unusual conversations in the app while automated systems send reminders and updates through the API. It also creates a synchronization problem: when a human sends a message in the Business app, the API-side application needs a trustworthy way to learn what happened.

The whatsapp-coexistence-sync-monitor example is a small Flask service that closes that gap. It:

verifies Telnyx webhook signatures before parsing an event
receives message.echo events created by Business-app activity
checks whether the number is currently in active coexistence state
acknowledges but skips events while the number is not active
writes accepted events to a PII-minimized JSONL audit trail

Architecture

The sample has three components:

WhatsApp Business app
        |
        | message.echo
        v
Flask webhook receiver
        |
        +-- Ed25519 signature and timestamp check
        +-- coexistence lifecycle-state gate
        +-- append-only JSONL audit log

The webhook route is POST /webhooks/whatsapp/echo. Before the body is decoded as JSON, the service verifies the telnyx-signature-ed25519 and telnyx-timestamp headers.

Verify the webhook before parsing it

The sample uses the current Telnyx Python SDK client and its webhook unwrapping helper:

client = telnyx.Client(
    api_key=TELNYX_API_KEY,
    public_key=TELNYX_PUBLIC_KEY,
)


def verify_webhook_signature(headers: dict, raw_body: bytes) -> bool:
    signature = headers.get("telnyx-signature-ed25519")
    timestamp = headers.get("telnyx-timestamp")

    if not signature or not timestamp:
        return False

    try:
        ts = int(timestamp)
    except ValueError:
        return False

    if abs(time.time() - ts) > MAX_TIMESTAMP_SKEW_SECONDS:
        return False

    try:
        client.webhooks.unwrap(
            payload=raw_body.decode("utf-8"),
            headers={
                "telnyx-signature-ed25519": signature,
                "telnyx-timestamp": timestamp,
            },
            key=TELNYX_PUBLIC_KEY,
        )
        return True
    except Exception:
        return False

The bounded timestamp window helps reject replayed requests. A missing, malformed, stale, or invalid signature returns False, and the route responds with 401 without processing the payload.

One SDK detail matters here: the unwrap argument is key, not public_key.

Check the number's coexistence state

A valid webhook does not automatically mean the echo stream is ready to use. The monitor also checks two lifecycle fields:

is_on_biz_app confirms that the number is registered on the WhatsApp Business app.
coexistence_state must be active before the event is written to the audit trail.

In live mode, StateChecker retrieves the configured number from the WhatsApp Business Account:

response = client.whatsapp.business_accounts.phone_numbers.list(self.waba_id)

for number in response.data:
    if number.phone_number == self.phone_number:
        return {
            "phone_number": number.phone_number,
            "is_on_biz_app": number.is_on_biz_app,
            "coexistence_state": number.coexistence_state,
        }

The result is cached for 60 seconds by default, so a burst of echo webhooks does not trigger the same lifecycle request repeatedly. At startup, the service also cross-checks the number through:

client.whatsapp.phone_numbers.retrieve_phone_number(self.phone_number)

DEMO_MODE=true replaces those live requests with simulated active state for local exploration. It does not disable webhook signature verification.

Acknowledge mid-sync events without logging them

If the number is not active, the monitor returns 200 but skips the audit write:

if not state_checker.is_active():
    return jsonify({
        "ok": "skipped",
        "reason": "number_not_active",
    }), 200

This is an intentional processing decision, not a server failure. Returning a success response prevents repeated webhook delivery for an event the application has already evaluated.

For active numbers, the handler records two additional verdicts:

origin_matches_biz_app = origin == "whatsapp_business_app"

cost = message.get("cost")
if cost is None:
    no_charge = True
elif (
    isinstance(cost, dict)
    and cost.get("amount") is None
    and cost.get("currency") is None
):
    no_charge = True
else:
    no_charge = False

Unexpected values are preserved as audit signals rather than silently discarded.

Keep the audit trail useful without storing phone numbers

The sample writes one JSON object per line:

log_entry = {
    "timestamp": datetime.now(timezone.utc).isoformat(),
    "event_id": event_id,
    "event_type": event_type,
    "origin": origin,
    "origin_matches_biz_app": origin_matches_biz_app,
    "text": message.get("text"),
    "message_id": message.get("id"),
    "no_charge": no_charge,
}

message_logger.append(log_entry)

The record intentionally omits from, to, and monetary values. The trail still shows what the application received and how it evaluated the event without becoming a store of customer phone numbers.

JSONL keeps the example easy to run and inspect. A production implementation can retain the same interface while replacing the file with a database or queue and applying the organization's own retention and access policies to message text.

Run the example

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/whatsapp-coexistence-sync-monitor

python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt

cp .env.example .env

Configure the environment with placeholders or your own credentials:

TELNYX_API_KEY=your_telnyx_api_key_here
TELNYX_PUBLIC_KEY=your_telnyx_public_key_here
WABA_ID=your_whatsapp_business_account_id
WHATSAPP_NUMBER=+1555XXXXXXXX
ECHO_LOG_PATH=echo_log.jsonl
DEMO_MODE=true
STATE_CHECK_TTL_SECONDS=60
MAX_TIMESTAMP_SKEW_SECONDS=300
PORT=5000

Then start the Flask service:

python app.py

The health endpoint is available at GET /health. To receive real webhooks, expose the Flask service through a public HTTPS URL, configure that URL in Telnyx, set DEMO_MODE=false, and provide the WhatsApp and Telnyx credentials for your account.

Where to take it next

The example deliberately keeps persistence simple. A production sync monitor could add:

a database-backed audit ledger
alerts when an echo fails origin or cost checks
drift detection between manual echoes and API-side conversation state
per-number dashboards for lifecycle state and skipped events
retention controls and redaction for message content

The key pattern is small: authenticate the event, verify the channel is in a usable state, record only what the application needs, and acknowledge every event according to an explicit policy.

You can inspect the complete implementation in the Telnyx code examples repository.

Resources

Build a Crash-Resistant AI Podcast Production Pipeline on Telnyx Edge

Sonam Gupta — Thu, 15 Oct 2026 22:36:03 +0000

Podcast post-production is rarely one operation. Before an episode can be published, someone must transcribe the recording, identify chapters, write show notes, create promotional copy, and produce a trailer. Each step depends on the one before it, which makes the workflow vulnerable to interruptions.

If processing fails after transcription, should the application transcribe the entire recording again?

I built Show Wrap Desk to explore a more durable approach.

It is a TypeScript application running on Telnyx Edge Compute that turns one podcast recording into:

A speaker-labeled transcript
Timestamped chapters
AI-generated show notes
A short trailer script
An MP3 trailer generated with Text-to-Speech

The workflow is owned by one Stateful Actor per show, so completed stages remain recorded when later work fails.

The pipeline

A producer starts the workflow with one request:

curl -X POST https://<your-worker>/wrap \
  -H "Content-Type: application/json" \
  -d '{
    "show": "engineering-notes",
    "audioUrl": "https://example.com/episode.mp3"
  }'

The endpoint immediately returns an episode ID and a 202 Accepted response while the actor continues processing:

transcribing → chapterizing → notes → wrapped

You can inspect progress at any time:

curl "https://<your-worker>/wrapstatus/<episodeId>?show=engineering-notes"

One Stateful Actor per show

The worker derives a deterministic actor name from the show:

const desk = env.DESKS.idFromName(`show:${show}`);
const result = await desk.onWrap(show, audioUrl);

Every episode submitted for the same show reaches the same WrapDesk actor.

The actor stores its workflow data in embedded SQL tables:

episodes tracks status, errors, and completion times.
chapters stores titles and timestamps.
assets stores transcripts, show notes, and trailer audio.

This gives the workflow a durable ledger without requiring a separate orchestration database.

Stable tasks prevent duplicate pipelines

Each episode is processed by a scheduled task with a stable ID:

await this.schedule(
  0,
  "wrapTick",
  { episodeId },
  { id: `wrap:${episodeId}` }
);

Scheduling the same ID again replaces the pending task instead of creating another copy.

After each successful stage, the actor updates the episode status and re-arms that task for the next stage. If the same audio URL is submitted again while processing is incomplete, the actor finds the existing episode and resumes it.

The individual stages also check for persisted output before doing work. A stored transcript is not transcribed again, existing chapters are preserved, and generated notes or trailer audio are reused.

A failure can repeat the currently executing external operation if it occurred before the result was saved. However, completed and persisted stages are not restarted.

Speaker-labeled transcription

The first stage downloads the recording and submits it to the Telnyx Speech-to-Text API using Deepgram Nova-3:

const form = new FormData();

form.append("file", blob, "episode.mp3");
form.append("model", "deepgram/nova-3");
form.append("response_format", "verbose_json");
form.append(
  "model_config",
  JSON.stringify({
    diarize: true,
    smart_format: true,
    punctuate: true,
  })
);

The response includes word-level timestamps and speaker labels. The actor normalizes those words and stores them as the transcript asset before advancing the episode.

Creating chapters

The application groups consecutive words from the same speaker into turns. A silence longer than 60 seconds begins a new chapter:

const turns = buildTurns(words);
const chapters = splitIntoChapters(turns, 60);

Boundary detection happens locally, so the application does not need an AI request to determine every timestamp.

Telnyx AI Inference then receives short summaries of the resulting sections and returns chapter titles. The sample uses zai-org/GLM-5.3-Flash through the Edge Compute TELNYX binding.

Writing show notes

The next inference request creates Markdown show notes containing:

A summary of the episode
Key takeaways
A pull quote
The generated chapter list

The transcript sent to the model is bounded to keep the example inside a manageable context window. A production implementation should introduce chunking or hierarchical summarization for long recordings.

Generating the trailer

The model selects a promising chapter and writes a trailer script capped at 400 characters.

That script is sent to the Telnyx Text-to-Speech WebSocket:

socket.send(JSON.stringify({ text: " " }));
socket.send(JSON.stringify({ text: script }));
socket.send(JSON.stringify({ text: "" }));

The actor collects base64 audio frames until the service sends:

{ "isFinal": true }

The sample defaults to azure.en-US-AvaMultilingualNeural, but HOST_VOICE can be changed to another supported voice.

The project does not automatically clone the podcast host’s voice from the submitted recording. A custom or cloned voice must be configured separately.

Running the example

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/show-wrap-desk

npm install
cp .env.example .env

Add your Telnyx API key, then generate the bindings and deploy:

npm run types
npm run deploy

Set the generated worker URL as DEPLOY_URL and run:

npm run bootstrap

The included demo uses a public-domain LibriVox recording.

Taking the pattern further

Before using this architecture in production, add authentication, audio-size limits, URL validation, retention controls, workflow telemetry, and human review for generated promotional content.

Large transcripts and audio files should also move to durable object storage instead of being kept directly in actor SQL.

The reusable idea is bigger than podcasting:

Assign each workload a durable actor.
Store the output of every stage.
Track progress with explicit statuses.
Schedule work using stable IDs.
Make each stage safe to attempt again.
Expose status through the same durable ledger.

That pattern can support transcription batches, media processing, document enrichment, report generation, and other long-running AI workflows.

The full source is available here:

https://github.com/team-telnyx/telnyx-code-examples/tree/main/show-wrap-desk

ai #typescript #podcasting #edgecomputing

Build a Self-Refreshing Company Handbook Assistant with TypeScript

Sonam Gupta — Wed, 14 Oct 2026 00:03:55 +0000

A company handbook assistant is easy to prototype.

Embed a document, run a similarity search, pass the closest result to an LLM, and return the answer.

The harder questions appear after the demo:

Who checks whether the asynchronous embedding job finished?
What happens if the worker restarts during indexing?
How does the index stay current when the handbook changes?
Where do you record the answers employees received?
How do you stop the model from answering when retrieval finds nothing relevant?

The policy-recall-line sample handles that entire lifecycle with one durable Stateful Actor per company.

It crawls and embeds a handbook URL, polls the asynchronous indexing job, retries failures, refreshes the source every 24 hours, answers grounded questions, and stores every exchange in an embedded SQL ledger.

No separate vector database, task queue, cron server, or state database is required.

What we are building

The application exposes these primary routes:

POST /company/{company}/source
POST /company/{company}/ask
POST /company/{company}/rpc/audit
POST /company/{company}/rpc/reindexNow
GET  /health

Each company gets its own PolicyLine actor.

That actor owns:

The registered handbook URL
The Telnyx Storage bucket containing its embeddings
Current indexing status
Retry counters
Scheduled polling and reindexing tasks
A SQL ledger containing embedding jobs and policy questions

The architecture looks like this:

Employee question
      |
      v
Edge Compute fetch handler
      |
      | company slug
      v
PolicyLine Stateful Actor
      |
      +--> Telnyx similarity search
      |
      +--> Telnyx-hosted LLM
      |
      +--> Embedded SQL audit ledger

The actor is not only the request handler. It is the lifecycle owner for that company’s policy knowledge.

Route every company to one actor

The public fetch handler extracts the company slug from the URL:

const match = path.match(/^\/company\/([^/]+)(\/.*)?$/);

const company = match[1];
const rest = match[2] ?? "/";

It then derives a deterministic actor ID:

const stub = env.LIB.idFromName(`company:${company}`);

Calling:

/company/acme/ask

always reaches the actor named:

company:acme

A request for another company reaches a different actor with separate state and SQL storage.

The remaining path is passed to the selected actor:

const innerReq = new Request(`https://internal${rest}`, {
  method: req.method,
  headers: req.headers,
  body: req.body,
});

return stub.fetch(innerReq);

This keeps the outer router small. Company-specific behavior and data remain inside the actor that owns them.

Register and embed a handbook

A company registers its handbook through:

POST /company/acme/source

Example request:

curl -X POST \
  https://your-worker.telnyxcompute.com/company/acme/source \
  -H "Content-Type: application/json" \
  -d '{
    "company": "acme",
    "url": "https://acme.example.com/handbook"
  }'

The actor first checks whether that company already has a registered source:

const slug = `company:${company}`;
const state = await this.getState();

if (state.sources[slug]) {
  return this.json(
    { error: "Source already registered" },
    409,
  );
}

It then starts an asynchronous embedding job using the Telnyx URL Embeddings API:

const res = await fetch(
  "https://api.telnyx.com/v2/ai/embeddings/url",
  {
    method: "POST",
    headers: {
      "content-type": "application/json",
      authorization: `Bearer ${key}`,
    },
    body: JSON.stringify({
      url,
      bucket_name: bucket,
    }),
  },
);

The endpoint crawls the URL and stores its embeddings in the configured Telnyx Storage bucket.

The response contains a task ID:

const data = await res.json();
const taskId = data.task_id;

The actor records that task in its embedded SQL ledger:

INSERT OR REPLACE INTO embed_tasks (
  task_id,
  source_slug,
  started_at,
  finished_at,
  status
)
VALUES (?, ?, ?, NULL, 'processing')

It also saves the source in durable actor state:

state.sources[slug] = {
  url,
  bucket,
  taskId,
  status: "indexing",
  lastIndexedAt: null,
  retries: 0,
};

await this.setState({ sources: state.sources });

The registration endpoint returns immediately:

{
  "slug": "company:acme",
  "taskId": "task_123",
  "status": "indexing"
}

Poll the asynchronous job without a separate queue

The embedding job does not finish during the registration request, so the actor schedules a polling task:

await this.schedule(
  POLL_INTERVAL_S,
  "onPollTask",
  { taskId, slug },
  { id: `poll:${taskId}` },
);

POLL_INTERVAL_S is set to 60 seconds in the sample.

The stable ID is important:

poll:<taskId>

If the task is still queued or processing, the actor schedules another check using that same ID:

await this.schedule(
  POLL_INTERVAL_S,
  "onPollTask",
  { taskId, slug },
  { id: `poll:${taskId}` },
);

This makes scheduling an upsert rather than an ever-growing collection of duplicate timers.

The poll handler checks:

GET /v2/ai/embeddings/{taskId}

If indexing succeeds, it updates both actor state and SQL:

source.status = "ready";
source.lastIndexedAt = new Date().toISOString();

await this.setState({ sources: state.sources });

this.ctx.storage.sql.exec(
  `UPDATE embed_tasks
   SET finished_at = ?, status = 'success'
   WHERE task_id = ?`,
  new Date().toISOString(),
  taskId,
);

If the task fails, the actor schedules a retry after 600 seconds:

await this.schedule(
  RETRY_DELAY_S,
  "onRetryTask",
  { slug },
  {
    id: `retry:${slug}`,
    maxRetries: MAX_RETRIES,
  },
);

The demo allows up to three retry attempts before marking the source as failed.

This is where the actor architecture earns its place. The indexing lifecycle is stored alongside the source that owns it, and scheduled tasks are not tied to one short-lived request handler.

Answer a policy question with retrieval

Once the source is ready, an employee can ask:

curl -X POST \
  https://your-worker.telnyxcompute.com/company/acme/ask \
  -H "Content-Type: application/json" \
  -d '{
    "company": "acme",
    "question": "What is the PTO carry-over cap?"
  }'

If the source is still indexing, the actor returns 202:

{
  "error": "Source is still indexing. Try again shortly."
}

When the source is ready, the actor sends the question to the similarity-search endpoint:

const res = await fetch(
  "https://api.telnyx.com/v2/ai/embeddings/similarity-search",
  {
    method: "POST",
    headers: {
      "content-type": "application/json",
      authorization: `Bearer ${key}`,
    },
    body: JSON.stringify({
      bucket_name: bucket,
      query,
      num_of_docs: 3,
    }),
  },
);

The application selects the result with the lowest distance:

export function selectTopHit(
  search: SearchHit[],
): SearchHit | null {
  const hits = search.filter(
    (hit) => hit && Number.isFinite(hit.distance),
  );

  if (hits.length === 0) return null;

  const best = hits.reduce((a, b) =>
    b.distance < a.distance ? b : a,
  );

  if (best.distance > SIMILARITY_FLOOR) {
    return null;
  }

  return best;
}

The sample sets SIMILARITY_FLOOR to 0.9. A lower distance means a closer match.

If no result passes the threshold, the actor returns a controlled response:

I couldn't find a matching policy in the handbook.
Please contact HR for clarification.

It does not ask the model to invent an answer.

The 0.9 threshold is a demonstration default, not a universal production setting. Test it against a representative set of questions and tune it for your content.

Compose the answer with Telnyx AI Inference

When retrieval finds an acceptable chunk, the actor passes only that excerpt and the employee’s question to the model:

const completion =
  await this.env.TELNYX.ai.openai.chat.createCompletion({
    model,
    max_tokens: 700,
    temperature: 0.2,
    messages: [
      {
        role: "system",
        content:
          "You are the company librarian. " +
          "Answer the employee's question using ONLY " +
          "the provided handbook excerpt. " +
          "Be concise and cite the excerpt.",
      },
      {
        role: "user",
        content:
          `Question: ${question}\n\n` +
          `Handbook excerpt:\n${chunk}`,
      },
    ],
  });

The default model is:

moonshotai/Kimi-K3

If the primary model fails or returns empty content, the sample retries with:

zai-org/GLM-5.3-Flash

The inference client is supplied through the Telnyx Edge binding:

this.env.TELNYX.ai.openai.chat.createCompletion()

The application does not place a separate inference provider key in its source code.

Record every answer in SQL

Whether retrieval succeeds or produces the controlled no-hit response, the exchange is written to qa_log:

this.ctx.storage.sql.exec(
  `INSERT INTO qa_log (
     asked_at,
     question,
     answer,
     top_distance,
     source_slug,
     answered
   )
   VALUES (?, ?, ?, ?, ?, ?)`,
  new Date().toISOString(),
  question,
  answer,
  topHit ? topHit.distance : null,
  slug,
  answered ? 1 : 0,
);

The table becomes a lightweight audit ledger:

CREATE TABLE IF NOT EXISTS qa_log (
  id           INTEGER PRIMARY KEY AUTOINCREMENT,
  asked_at     TEXT NOT NULL,
  question     TEXT NOT NULL,
  answer       TEXT NOT NULL,
  top_distance REAL,
  source_slug  TEXT NOT NULL,
  answered     INTEGER NOT NULL
);

You can retrieve the 20 most recent rows through:

curl -X POST \
  https://your-worker.telnyxcompute.com/company/acme/rpc/audit

This makes questions such as “What did the assistant say about travel expenses?” answerable from structured records rather than scattered chat logs.

Refresh the handbook every night

A policy assistant becomes risky when the source changes but the index does not.

When the handbook is first registered, the actor schedules a second stable task:

await this.schedule(
  REINDEX_INTERVAL_S,
  "onReindexTask",
  { slug },
  { id: `reindex:${slug}` },
);

REINDEX_INTERVAL_S is 86,400 seconds, or 24 hours.

When the task runs, it starts a fresh embedding job, resets the retry count, schedules polling for the new task, and re-arms the next refresh:

const taskId = await this.startEmbed(
  source.url,
  source.bucket,
  slug,
);

source.taskId = taskId;
source.status = "indexing";
source.retries = 0;

await this.setState({ sources: state.sources });

During reindexing, /ask returns 202 because the source status is no longer ready.

The project also exposes an RPC for immediate reindexing:

curl -X POST \
  https://your-worker.telnyxcompute.com/company/acme/rpc/reindexNow \
  -H "Content-Type: application/json" \
  -d '{"slug": "company:acme"}'

Run the project

Clone the code and install its dependencies:

git clone \
  https://github.com/team-telnyx/telnyx-code-examples.git

cd telnyx-code-examples/policy-recall-line
npm install

Create your local configuration:

cp .env.example .env

Set your Telnyx API key and create a Telnyx Storage bucket for the handbook.

Authenticate the Edge CLI:

telnyx-edge auth api-key set "$TELNYX_API_KEY"

Before deployment, run the project checks:

npm run typecheck
npm run smoke

Generate the Edge types and deploy:

telnyx-edge types
telnyx-edge ship

Add the deployed worker URL to .env as BASE_URL, then register the configured handbook:

npm run bootstrap

The full setup and API documentation are included in the repository.

Production considerations

This is a focused sample, so several controls should be added before using the pattern with real company policies:

Authenticate /source, /ask, /rpc/audit, and /rpc/reindexNow.
Authorize users against the requested company actor.
Validate handbook URLs before allowing the service to crawl them.
Tune the similarity threshold with an evaluation dataset.
Define retention and deletion policies for the Q&A ledger.
Add monitoring for sources that remain in indexing or failed.
Decide whether questions or answers may contain sensitive employee data.
Add citations that identify the source page or section more precisely.
Consider serving the existing index while a refresh is in progress.
Expand the source model if each company needs more than one handbook URL.

The current sample permits one registered source per company and returns 409 when that source already exists.

Why this is more than a RAG demo

The retrieval pipeline is only one part of the application.

The durable actor also owns:

Source registration
Asynchronous task state
Polling
Retry limits
Nightly refreshes
Inference fallback
The Q&A audit ledger

That is the difference between a document-question endpoint and a service that can keep operating after the first request finishes.

The model answers the question. The actor makes sure there is a current, searchable handbook available when the next question arrives.

Resources

Build a Voice Assistant That Remembers Every Caller

Sonam Gupta — Fri, 09 Oct 2026 17:20:53 +0000

Most voice assistants begin every call with a blank slate. The caller gives their name again. They explain why they called last time. They repeat their preferences. Any context gathered during the previous conversation is trapped in a transcript, CRM record, or database that the assistant cannot immediately use.

In this tutorial, we will build a different pattern: one durable intake dossier per caller. When someone calls, the assistant loads that caller’s history before the conversation begins. It can greet them by name, reference their provider, and include relevant account context. After the call, it files the new reason, summary, and next step back into the same dossier.

Sensitive credentials are encrypted before they reach the assistant, and the complete backend runs as a TypeScript application on Telnyx Edge Compute. The complete sample is available here:

View the per-caller intake code on GitHub

What we are building

The sample uses a fictional healthcare intake workflow, but the underlying architecture is useful anywhere conversations need durable customer context.

The flow looks like this:

Incoming call
    |
    v
Telnyx AI Assistant
    |
    | initialization webhook
    v
Edge Compute router
    |
    | actor ID derived from caller number
    v
One Stateful Actor per caller
    |
    | embedded SQL + durable state
    v
Dynamic variables + encrypted credential

After the conversation:

Post-conversation webhook
    |
    v
Same caller actor
    |
    v
File visit reason, summary, and next step

The example exposes four routes:

POST /webhook/initialization
POST /webhook/post-conversation
GET  /dossier/:phone
GET  /health

The two webhook routes form the main lifecycle. The initialization webhook retrieves context before the call, and the post-conversation webhook saves what happened afterward.

Route each caller to one actor

The key design decision is how actors are identified.

The initialization payload includes telnyx_end_user_target, which represents the caller for this interaction. The application normalizes that value and uses it as the Stateful Actor ID:

const phoneDigits = normalizePhoneDigits(
  event.telnyx_end_user_target
);

const actorId = env.DOSSIERS.idFromName(phoneDigits);
const dossier = env.DOSSIERS.get(actorId);

Every request associated with the same normalized phone number reaches the same actor.

That actor owns the caller’s:

Identity information
Provider preference
Account balance
Visit history
Most recent activity
Filing state

This removes the need for application code that repeatedly queries a shared database, reconstructs a profile, and coordinates simultaneous updates.

The routing key is also the ownership boundary.

A phone number is useful for routing, but it should not automatically be treated as verified identity. Production systems should apply their own authentication and trust policies before revealing sensitive information.

Store relational history inside the actor

Each actor has a private embedded SQL database. The sample creates three tables:

CREATE TABLE IF NOT EXISTS identity (
  patient_name TEXT NOT NULL,
  balance_due TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS prefs (
  provider TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS visits (
  id INTEGER PRIMARY KEY AUTOINCREMENT,
  at TEXT NOT NULL,
  reason TEXT NOT NULL,
  summary TEXT NOT NULL,
  next_step TEXT NOT NULL
);

Why SQL instead of one large JSON object?

Visit history is naturally relational. The application may need the most recent visit, a chronological timeline, or filtered records. SQL provides those operations without requiring the entire history to be loaded and rewritten for every update.

The sample seeds demonstration data for a fictional patient named Sarah with a preference for Dr. Lee. Replace this with your own onboarding, CRM, scheduling, or account-data integration.

The important part is that the data belongs to one actor. It is not a globally shared table containing every caller.

Verify the initialization webhook

The initialization endpoint receives a request from Telnyx before the assistant begins the conversation:

if (
  request.method === "POST" &&
  url.pathname === "/webhook/initialization"
) {
  // Verify signature, parse payload, route to actor.
}

Before using the payload, the application verifies its Ed25519 signature using:

The raw request body
The Telnyx signature header
The Telnyx timestamp header
The public key stored in TELNYX_PUBLIC_KEY

The endpoint fails closed if the public key is not configured or the signature cannot be verified.

That matters because the request determines which caller dossier will be loaded. An unsigned request should never be able to select an identity and retrieve its context.

After verification, the application confirms that the event is an assistant.initialization event and extracts telnyx_end_user_target.

You can learn more about Telnyx webhook signing in the documentation.

Give the assistant runtime context

Once the request reaches the correct actor, handleInitialization() reads the caller’s information and returns two groups of values:

{
  "dynamic_variables": {
    "patient_name": "Sarah",
    "provider": "Dr. Lee",
    "last_visit": "Annual wellness visit",
    "balance_due": "$0.00"
  },
  "encrypted_dynamic_variables": {
    "portal_token": "..."
  }
}

Regular dynamic variables can be referenced by the assistant during the conversation:

Hello {{patient_name}}. I see your preferred provider is {{provider}}.
How can I help you today?

They can also provide concise context without dumping an entire conversation history into the prompt.

Telnyx sends initialization requests with a default timeout of 1.5 seconds. The timeout can be configured up to 10 seconds. This sample uses an 8-second timeout to leave room for initialization and cold starts.

See the Dynamic Variables guide for the payload format and assistant configuration.

Keep credentials out of the prompt

Not every runtime value should become a normal dynamic variable.

Regular dynamic variables can appear in prompts and logs. That makes them unsuitable for API keys, portal credentials, access tokens, or other sensitive values.

For the fictional portal integration, the actor generates a fresh random token and encrypts it using AES-256-GCM:

const tokenBytes = crypto.getRandomValues(new Uint8Array(32));
const nonce = crypto.getRandomValues(new Uint8Array(12));

const ciphertext = await crypto.subtle.encrypt(
  {
    name: "AES-GCM",
    iv: nonce,
  },
  encryptionKey,
  tokenBytes
);

The nonce and ciphertext are combined and encoded as a base64url value. The encrypted result is returned under encrypted_dynamic_variables, not dynamic_variables.

The encryption key must decode to exactly 32 bytes and must be configured in two places:

As the Edge Compute secret PORTAL_ENC_KEY
As the matching Telnyx integration secret portal_enc_key

This lets the assistant use an encrypted credential through the intended integration flow without adding its plaintext value to the conversational prompt.

Telnyx documents this pattern in its Per-Caller Credentials guide.

File the conversation afterward

Once the conversation ends, Telnyx calls the post-conversation endpoint:

POST /webhook/post-conversation

The request contains structured values such as:

{
  "telnyx_end_user_target": "+15551234567",
  "visit_reason": "Prescription refill",
  "follow_up": "Confirm pharmacy details",
  "next_step": "Send request to provider"
}

The endpoint first checks the bearer token stored in DOSSIER_WEBHOOK_AUTH. It then normalizes the caller number and routes the request to the same actor used during initialization.

Inside the actor, the record is written to the visits table:

INSERT INTO visits (
  at,
  reason,
  summary,
  next_step
) VALUES (?, ?, ?, ?)

The demo also checks for an existing record with the same reason and next step before inserting. If the webhook is retried, the actor can return the existing visit ID rather than filing an obvious duplicate.

For a production system, I would use a stable event or tool invocation ID as the idempotency key whenever one is available. Matching text fields is useful for a sample, but it is not a universal deduplication strategy.

Run the project locally

You will need:

Node.js 20 or newer
Telnyx Edge CLI 0.2.2 or newer
A Telnyx API key
A configured Telnyx AI Assistant

Clone the repository and enter the example:

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/per-caller-intake

Install the dependencies:

npm install

Check the TypeScript:

npx tsc --noEmit

Then run the included smoke tests:

npm run smoke

The smoke test covers several security-sensitive paths:

AES-GCM encryption and decryption
Valid Ed25519 signatures
Modified request bodies
Stale timestamps
Incorrect public keys
Initialization payload parsing

Configure the secrets

Generate a 32-byte encryption key:

KEY=$(openssl rand -base64 32 | tr '+/' '-_' | tr -d '=')

Store it in Edge Compute:

telnyx-edge secrets add PORTAL_ENC_KEY "$KEY"

Store the same value in the corresponding Telnyx integration secret as portal_enc_key.

Next, retrieve your Telnyx public key:

PUBLIC_KEY=$(curl -s \
  -H "Authorization: Bearer $TELNYX_API_KEY" \
  https://api.telnyx.com/v2/public_key \
  | jq -r '.data.public')

Add it as another Edge secret:

telnyx-edge secrets add TELNYX_PUBLIC_KEY "$PUBLIC_KEY"

Finally, create a strong token for the post-conversation webhook and save matching copies as:

DOSSIER_WEBHOOK_AUTH
dossier_webhook_auth

The first belongs to your Edge function. The second belongs to the assistant integration that invokes the webhook.

Deploy and connect the assistant

Deploy the application:

telnyx-edge ship

After deployment, use the generated function URL for the assistant’s dynamic variables webhook:

https://your-function.telnyxcompute.com/webhook/initialization

Configure post-conversation processing to call:

https://your-function.telnyxcompute.com/webhook/post-conversation

The sample uses an 8-second initialization timeout:

{
  "dynamic_variables_webhook_timeout_ms": 8000,
  "post_conversation_settings": {
    "enabled": true
  }
}

You can inspect a demonstration dossier through:

GET /dossier/:phone

That route is convenient for development, but it should be authenticated or removed before production use.

Where this pattern fits

The example uses patient intake, but the architecture is not healthcare-specific.

A support line could keep one dossier per customer, including open cases, product ownership, and previous troubleshooting steps.

A property-management assistant could remember the caller’s unit, maintenance history, and access preferences.

A financial-services assistant could load account context after authentication and store structured outcomes from each conversation.

A delivery assistant could maintain one durable customer profile with preferred drop-off instructions and recent shipment issues.

The reusable pattern is:

Derive a stable actor ID from the interaction.
Load only the context needed for the current conversation.
encrypt credentials that should not enter the prompt.
Store the structured outcome after the conversation.
Route the next interaction back to the same actor.

With that lifecycle in place, a voice assistant stops behaving like a stateless demo and starts acting like part of the application.

Production considerations

Before adapting the sample for real customer data:

Authenticate callers before exposing sensitive context.
Define retention and deletion policies.
Collect only the data the assistant needs.
Protect or remove diagnostic dossier routes.
Use stable idempotency keys for webhook retries.
Rotate both copies of shared encryption keys together.
Review consent, privacy, and regulatory requirements for your industry.
Decide whether a phone number is sufficient as an actor key for your threat model.

Resources

Build a Live Supervisor Dashboard for AI Support Calls

Sonam Gupta — Fri, 09 Oct 2026 16:56:54 +0000

Most voice AI quality checks happen after the call ends. A transcript is stored, someone reviews a small sample later, and any mistake has already reached the customer.

What if a supervisor could watch an AI support call while it was happening, send the assistant a private coaching instruction, and join the same conversation when human help was needed?

That is what the Live Support Coach Room example demonstrates. It is a TypeScript application running on Telnyx Edge Compute with:

a live supervisor dashboard
one durable room for each AI conversation
policy-based coaching nudges
a WebRTC softphone for human escalation
a simulator that exercises the workflow without placing a real call

This tutorial focuses on the architecture and the fastest way to explore it locally and on Telnyx Edge.

What the finished flow looks like

The application uses two browser tabs:

The caller simulator creates a conversation and submits caller or assistant turns.
The supervisor dashboard lists active rooms, streams the transcript, shows policy flags, and exposes an escalation button.

Try entering the same account number twice in the simulator. The room detects that the caller has repeated sensitive account information without identity verification and injects a coaching instruction telling the assistant to verify the caller's date of birth.

The same room can then dial a supervisor and join that person to the active AI conversation.

The high-level flow is:

AI Assistant event stream
          |
          v
     AssistRelay
          |
          v
CoachRoom for this conversation
   |          |           |
   |          |           +--> Call Control + WebRTC escalation
   |          +--------------> coaching policy + injected nudge
   +-------------------------> supervisor dashboard WebSocket

Why the event stream is a side channel

The assistant's websocket_settings points to the application's /agents/assist endpoint. Telnyx opens one WebSocket for each conversation and sends events such as:

session.created
conversation.item.created
response.text.delta
telnyx.call.answered
telnyx.call.hangup
session.ended

This socket observes the conversation and can inject text turns, but it does not carry the call media and does not replace the model.

That distinction matters. If the coaching application becomes unavailable, the call continues. The caller does not lose audio because the monitoring surface crashed.

The conversation event stream documentation describes it as a live side channel. It is currently beta, and clients should ignore event types they do not recognize because new types may be added.

The stream also is not a guaranteed transcript archive. Events that occur while the socket is unavailable are not replayed. Use it for live supervision, and use the Conversations API after the call when you need a complete system-of-record transcript.

Configure the assistant with the deployed WebSocket URL:

curl -X POST "https://api.telnyx.com/v2/ai/assistants/$ASSISTANT_ID" \
  -H "Authorization: Bearer $TELNYX_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "websocket_settings": {
      "enabled": true,
      "url": "wss://live-support-coach-room-<id>.telnyxcompute.com/agents/assist",
      "auth_ref": "<your-integration-secret-reference>"
    }
  }'

auth_ref points to an integration secret. Telnyx resolves it and sends the value as a bearer token when opening the WebSocket.

Route every conversation to its own durable room

The application mounts two Agent SDK WebSocket surfaces under /agents:

const handleAgents = mountAgents<Env>((env) => ({
  assist: env.RELAY,
  "coach-room": env.COACHROOMS,
}));

/agents/assist receives the assistant event stream. On session.created, the relay uses the conversation ID to locate that conversation's CoachRoom actor.

Each room owns the state for exactly one conversation:

transcript turns
policy flags
nudge count
stream status
escalation state
a durable silence timer
the final coach log

This is a good fit for a stateful actor: the actor identity is the conversation identity, and related events are handled by the same durable instance.

A separate CoachRegistry actor tracks active and ended rooms for the dashboard picker.

Stream snapshots and live changes to the dashboard

The supervisor connects to:

wss://<function-host>/agents/coach-room/{conversation_id}

When a supervisor opens a room, the dashboard receives its current state and then follows live updates. This means a supervisor joining halfway through a call can see what happened before opening the dashboard instead of starting with an empty screen.

The application also exposes regular HTTP routes:

Route	Purpose
`GET /rooms`	List active and ended rooms
`GET /rooms/{id}/snapshot`	Read the current room state
`GET /rooms/{id}/log`	Read the room's coach log
`POST /rooms/{id}/join`	Escalate to a human supervisor
`GET /health/liveness`	Basic liveness check
`GET /health/readiness`	Configuration readiness check

Inject a coaching nudge

The room watches conversation events and evaluates a deliberately small policy. When a rule fires, it sends a conversation.item.create frame back over the assistant stream.

Conceptually, the frame looks like this:

{
  "type": "conversation.item.create",
  "item": {
    "type": "message",
    "role": "assistant",
    "content": [
      {
        "type": "input_text",
        "text": "Verify the caller's date of birth before sharing account details."
      }
    ]
  }
}

The instruction becomes part of the live conversation context, allowing the assistant to change course without ending or restarting the call.

Two settings constrain the policy:

NUDGE_MAX_PER_CALL=3
SILENCE_SECS=90

The first prevents an overactive policy from flooding one conversation. The second schedules a check-in after extended silence. The silence watcher is cancelled when session.ended arrives.

For a production deployment, the simple rule can be replaced with policies for compliance language, authentication, escalation risk, or objection handling.

Bring a human into the AI conversation

When the supervisor clicks Escalate to supervisor, the room:

creates an outbound call to the configured supervisor device
joins that call leg to the active AI conversation with ai_assistant_join
answers the call in the browser through the Telnyx WebRTC SDK

The softphone answers muted and unmutes after the join succeeds. This avoids exposing setup noise to the caller.

The human joins the existing conversation as another participant rather than receiving a cold transfer. Telnyx's multi-participant call documentation covers the broader pattern for adding people to active AI calls.

Run the simulator first

The included simulator is the quickest way to understand the state flow because it does not require telephony.

Clone and install the project:

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/live-support-coach-room
npm install

Create your environment file:

cp .env.example .env

Authenticate and prepare the Edge function:

telnyx-edge auth api-key set <your_telnyx_api_key>
telnyx-edge new-func -l ts -n live-support-coach-room --from-dir .
telnyx-edge types

Run the smoke test:

npx tsx smoke_test.ts

Deploy it:

telnyx-edge ship

Then open these pages from the deployed function:

https://<function-host>/caller
https://<function-host>/dashboard

Start a simulated call in the caller tab. Enter an account number twice, then watch the policy flag and coaching instruction appear in the dashboard.

Configure the live escalation path

The simulated coaching flow works without a call, but live escalation needs the following Edge secrets:

telnyx-edge secrets add TELNYX_API_KEY "<your-key>"
telnyx-edge secrets add COACH_AUTH "<integration-secret-value>"
telnyx-edge secrets add CALL_CONTROL_CONNECTION_ID "<connection-id>"
telnyx-edge secrets add TELNYX_NUMBER "+1555XXXXXXXX"
telnyx-edge secrets add SUPERVISOR_DEVICE "+1555XXXXXXXX"

Use placeholders in source control and store the real values as secrets. The browser softphone also needs a WebRTC login token supplied by the supervisor at runtime.

Production considerations

This sample demonstrates the architecture, but a production coach room should add:

role-based access for supervisors
short-lived browser credentials
policy versioning and approval workflows
audit retention and deletion rules
metrics for stream gaps and reconnects
explicit consent and transcript handling policies
a post-call source of truth from the Conversations API

The central design still holds: keep observability outside the media path, give each conversation a durable owner, and make human intervention a first-class call participant.

Where to take it next

The same structure can support more than customer service coaching:

compliance monitoring for regulated calls
live sales enablement and objection handling
interpreter or specialist escalation
healthcare scheduling supervision
fraud-review assistance
training dashboards for new agents

You can keep the plumbing and replace the policy inside CoachRoom with the rules that matter to your application.

Resources

Build Persistent Memory for an AI Agent with Python

Sonam Gupta — Wed, 07 Oct 2026 22:14:03 +0000

AI agents are good at responding to the conversation in front of them. The harder problem is remembering something after that conversation ends.

A user might tell a support agent:

their preferred contact method
an account identifier
a product configuration
the issue they previously reported

Without persistent memory, the next session begins from zero and the agent asks for the same information again.

In this tutorial, we’ll build a small Python CLI that sends a conversation to Telnyx Agent Memory, waits for the facts to be extracted, and then recalls those facts with a natural-language query.

The complete example is available here:

github.com/team-telnyx/telnyx-code-examples/tree/main/persistent-ai-agent-memory

What we’re building

The example follows a simple three-step workflow:

Conversation transcript
        |
        v
1. Ingest transcript
        |
        v
2. Poll async operation
        |
        v
3. Recall ranked facts

The CLI:

Submits a support conversation for a profile.
Receives an asynchronous operation ID.
Polls until fact extraction finishes.
Asks what the user’s preferred contact method is.
Prints the matching memories in relevance order.

This pattern gives an agent continuity without placing every previous conversation into its prompt.

The Agent Memory model

Agent Memory organizes information using a few core concepts:

Namespace: an isolation boundary for an application or environment.
Profile: the person or entity the memories describe.
Source: the original session or fact submitted to the API.
Memory: an individual fact extracted from a source.
Operation: an asynchronous write job that can be tracked to completion.

The example uses the default namespace and a profile named user_123.

A real application could use an existing customer, account, or caller identifier as the profile ID.

The three API calls

Every endpoint is relative to:

https://api.telnyx.com/v2/ai/memory

The example uses these calls:

POST /namespaces/{namespace}/profiles/{profile_id}/ingest
GET  /namespaces/{namespace}/operations/{operation_id}
POST /namespaces/{namespace}/profiles/{profile_id}/recall

Let’s walk through each one.

1. Ingest a conversation

First, load the API key and prepare the authentication headers:

import os
import requests
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.environ["TELNYX_API_KEY"]
BASE_URL = "https://api.telnyx.com/v2/ai/memory"

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json",
    "Accept": "application/json",
}

Here is a shortened version of the support transcript:

transcript = [
    {
        "role": "user",
        "content": "Messages are failing for some international destinations.",
    },
    {
        "role": "assistant",
        "content": "Let me check your account configuration.",
    },
    {
        "role": "user",
        "content": "My preferred contact method is email at user@example.com.",
    },
]

Submit the messages to a profile:

namespace = "default"
profile_id = "user_123"
session_id = "demo-session-001"

response = requests.post(
    f"{BASE_URL}/namespaces/{namespace}/profiles/{profile_id}/ingest",
    params={"session_id": session_id},
    headers=headers,
    json={"messages": transcript},
    timeout=30,
)

response.raise_for_status()
operation_id = response.json()["data"]["operation_id"]

print(f"Ingest accepted: {operation_id}")

The API returns 202 Accepted, not a completed memory.

A typical response looks like this:

{
  "data": {
    "operation_id": "op_abc123",
    "profile_id": "user_123",
    "session_id": "demo-session-001",
    "source_id": "src_xyz789"
  }
}

Fact extraction happens asynchronously. The operation_id is the handle we use to track it.

2. Poll until the write finishes

A memory cannot be recalled until its write operation has completed.

The sample checks the operation every two seconds and stops after reaching a terminal state:

import time

terminal_statuses = {"completed", "failed", "cancelled"}

while True:
    response = requests.get(
        f"{BASE_URL}/namespaces/{namespace}/operations/{operation_id}",
        headers=headers,
        timeout=30,
    )
    response.raise_for_status()

    status = response.json()["data"]["status"]
    print(f"Operation status: {status}")

    if status in terminal_statuses:
        break

    time.sleep(2)

The normal progression is:

pending -> processing -> completed

Production code should also enforce a timeout. The complete example stops polling after 60 seconds and raises an error rather than waiting forever.

This explicit operation lifecycle is useful because your application can distinguish between:

a write that is still processing
a write that completed
a write that failed
a write that was cancelled

If the client restarts while polling, it can continue checking the same operation instead of submitting the entire conversation again.

3. Recall relevant facts

Once ingestion completes, query the profile using natural language:

response = requests.post(
    f"{BASE_URL}/namespaces/{namespace}/profiles/{profile_id}/recall",
    headers=headers,
    json={
        "query": "What is the user's preferred contact method?",
        "top_k": 10,
    },
    timeout=30,
)

response.raise_for_status()
facts = response.json()["data"]

The results are ranked by relevance:

for fact in facts:
    print(f"[score={fact['score']}] {fact['text']}")

A result may look like this:

{
  "id": "mem_abc123",
  "text": "The user's preferred contact method is email at user@example.com.",
  "recorded_at": "2026-10-05T12:00:05Z",
  "score": 0.92
}

The response contains the extracted fact rather than requiring the application to search the original transcript itself.

That fact can now be added to an agent’s context when a new conversation begins.

Run the complete example

Clone the repository:

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/persistent-ai-agent-memory

Create a .env file:

echo "TELNYX_API_KEY=your_telnyx_api_key_here" > .env

Install the dependencies:

pip install -r requirements.txt

Run the CLI:

python app.py

The example requires only a Telnyx API key. It does not require a phone number, messaging profile, or separate database.

Details worth handling in production

The sample includes a few useful safeguards that are easy to overlook.

Encode path identifiers

Namespace, profile, and operation identifiers should be encoded before being inserted into a URL:

from urllib.parse import quote

profile_id = quote(profile_id, safe="")

This prevents reserved characters in an existing customer identifier from changing the request path.

Validate inputs before sending them

The example enforces several API constraints:

if len(session_id) > 128:
    raise ValueError("session_id must be <= 128 characters")

if not query or len(query) > 4096:
    raise ValueError("query must be between 1 and 4096 characters")

if not 1 <= top_k <= 100:
    raise ValueError("top_k must be between 1 and 100")

Failing locally gives developers a clearer error than sending an invalid request and debugging it later.

Do not recall immediately after ingesting

An accepted write is not yet a recallable memory.

If recall returns an empty list immediately after ingestion, first confirm that the operation reached completed.

Choose isolation boundaries deliberately

Profiles are isolated from one another, and namespaces provide an additional boundary between applications or environments.

For example:

namespace: production-support
profile: customer_1024

You might use separate namespaces for development and production, while each customer or caller receives a unique profile inside that namespace.

Custom namespaces must be created before use. The default namespace is available without a separate provisioning step.

Plan for memory deletion

Persistent memory should come with a deletion strategy.

The Agent Memory API includes endpoints for deleting an individual source or an entire profile. That allows an application to remove one imported session or erase everything associated with a profile when required.

Where this pattern fits

The same ingest, poll, and recall workflow can support:

Customer support agents that remember previous issues and preferences
Sales assistants that maintain account context across conversations
Scheduling agents that recall availability or communication preferences
Voice agents that recognize returning callers
Internal copilots that retain user-specific working context

The communication channel can change while the memory model stays the same. SMS, voice, email, and browser chat can all resolve to the same profile identifier.

Final thoughts

A longer prompt is not the same thing as long-term memory.

Prompt context is temporary. Persistent memory gives an agent a durable place to store what it has learned and retrieve only the facts relevant to the next interaction.

The implementation here stays intentionally small:

ingest -> poll -> recall

That is enough to move an agent from isolated sessions toward real continuity.

Resources

Build a Multi-Tenant AI Spend Meter with Telnyx AI Gateway

Sonam Gupta — Tue, 06 Oct 2026 19:31:06 +0000

Adding AI features to a multi-tenant product creates an operational question that application-level token counters do not fully answer:

How much did each tenant actually spend, and what stops the next request when the budget is gone?

Your application can estimate usage, but retries, concurrent requests, streaming behavior, and process failures can make that estimate drift away from what the inference layer served.

The open-source ai-gateway-tenant-meter example moves the control point closer to the model request. It combines Telnyx AI Gateway with one durable Edge actor per tenant.

The gateway meters and enforces. The actor remembers and reports.

What the example builds

The TypeScript application exposes a small HTTP API:

POST /provision creates a tenant ledger, gateway token group, and token key
GET /spend?tenantId=... returns current spend, model usage, guardrail findings, alerts, and budget state
POST /adjust changes the tenant budget with optimistic concurrency
GET /health confirms the worker is available

Each provisioned tenant receives:

a 30-day AI Gateway budget
an allowlist of models
prompt and response guardrails
a one-time gateway token key
one durable SpendLedger actor
private SQL tables for daily spend, alerts, and guardrail events
an hourly rollup task
SMS notifications at configurable budget thresholds

By default, the warning threshold is 80% and the hard threshold is 100%. The sample starts with DEMO_MODE=true, so alerts are logged instead of sent while you test the flow.

Provision one gateway scope per tenant

The worker routes a tenant ID to a stable actor identity. The actor then creates a gateway token group with its own budget and guardrail policy:

const group = await client.createTokenGroup(
  tenantId,
  allowedModels,
  monthlyBudget,
);

const key = await client.createTokenKey(
  `${tenantId}-primary`,
  group.id,
);

The token group request includes the controls that should travel with that tenant:

const GUARDRAILS_POLICY = {
  secrets: {
    prompt: "block",
    response: "block",
  },
  dlp: {
    profiles: ["financial"],
    prompt: "flag",
    response: "flag",
  },
  streaming: "buffered",
};

The tenant's application receives a one-time ltg_sk_... key and uses the OpenAI-compatible inference endpoint:

curl https://llm.telnyx.com/v1/chat/completions \
  -H "Authorization: Bearer ltg_sk_..." \
  -H "Content-Type: application/json" \
  -d '{
    "model": "Kimi-K2.6",
    "messages": [
      {"role": "user", "content": "Summarize this support request."}
    ]
  }'

Every request made with that key is associated with the tenant's token group. When the group exhausts its budget, the gateway returns 403 budget_exceeded rather than relying on an application process to notice the overage first.

Keep the operational history in a durable actor

Gateway enforcement answers whether the next inference request may run. The actor answers a different set of questions:

Has the 80% alert already been sent?
Which models generated the spend?
What guardrail detectors fired?
When does the current budget period reset?
Who changed the budget, and from what value?

Each tenant maps to one SpendLedger actor. That actor runs an hourly durable task and writes normalized usage into private SQL.

private armRollup(): Promise<string> {
  return this.every(
    3600,
    "rollup",
    undefined,
    { id: "hourly-rollup" },
  );
}

The SQL update is idempotent by tenant and day:

INSERT INTO spend_days (
  tenant,
  day,
  spend,
  input_tokens,
  output_tokens,
  blocked,
  flagged
)
VALUES (?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(tenant, day) DO UPDATE SET
  spend = excluded.spend,
  input_tokens = excluded.input_tokens,
  output_tokens = excluded.output_tokens,
  blocked = excluded.blocked,
  flagged = excluded.flagged;

If the runtime restarts, the actor's state and SQL history remain available. The next rollup can continue without rebuilding the tenant ledger from scratch.

Store guardrail findings without storing the matched text

The actor reads gateway guardrail events and stores detector metadata such as:

stage
outcome
detector
finding code
count

It does not store the sensitive text that triggered the finding.

That gives an operations team evidence that a secrets or financial-data rule fired without turning the ledger itself into another repository of sensitive prompt data.

Send each budget alert once per period

During the rollup, the actor compares the current spend with the tenant's configured budget.

const pct = pctOf(monthToDate, current.monthlyBudget);

if (pct >= warnPct && !current.alerted80) {
  await recordAlert("80", warningMessage);
}

if (pct >= hardPct && !current.alerted100) {
  await recordAlert("100", hardLimitMessage);
}

The alerted80 and alerted100 flags live in durable actor state. That prevents a restart or repeated hourly rollup from sending the same notification again.

When the gateway reports a new resets_at value, the actor recognizes that a new budget period has started and clears the alert guards.

With DEMO_MODE=false, alerts are sent through the Telnyx messaging binding:

await this.env.TELNYX.messages.send({
  to: this.env.ADMIN_SMS_TO,
  from: this.env.ADMIN_SMS_FROM,
  text: body,
});

Protect gateway mutations from retries and concurrent edits

The example centralizes the AI Gateway wire contract in a GatewayClient.

Every POST, PATCH, and DELETE includes an Idempotency-Key:

headers["Idempotency-Key"] = crypto.randomUUID();

Budget updates first read the token group's current version, then send that version through If-Match:

const group = await client.getTokenGroup(tokenGroupId);

const updated = await client.patchTokenGroup(
  group.id,
  group.version,
  { max_budget: newBudget },
);

If another writer changes the group first, the API returns 412 precondition_failed instead of silently overwriting the newer value.

Run the offline checks first

Clone the example and run its smoke test before using live credentials:

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/ai-gateway-tenant-meter

npm install
cp .env.example .env
npm run smoke
npm run typecheck

The smoke test verifies the gateway request shape, data response envelopes, idempotency headers, If-Match handling, spend calculations, alert guards, and finding normalization without making live API requests.

For a deployed test, configure TELNYX_API_KEY, ADMIN_SMS_FROM, and ADMIN_SMS_TO, keep DEMO_MODE=true, and deploy:

npm run deploy
curl -s https://<your-deployment>/health

Then provision a tenant:

curl -X POST https://<your-deployment>/provision \
  -H "Content-Type: application/json" \
  -d '{
    "tenantId": "demo-tenant",
    "monthlyBudget": 100
  }'

Store the returned token key securely. The tenant's application uses that credential for model requests through https://llm.telnyx.com/v1.

Where this pattern fits

This architecture is useful for platforms that operate AI features for multiple customers, teams, or environments:

SaaS products offering embedded AI assistants
managed-service providers running customer-specific agents
internal AI platforms with departmental budgets
agencies managing inference workloads for clients
development platforms separating staging and production spend

The reusable idea is straightforward: put budget enforcement at the inference gateway, and give each tenant a durable operational ledger that can explain what happened.

Resources

Build a Durable Chargeback Review Workflow on Telnyx Edge

Sonam Gupta — Fri, 02 Oct 2026 21:43:49 +0000

A chargeback is not a single API request. It is a case that can remain open for days while evidence arrives, deadlines approach, and a reviewer decides what should happen next.

That makes chargeback handling an interesting systems problem. The application needs to remember the case, wake up at the right time, evaluate evidence consistently, communicate with the customer, and preserve an audit trail even when the runtime restarts.

The chargeback-adjudication TypeScript example models each dispute as a durable actor running on Telnyx Edge Compute.

The sample combines:

One Stateful Actor per dispute
Telnyx Decision Models for structured evidence evaluation
Scheduled tasks for decision and response deadlines
SQL for evidence, review queues, and an append-only audit ledger
SMS for customer updates and reviewer escalation

The repository uses synthetic evidence and defaults to DEMO_MODE=true, so SMS messages are logged instead of sent while you explore the workflow.

The architecture

The central design choice is simple: the actor is the dispute case.

Chargeback webhook
        |
        v
DisputeCase actor, keyed by disputeId
        |
        +-- SQL evidence and audit rows
        +-- Scheduled decision task
        +-- Telnyx Decision Models request
        +-- Policy and human-review routing
        +-- Customer or reviewer SMS

Because the actor is keyed by disputeId, later evidence is routed back to the same durable case instead of being reconstructed from a stateless webhook request.

Create a case from a webhook

The Edge handler accepts a chargeback event at POST /webhook/chargeback:

if (path === "/webhook/chargeback" && req.method === "POST") {
  const payload = parseJson<ChargebackPayload>(await req.text());

  if (!payload || !payload.disputeId || !payload.customer || !payload.orderId) {
    return new Response(
      JSON.stringify({
        error: "disputeId, customer, and orderId required",
      }),
      { status: 400 },
    );
  }

  const stub = e.DISPUTES.idFromName(payload.disputeId);
  const result = await stub.onChargeback(payload);

  return new Response(JSON.stringify(result), { status: 200 });
}

idFromName() gives every dispute a stable actor identity. Inside onChargeback(), the actor stores the case state, seeds the sample evidence tables, calculates the response window, and schedules a decision task with a stable ID:

this.schedule(0, "decide", {}, {
  id: `decide:${disputeId}`,
});

Stable task IDs are useful when webhook delivery or application logic retries. Repeated work can converge on the same scheduled operation instead of creating a new timer every time.

Evaluate several questions in one Decision Models request

The actor assembles the order, delivery record, contact history, and any newly submitted evidence into one shared state object.

It sends that context to:

POST /v2/ai/typesafe/v1/systemone

One request asks three named questions:

const body = {
  state: JSON.stringify(evidence),
  questions: {
    decision: {
      type: "choice",
      instructions: "Rule on the chargeback.",
      criteria: {
        approve_rebate: "Delivery evidence supports the customer's order.",
        request_evidence: "Evidence is inconclusive; more proof is needed.",
        deny: "Evidence supports the merchant; deny the dispute.",
      },
    },
    loseProb: {
      type: "score",
      instructions: "Rate the merchant's risk of losing.",
      criteria: [
        "Clearly likely to win",
        "Uncertain",
        "Clearly likely to lose",
      ],
    },
    fraud: {
      type: "noul",
      instructions: "Does this look like a fraud attempt?",
    },
  },
};

These question types serve different purposes:

choice selects one supplied category.
noul produces a signal between zero and one.
score returns an expected position on the ordered rubric.

A Decision Models score should not automatically be treated as a calibrated probability. With three rubric entries, the result falls along the rubric’s zero-to-two range and can be fractional.

Production policy should be tested against representative, labeled cases rather than treating the result as certainty.

The endpoint defaults to telnyx/decision-flash when model is omitted. You can also set the model explicitly based on the context requirements of your workflow.

Keep model output separate from business policy

The model produces structured evidence signals. Application code still owns the business action.

The sample’s policy layer can:

Approve a rebate and notify the customer
Request additional evidence
Deny the claim
Route a suspected-fraud case to a human review queue

if (noul > FRAUD_THRESHOLD) {
  await db
    .prepare("INSERT INTO reviewQueue VALUES (?, ?, ?)")
    .bind(disputeId, new Date().toISOString(), "fraud_hold")
    .all();

  await sendSms(
    reviewerOnCall,
    `Review required for ${disputeId}`,
  );

  return;
}

The threshold in this sample is illustrative. Before using a similar rule in production, evaluate it against your own data, add authorization and reviewer controls, and clearly define which outcomes may be automated.

Decision Models help determine what deserves attention. They do not remove the human reviewer from high-risk decisions.

Re-evaluate when new evidence arrives

A dispute can change after its initial review.

The second webhook route, POST /webhook/inbound-message, sends an SMS or MMS response back to the existing actor:

async onNewEvidence(
  text: string,
  mediaUrl?: string,
): Promise<void> {
  const evidence = await this.assembleEvidence(mediaUrl);

  const result = await this.judgeWithDecisionModel({
    ...evidence,
    newEvidence: text,
  });

  await this.appendAudit("re-evaluated", result);
  await this.applyPolicy(result);
}

The case keeps its identity and previous state. New evidence becomes another event in the same lifecycle, and the resulting evaluation is appended to the ledger.

Preserve an audit trail in SQL

Every material event is written as a new row rather than overwriting the previous decision:

await db
  .prepare("INSERT INTO audit VALUES (?, ?, ?, ?)")
  .bind(
    disputeId,
    new Date().toISOString(),
    event,
    JSON.stringify(payload),
  )
  .all();

This append-only pattern makes it possible to reconstruct:

What evidence the application had
Which model output it received
Which policy branch was selected
When the state changed
Whether new evidence caused a re-evaluation

For a real financial or healthcare workflow, this example is only a starting point.

Add authentication, webhook verification, encryption, retention controls, idempotency around external actions, reviewer permissions, monitoring, and a domain-specific compliance review before processing sensitive data.

Run the sample

Clone the examples repository and install the dependencies:

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/chargeback-adjudication

npm install
cp .env.example .env

The sample configuration includes:

TELNYX_API_KEY=your_telnyx_api_key_here
RESPONSE_DEADLINE_DAYS=7
REVIEWER_ONCALL_E164=+1555XXXXXXXX
DEMO_MODE=true

Authenticate the Edge CLI, run the smoke test, and deploy:

telnyx-edge auth api-key set <your_telnyx_api_key>
npx tsx smoke_test.ts
telnyx-edge ship

DEMO_MODE=true suppresses real SMS delivery and uses synthetic evidence. The deployed adjudication path still requires a Telnyx API key to call Decision Models.

Where else this pattern fits

The reusable idea is larger than chargebacks: assign one durable actor to every long-running case, keep its evidence and history nearby, request structured model outputs, and let explicit application policy determine what happens next.

The same architecture could support:

Insurance claim review
Returns and refund exceptions
Account-verification cases
Compliance investigations
Support escalations with deadlines

Resources

Build Automated Post-Call QA Scoring with Telnyx Decision Models

Sonam Gupta — Fri, 02 Oct 2026 20:36:24 +0000

Support teams have a math problem.

Thousands of customer calls happen every week, but supervisors can manually review only a small fraction of them. Even when a call is graded, that score often ends up isolated in a spreadsheet. It cannot tell you whether an agent had one difficult conversation or is developing a pattern.

That creates two distinct engineering problems:

Coverage: evaluate more completed calls without requiring someone to listen to each recording.
Memory: preserve each agent's results long enough to identify trends and make better coaching decisions.

The open-source post-call-qa-scoring example addresses both. It uses Telnyx Decision Models to grade transcripts and a durable QAAgent actor to maintain the quality history for each support agent.

The result is not an autonomous disciplinary system. It is a way to surface calls and trends that deserve human attention.

What the application does

When a support call ends, the application:

receives the transcript from a call-conversation-ended webhook
uses transcription-saved as a fallback when the transcript is not embedded
routes the transcript to a durable actor keyed by the support agent ID
asks a Telnyx Decision Model for three structured grading signals
stores the result in private per-actor SQL
recomputes a five-call rolling average
flags or clears a coaching recommendation based on that trend
escalates potential compliance breaches to a manager over SMS
sends the team lead a daily per-agent digest

Each call ID is stored as a SQL primary key before grading is scheduled. The scheduled task also uses a stable ID, grade:<callId>, so webhook retries do not create duplicate grades.

Why use one actor per support agent?

The central design decision is this:

The actor is the agent's quality profile.

The worker resolves an actor with idFromName(agentId). Every later transcript for that support agent reaches the same durable actor.

const actorId = env.QA_AGENT.idFromName(agentId);
const qaAgent = env.QA_AGENT.get(actorId);

await qaAgent.recordCallEnded(
  callId,
  transcript,
  agentId,
  digestEnabled,
);

That actor owns the score history, rolling average, coaching state, breach records, and digest schedule. The application does not have to reconstruct the agent's history from a cache every time another call ends.

Grade the transcript into structured fields

Free-text model output is awkward for QA automation. If the model returns a paragraph, downstream code has to interpret that prose before it can calculate an average or apply an escalation threshold.

The example instead calls the Telnyx Decision Models endpoint:

POST /v2/ai/typesafe/v1/systemone

One request evaluates the transcript against three question types:

choice: pass or a specific failure category
noul: a 0–1 compliance-breach signal
score: an overall quality score from 0–5

The application can then apply ordinary program logic to typed values:

interface DecisionResult {
  choice: string;
  noul: number;
  score: number;
}

A call can pass its general quality evaluation and still carry a serious compliance signal. The sample therefore treats these fields independently. When noul > 0.8, it records the call in the breaches table and sends an immediate manager-review alert regardless of the pass/fail choice.

Decision Models do not replace the reviewer here. They decide which evidence should be placed in front of one.

Turn individual grades into a trend

After every successful grade, the actor queries its SQL history and recomputes a five-call rolling average.

SELECT ts, choice, noul, score, status, last_error
FROM scores
WHERE status = 'graded'
ORDER BY ts DESC
LIMIT 5;

If the average falls below QA_COACHING_FLOOR, the actor flags the agent for coaching. If later calls bring the average back above the threshold, it clears the flag automatically.

That distinction matters. One low score may be an isolated call. A declining rolling average is a stronger reason to investigate, coach, or review the underlying conversations.

Keep failures visible

The grading task uses bounded retries rather than retrying forever. Transient failures back off for 10, 30, and 60 seconds. If grading still cannot complete, the call is stored with status="ungraded" and a last_error value.

The daily digest can therefore distinguish between a low score and a transcript that was never successfully graded. Failed AI work does not quietly disappear from the reporting pipeline.

Try the complete pipeline without placing a call

The repository includes a synthetic demo endpoint. After deploying the example, call:

curl -X POST https://<edge-function-url>/demo/trigger \
  -H "Content-Type: application/json" \
  -d '{
    "agentId": "demo-agent",
    "callId": "demo-001",
    "transcript": "Agent: This call may be recorded for quality purposes..."
  }'

The endpoint routes the transcript through the same actor, Decision Models request, SQL history, trend calculation, and escalation logic used by the webhook flow. It is intended for demonstration and testing, so you can inspect the architecture before connecting a live call application.

Run the example

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/post-call-qa-scoring

npm install
cp .env.example .env
npm run typecheck
npm test
npm run deploy

For live Decision Models calls and SMS notifications, configure the Telnyx API key, a messaging-enabled Telnyx number, and the team lead's E.164 number as Edge secrets. The repository README documents every supported setting, including the coaching floor and digest hour.

Where to take this pattern next

The sample uses a regional-bank support scenario, but the architecture is useful anywhere conversations need consistent review over time:

contact-center coaching
healthcare communication audits
insurance disclosure checks
sales-call quality monitoring
customer-service escalation review

The important pattern is broader than call scoring: produce structured decisions, keep durable history around the entity being measured, and route high-stakes outcomes to a human.