DEV Community: Harpreet Singh Seehra

Monitor Call Quality in Real Time with Telnyx Webhooks and Python

Harpreet Singh Seehra — Wed, 09 Sep 2026 11:06:00 +0000

Call quality is not a single number you check once. It is a stream of measurements — MOS, jitter, latency, packet loss — that arrive while a call is active and that you need to inspect afterward. This sample puts that stream into a Flask dashboard with live SSE updates, SQL-backed history, and threshold alerting, all verifiable without a Telnyx account.

I work with Telnyx. The code is available here: https://github.com/team-telnyx/telnyx-code-examples/tree/main/call-quality-monitor

What the sample does

The Call Quality Monitor receives Telnyx call quality webhooks, verifies each payload's Ed25519 signature, and stores the metrics in SQLite for historical analysis. An in-memory key-value store tracks per-call state so the dashboard can show what is happening right now. When MOS drops below a threshold or jitter exceeds a limit, an alert fires and is pushed to the browser over a Server-Sent Events stream.

The dashboard at / shows live metrics and alerts. The API exposes filtered queries by call ID, time range, and aggregate statistics. A demo server (demo/demo_server.py) generates an Ed25519 keypair on first run, signs webhook payloads, and posts them to the real /webhooks/call-quality endpoint — the full pipeline runs end-to-end on localhost:5555 without Telnyx credentials or ngrok.

Architecture

Telnyx Voice
    │
    │  Call quality webhooks (MOS, jitter, latency, packet loss)
    ▼
Flask App (app.py)
    │
    ├── Ed25519 signature verification (pynacl)
    │
    ├── Process quality metrics
    │   ├── SQLite INSERT (historical analytics)
    │   └── In-memory KV update (per-call state)
    │
    ├── Threshold checker
    │   ├── MOS < 3.5  → alert
    │   ├── Jitter > 30ms  → alert
    │   └── Latency > 150ms  → alert
    │
    └── SSE broadcast → browser dashboard

Run the demo locally

No Telnyx account is needed. The demo server generates its own Ed25519 keypair, signs webhook payloads, and posts them to the real Flask endpoint.

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/call-quality-monitor

pip install -r requirements.txt
pip install requests  # needed by the demo webhook sender

python demo/demo_server.py

Open http://localhost:5555/ and use the demo controls:

Start 3 Calls — Simulates three concurrent calls with varying quality metrics. One call is healthy, one degrades over time, and one starts bad. The dashboard updates live via SSE.
Trigger Alert — Pushes a metric that crosses the MOS threshold and watch the alert appear in real time.
Reset Data — Clears the SQLite database and in-memory state.

The demo server runs on port 5555 and hosts both the Flask app and a webhook sender thread that posts signed payloads to /webhooks/call-quality. The Ed25519 keypair is generated on first run and stored in a local file. No external service is contacted.

How webhook verification works

Telnyx signs every webhook payload with an Ed25519 key. The Flask app verifies each signature using pynacl before processing the payload. This prevents forged webhook deliveries from injecting fake quality metrics into your dashboard.

import nacl.signing
import nacl.encoding

verify_key = nacl.signing.VerifyKey(
    base64.b64decode(public_key_b64),
    encoder=nacl.encoding.RawEncoder
)
verify_key.verify(signed_payload, signature)

The demo server generates its own keypair on first run, signs payloads with the private key, and the Flask app verifies them with the corresponding public key. In production, Telnyx holds the private key and you configure the public key from the Portal.

Why SSE instead of WebSocket

Server-Sent Events are unidirectional (server to browser), which is all a monitoring dashboard needs. SSE runs over standard HTTP, works through proxies without extra configuration, and reconnects automatically. The browser's EventSource API is simpler than WebSocket and sufficient for a read-only live feed.

Connect to live Telnyx webhooks

For production use, configure your Telnyx webhook URL in the Portal:

https://<your-domain>/webhooks/call-quality

Set these environment variables:

Variable	Description	Example
`TELNYX_API_KEY`	Your Telnyx API key	`KEY019...`
`TELNYX_PUBLIC_KEY`	Your Telnyx public key (Ed25519)	`-----BEGIN PUBLIC KEY-----...`
`DB_PATH`	SQLite database path	`quality.db`
`MOS_THRESHOLD`	Minimum acceptable MOS score	`3.5`
`JITTER_THRESHOLD`	Maximum jitter in ms	`30`
`LATENCY_THRESHOLD`	Maximum latency in ms	`150`
`PORT`	Flask server port	`5000`

API endpoints

Method	Endpoint	Description
`POST`	`/webhooks/call-quality`	Receive and verify Telnyx call quality webhooks
`GET`	`/`	Live SSE dashboard
`GET`	`/events`	SSE stream endpoint
`GET`	`/api/quality/<call_id>`	All metrics for a specific call
`GET`	`/api/quality`	All metrics with optional filters
`GET`	`/api/quality/stats`	Aggregate statistics
`GET`	`/api/quality/alerts`	All threshold alerts
`GET`	`/health`	Health check

Testing

python -m pytest smoke_test.py -v

9 tests covering webhook verification, metric storage, threshold alerting, API endpoints, and SSE streaming. All pass.

Build a Call, SMS, and Webhook Scheduler with the Telnyx Agent SDK

Harpreet Singh Seehra — Tue, 08 Sep 2026 14:16:27 +0000

This walkthrough uses a local TypeScript scheduler with a browser dashboard. Calls, messages, and outbound webhooks are simulated by default; scheduling and saved execution records are real.

I work with Telnyx. The code is available here: https://github.com/team-telnyx/telnyx-code-examples/tree/main/edge-cron-scheduler

How the Agent SDK, KV, and SQL fit together

One named scheduler actor owns the job registry. Job definitions are stored in actor key-value storage. Each execution receives a separate SQL record, so a new run does not overwrite the result of the previous run.

The Agent SDK registers a recurring polling task with every(60, "poll"). When the poller finds a due job, it dispatches execution through queue("execute", task). Cron parsing determines the next matching calendar time; the polling interval determines when the application checks for work.

Dashboard or authenticated HTTP client
                 |
          CronAgent("scheduler")
          |                  |
     KV job registry     recurring poll
                             |
                       queued execution
                             |
                 SQL claim and outcome
                             |
                    call / SMS / webhook

The local runner uses the same Agent SDK with a disk-backed SQLite adapter. It serializes HTTP and alarm turns, stores the registry and execution history, and resumes saved timers after restart. Its role is to host the sample locally. The walkthrough demonstrates that concrete configuration and does not imply that a browser tab is responsible for running the scheduler.

Run the demo locally

Use Node.js 22.13 or newer. The sample includes a lockfile, so npm ci installs the recorded dependency versions.

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/edge-cron-scheduler
npm ci
cp .env.example .env
npm start

Open the dashboard address printed in the terminal. With an automatically generated local token, the startup link includes a fragment that connects the current browser tab. The page removes that fragment from the address bar. If SCHEDULER_TOKEN is configured, enter it in the connection form instead; the server does not print the configured credential.

The default mode simulates outbound calls, messages, and webhooks. The scheduler, stored jobs, and execution history still run. No Telnyx account credentials are needed for this local simulation. Read the visible mode label before interpreting a result: a simulated success demonstrates the workflow, not delivery to a real recipient.

The server binds to loopback. Keep the generated local database in its ignored data directory, and keep account configuration in the ignored environment file. The repository contains example destinations rather than customer contact data.

Create a daily reminder in the dashboard

Start with Load demo jobs. The dashboard creates examples for a customer check-in call, a team SMS reminder, and a webhook heartbeat. Their every-minute schedules make automatic execution visible during a short walkthrough.

Choose Create job to add a reminder of your own. Enter a name, select SMS, supply an example destination and message, then choose the daily schedule. The expression 0 9 * * * means 09:00 UTC each day. Saving the job creates its first future scheduled occurrence; it does not immediately send a message.

The next-due field is useful when checking that a schedule matches the intended clock time. All schedules use UTC. Five fields are accepted; seconds and per-job timezone overrides are not part of this interface. For a weekly or less frequent action, use the custom cron input and inspect the resulting next run before leaving it enabled.

A manual run is a separate action. Run now queues one eligible job, while Run all queues the jobs that have no dependencies. This makes it possible to demonstrate execution without waiting until the next daily occurrence.

Use the authenticated HTTP API

The dashboard uses the same API available to scripts. Export the scheduler token in your own terminal before making authenticated requests. The placeholder below must be replaced locally; it is not a shared credential.

export SCHEDULER_TOKEN='paste-your-local-token'
curl -sS http://127.0.0.1:8787/jobs   -H "Authorization: Bearer $SCHEDULER_TOKEN"   -H 'Content-Type: application/json'   -d '{"id":"daily-reminder","name":"Daily reminder","cron":"0 9 * * *","type":"sms","target":"+18005550102","payload":{"text":"Your daily check-in is ready."}}'

Queue that job manually with POST /jobs/daily-reminder/run, then inspect GET /logs. A manual request returns an accepted response with a run identifier. Read the execution history to determine the eventual outcome; accepting the queue request is not the same as completing the work.

The public root route serves only the HTML shell. Data and management routes require the bearer token. The browser keeps its token in session storage for the current tab, and Disconnect clears the token and displayed job data.

Read execution history and demonstrate a failure

Each execution records an identifier, scheduled time, start and finish times, status, result, and notification outcome. Open Details to inspect those fields. Filtering to failed runs is useful when a busy schedule has produced several successful executions around a single error.

The Try a failure control creates and queues a demo webhook job that intentionally fails. The result records the simulated failure and a simulated SMS alert. This provides a repeatable way to demonstrate the error path without contacting a real endpoint or sending a real notification.

The notification has its own outcome because failure reporting can fail too. A webhook error and an unsuccessful alert attempt are two facts worth retaining. The sample attempts an alert once; it does not include a separate service that retries alerts indefinitely.

For calls and messages in live mode, a success result means the API accepted the request. The sample does not consume final delivery webhooks. An answered call, delivered SMS, or human response would need its own downstream tracking.

Understand repeats, restarts, and dependencies

A stable run identifier and a SQL claim suppress repeated execution of the same queued task. This does not establish exactly-once effects across the network. If the process stops after recording a running execution, the external outcome may be uncertain. Inspect that record before requesting another attempt.

After downtime, an overdue job runs once for its oldest pending slot and skips intermediate missed slots. That avoids a burst of catch-up calls or messages on restart. It also means this design is not a complete replay system for every missed occurrence.

Dependencies are supported for jobs with the same schedule. A dependent job requires its parent to succeed for the same scheduled slot; a failed dependency produces a skipped execution. Manual runs are restricted to jobs without dependencies. These rules give the sample a small, explicit execution model that can be reasoned about from its stored records.

Resources

Close the Migration-Visibility Gap: SMS Alerts for Failed SQL Migrations in 280 Lines of Python

Harpreet Singh Seehra — Tue, 08 Sep 2026 11:05:26 +0000

The first time a database migration broke on me in production, I found out from a user.

Not from a dashboard, not from a Slack alert, not from anything I built. A user filed a ticket saying the orders page was throwing a 500. I SSH'd in, tailed the logs, found the migration that ran at 02:00, and saw it had failed two hours earlier. The schema version had not bumped. The deploy had "succeeded" because the migration runner swallowed the exception. Nobody looked.

That hour — the gap between the migration failing and someone knowing — is the whole problem. Not the migration itself. Migrations break. That's a fact of life. The bug is that nobody finds out fast enough to roll it back before users see it.

So I built the thing I wished I'd had that night. It is a 280-line Python + Flask service that runs schema migrations, checks the result, and text-messages the on-call the instant a step fails — over Telnyx SMS, with Ed25519-signed webhook delivery receipts so you know the alert actually reached the device.

The code is open source and cloneable: https://github.com/team-telnyx/telnyx-code-examples/tree/main/sql-migration-agent

What the Agent Does

The agent models a migration as a single in-code-path operation. It reads the current schema version from a SQL database. It fetches the migration script. It executes the steps in order. If every step succeeds, it bumps the schema version. If any step fails, it does not bump the version, it rolls back, and it sends an SMS to the on-call number the moment the migration breaks.

The key SDK calls in the Telnyx Python SDK v4:

import telnyx

client = telnyx.Telnyx(
    api_key=os.getenv("TELNYX_API_KEY"),
    public_key=os.getenv("TELNYX_PUBLIC_KEY"),
)

# Fire the failure alert the moment a step breaks
client.messages.send(
    from_=os.getenv("TELNYX_FROM_NUMBER"),
    to=notify_phone,
    text=f"MIGRATION FAILED: {migration_id} — {error}",
)

# Verify the signed delivery receipt that comes back
raw_body = request.get_data(as_text=True)
event = client.webhooks.unwrap(payload=raw_body, headers=request.headers)

The webhooks.unwrap call does Ed25519 signature verification against the public key set on the client at construction. If the signature does not verify, the request is rejected with a 400 — no delivery is recorded, no status is updated.

Why the Webhook Roundtrip Matters

Most alerting systems I've worked with treat "I sent the alert" as the end of the story. This one does not. Telnyx sends a signed webhook back when an SMS is delivered, and the app verifies the signature before recording the delivery. If verification fails, the request is rejected — the app does not record a delivery it cannot prove happened.

The difference between "I sent the alert" and "the on-call saw the alert" is the difference between an alerting system and an alert-hoping system. In production, the device could be offline, the carrier could drop the message, the SIM could be swapped. The signed webhook roundtrip is what closes that last loop.

Why Ed25519 and Not HMAC

Most webhook signature verification in the wild is HMAC-SHA256. You share a secret with the sender, the sender signs the payload, you verify with the same secret. It works, but it creates a key-distribution problem in any multi-service setup. Every service that verifies needs the secret. Rotating the secret means coordinating across all of them.

Ed25519 is asymmetric. The verifier only needs the public key. The sender keeps the private key. You can pin the public key in a config map, share it across services, put it in a README — it does not matter who sees it, because seeing the public key does not let you forge signatures.

For an alerting system that talks to a webhook endpoint you control, this matters less. For an alerting system that other services also verify — say, a multi-tenant setup where each tenant has their own webhook — it matters a lot.

Run It Locally — No Telnyx Account Needed

The demo launcher (demo/demo_server.py) is a single file that stands up a SQLite store, stubs the SMS client, generates an Ed25519 keypair on first run, and serves a dashboard at http://localhost:5555. The whole loop runs in-process — no network calls, no credentials.

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/sql-migration-agent
cp .env.example .env
pip install -r requirements.txt
python demo/demo_server.py

Open http://localhost:5555. The dashboard boots with zero migrations and a baseline schema version. Click "Run 001: Add users table" for the happy path — the migration succeeds, the schema version bumps to 001, no SMS fires because nothing broke. Then click "Run 003: Add orders (FAIL)" — the migration has a deliberate syntax error, the schema version does not advance, and a new entry appears in the SMS log: the on-call just got a text reading MIGRATION FAILED: 003 — syntax error near column.

To verify the webhook roundtrip, click "Send webhook (delivery receipt)". The demo server signs a payload with the Ed25519 key it generated on startup, POSTs it to its own /webhooks endpoint, and the app verifies the signature before recording the delivery. The SMS log row updates to delivered with a green check. That is the full loop — migration runs, fails, SMS fires, signed webhook comes back, app verifies, status lands in the log. Every step that would happen in production, happening on your laptop.

To wire real SMS, drop your Telnyx API key, public key, messaging profile number, and a destination phone number into .env. The stubbed client is replaced automatically.

Where This Fits

Three places this pattern earns its keep:

Scheduled production migrations — Run them on a cron, get a text the second any step fails. No polling, no dashboards to babysit. The on-call either hears "all good" or hears "migration 003 broke" — never silence.
CI/CD gates — Wire this into your deploy pipeline so a failing migration blocks the release and pages the on-call before bad code reaches prod. The migration either succeeds and the deploy proceeds, or it fails and the on-call is notified before any user is.
Multi-tenant SaaS — Run a migration per customer database and get per-tenant failure alerts, so you know exactly which tenant's schema broke and can roll them back individually without affecting the others.

What I Left Out (and Why)

The sample ships a real Ed25519 roundtrip and a real SMS send path (when wired with credentials), but it deliberately stubs a few things to keep the demo runnable on a laptop:

SMS send is stubbed in the demo. The demo launcher replaces telnyx_client.messages with an in-memory recorder. The signed webhook roundtrip is real — the demo generates its own keypair and signs the delivery receipt — but the SMS itself is captured to a local log, not sent over the air.
Migration scripts come from an in-memory map. In production, you would fetch them from a shared store. The Telnyx CloudFS API is the natural fit — same API key, same platform.
Schema versioning is in-memory. The demo uses a Python dict for schema versions. In production, use a real SQL DB — the schema_versions table is the source of truth.

These stubs are intentional. They keep the demo under 280 lines, runnable with no credentials, and focused on the part that matters: the failure-to-notification loop.

What I'd Add Next

Two things I'd add for production:

Per-tenant routing. Right now the on-call number is a single env var. In a multi-tenant setup, the alert should go to the team that owns the tenant. A tenant_to_oncall map, looked up before the SMS send, takes about 15 lines.
Retry with backoff. If the SMS send itself fails, the agent should retry with exponential backoff and then escalate to a fallback channel. Maybe 30 lines.

Neither is in the sample, because neither is the point. The point is closing the visibility gap. The hour between "migration failed" and "someone knew" is the bug. This 280-line agent fixes it.

Build a Webhook Aggregator with Fanout in Flask and the Telnyx Python SDK

Harpreet Singh Seehra — Tue, 01 Sep 2026 12:38:06 +0000

Webhook endpoints have a way of starting simple and becoming a problem. The first version is a Flask route that receives a POST, does something, and returns 200. Then Telnyx redelivers an event because your server was slow, and the same call gets answered twice. Then a burst of call and SMS events lands in the same second, and your business logic blocks on the network. Then you need to explain what happened at 3am, and your logs are not enough.

This sample is a small, deliberate answer to that. It takes raw Telnyx webhooks and runs them through a six-stage pipeline: receive, verify, dedup, log, fan out, process. Your business logic never sees a duplicate, never sees a tampered event, and never blocks on the network.

Code: https://github.com/team-telnyx/telnyx-code-examples/tree/main/webhook-aggregator-fanout

Architecture

Telnyx webhook POST
        |
        v
[1] Receive   — Flask route reads raw body + headers
        |
        v
[2] Verify    — Ed25519 signature check via Telnyx SDK v4
        |
        v
[3] Dedup     — TTL-based in-memory KV store (default 300s)
        |
        v
[4] Log       — SQLite insert (INSERT OR IGNORE for race safety)
        |
        v
[5] Fan out   — Route to call queue or SMS queue by event type
        |
        v
[6] Process   — Drain queue: answer call + play greeting, or reply to SMS

The entire pipeline lives in one file: app.py. No Redis, no Celery, no message broker. The storage primitives are intentionally boring — SQLite for the audit log, a Python dict for the dedup KV — so you can run the whole thing locally without infrastructure.

Verify before you trust

The first thing the webhook handler does is verify the Ed25519 signature. The Telnyx Python SDK v4 makes this a single call:

import telnyx
from telnyx.lib.webhooks_ed25519 import unwrap_with_ed25519

telnyx_client = telnyx.Telnyx(
    api_key=os.getenv("TELNYX_API_KEY"),
    public_key=os.getenv("TELNYX_PUBLIC_KEY"),
)

@app.route("/webhooks", methods=["POST"])
def webhook_handler():
    raw_body = request.get_data(as_text=True)
    verified_event = unwrap_with_ed25519(
        telnyx_client, raw_body, request.headers
    )

unwrap_with_ed25519 reads the Telnyx-Ed25519-Signature and Telnyx-Ed25519-Timestamp headers, verifies the signature against your Telnyx public key, and returns a typed UnwrapWebhookEvent. If the signature is invalid or the timestamp is outside the replay window, it raises — and the handler returns 401 before any business logic runs.

This matters because the public key is not a secret. Anyone who intercepts the webhook traffic can read it. What they cannot do is produce a valid signature without your Telnyx private key. Verification is the difference between trusting the payload and trusting the source.

Deduplicate with a TTL KV store

Telnyx retries webhook delivery if your server does not respond within the timeout. That means the same event can arrive two, three, or five times within a few minutes. Without deduplication, every retry triggers your business logic again — a call gets answered twice, a customer gets three identical SMS replies.

The sample uses an in-memory dict with a TTL:

dedup_store = {}  # {event_id: (timestamp, timestamp)}
DEDUP_TTL_SECONDS = int(os.getenv("DEDUP_TTL_SECONDS", "300"))

def is_duplicate(event_id):
    current_time = time.time()
    # Clean up expired entries
    expired = [k for k, (ts, _) in dedup_store.items()
               if current_time - ts > DEDUP_TTL_SECONDS]
    for key in expired:
        del dedup_store[key]
    if event_id in dedup_store:
        return True
    dedup_store[event_id] = (current_time, current_time)
    return False

Every check sweeps expired entries first, so the store never grows unbounded. The default TTL is 300 seconds, which covers Telnyx's retry window. The event ID comes from the Telnyx event payload when present, and falls back to a SHA-256 of the sorted payload when it does not.

The dedup check happens before the SQLite log and before the fanout. A duplicate event returns {"status": "duplicate"} with a 200 — because returning a non-2xx would make Telnyx retry again.

Log to SQLite before you act

Every non-duplicate event is written to SQLite before any action fires:

def log_event(event_id, event_type, payload):
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute(
        "INSERT OR IGNORE INTO webhook_events "
        "(event_id, event_type, payload, received_at, processed_at) "
        "VALUES (?, ?, ?, ?, ?)",
        (event_id, event_type, json.dumps(payload),
         datetime.now(timezone.utc).isoformat(),
         datetime.now(timezone.utc).isoformat())
    )
    conn.commit()
    conn.close()

INSERT OR IGNORE on the UNIQUE event_id column makes the log idempotent — even if two identical events slip past the in-memory dedup in a race, SQLite enforces uniqueness at the storage layer.

The log exists so that when something goes wrong at 3am, you have a real table: event_id, event_type, payload, received_at, processed_at. You can query it, export it, or build a dashboard on top of it. It is operational data, not a log file you grep.

Fan out by event type

After verification, dedup, and logging, the event is routed to one of two in-memory queues based on its type:

ACTION_TYPES = ["call", "sms"]
action_queues = defaultdict(list)

# In webhook_handler:
if "call" in event_type.lower():
    enqueue_action("call", event)
elif "message" in event_type.lower() or "sms" in event_type.lower():
    enqueue_action("sms", event)

The fanout is a router, not a dispatcher. It decides which queue gets the event; it does not execute the action inline. The queue is drained in the same request for this sample, but in production you would move the drain to a separate worker process.

The queue separation matters because call actions and SMS actions have different latency profiles and different failure modes. A call that takes 800ms to answer should not block an SMS reply that could fire in 50ms. Separating the queues lets you prioritize, retry, and monitor each action type independently.

Process: answer calls, reply to SMS

The queue drain uses the Telnyx Python SDK v4 client to take the actual action:

def process_call_action(event_data):
    payload = event_data.get("data", {}).get("payload", {})
    call_control_id = payload.get("call_control_id")
    telnyx_client.calls.answer(call_control_id=call_control_id)
    telnyx_client.calls.playback_start(
        call_control_id=call_control_id,
        audio_url="https://example.com/greeting.mp3"
    )

def process_sms_action(event_data):
    payload = event_data.get("data", {}).get("payload", {})
    from_number = payload.get("from")
    to_number = payload.get("to")
    text = payload.get("text", "")
    telnyx_client.messages.create(
        from_=to_number,
        to=from_number,
        text=f"Thanks for your message! We received: {text[:50]}..."
    )

Note the field access: event_data.get("data", {}).get("payload", {}). The Telnyx webhook envelope wraps the payload inside data.payload, and the SDK v4 typed event follows that structure. If you are migrating from SDK v2, this is the most common breakage point — v2 gave you the flat payload, v4 gives you the wrapped envelope.

Run the demo without Telnyx credentials

The sample ships with a single-file demo launcher (demo/demo_server.py) that runs the entire pipeline without Telnyx credentials or ngrok. It generates an Ed25519 keypair at startup, injects the public key into the Telnyx client, stubs the calls and messages API surfaces, and serves a dashboard at http://localhost:5555/.

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/webhook-aggregator-fanout
python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt
python demo/demo_server.py

Open the dashboard, click "Send Call Webhook" or "Send SMS Webhook," and watch the six-stage pipeline execute live. The dashboard shows the dedup KV, the SQLite event log, and the fanout queues in real time.

To test deduplication, send the same event twice — the first returns {"status": "success"}, the second returns {"status": "duplicate"}, and only one row appears in the events table.

Run it for real

cp .env.example .env
# Set TELNYX_API_KEY and TELNYX_PUBLIC_KEY in .env
flask --app app run --port 5000

Point your Telnyx webhook URL to https://your-public-url/webhooks (use ngrok for local testing) and configure your call and messaging profiles to deliver events there.

Where to take it next

For production, three things:

Move the queue drain to a worker. The sample drains queues in the same request so the pipeline is visible end to end. In production, enqueue to Redis or a real queue and drain from a separate process.
Add a dead-letter queue. If an action fails, it should not disappear. Log it, retry with backoff, and surface it for manual intervention.
Add monitoring on the queue depths. If the call queue is growing faster than it drains, you have a capacity problem before you have a customer problem.

The core pattern is simple: verify before you trust, deduplicate before you act, log before you process, and fan out before you execute. Everything else is infrastructure.

Build a Network Incident Agent with Telnyx Edge and the Agent SDK

Harpreet Singh Seehra — Thu, 27 Aug 2026 13:15:45 +0000

An incident is more than an alert. It has a lifecycle, affected customers, status changes, communication history, a root cause, and work that must happen after service is restored.

This sample models the incident itself as a durable Telnyx Edge actor.

Code: https://github.com/team-telnyx/telnyx-code-examples/tree/main/network-incident-agent

Architecture

NetworkIncidentAgent(incidentId)
├── durable state: status, severity, services, counters
├── KV: affected customers
├── SQL: ordered incident timeline
├── Messaging: proactive customer SMS
├── Voice: incident-aware Call Control response
├── CloudFS: RCA JSON
└── schedule(): delayed recurrence check

Using the incident ID as the actor name makes routing deterministic. The same ID always resolves to the same durable entity.

Validate lifecycle transitions

The actor limits state changes to valid transitions. This prevents an operator or automation from silently moving a detected incident directly to closed.

const ALLOWED_TRANSITIONS = {
  detected: ["investigating"],
  investigating: ["restoring", "resolved"],
  restoring: ["resolved", "investigating"],
  resolved: ["closed", "investigating"],
  closed: ["investigating"],
};

Every accepted transition updates state and inserts an event into the actor's SQL timeline.

Notify customers without exposing them

Affected phone numbers live in KV, but browser responses contain only masked values. In live mode, notifications use the Telnyx binding:

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

The dashboard starts in demo mode, where delivery is simulated. This makes the complete workflow safe to run during development or a screen recording. Real delivery requires an explicit live-mode selection and valid Telnyx configuration.

Use the same context for inbound calls

When a call.initiated webhook arrives, the application reads the incident ID from client_state, resolves the actor, and retrieves its current voice message. It then answers the call and speaks that context through Call Control.

SMS and voice therefore read from the same incident state rather than maintaining separate status copies.

Write the RCA and check for recurrence

After resolution, the actor serializes the incident, affected-service metadata, root cause, and SQL timeline into JSON. It writes the file to CloudFS through a temporary path and atomic rename.

It then schedules checkRecurrence:

await this.schedule(
  delay,
  "checkRecurrence",
  { incidentId },
  { id: `recurrence-${incidentId}` },
);

If the incident remains resolved, the task closes it. If its state has regressed, it records a recurrence.

Run the demo

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/network-incident-agent
cp .env.example .env
npm install
npm run build
mkdir -p /tmp/network-incident-cloudfs
CLOUDFS_MOUNT_PATH=/tmp/network-incident-cloudfs npm start

Open the local URL printed by telnyx-edge and run the incident lifecycle. Leave live SMS disabled unless you have configured an approved sender and opted-in test destinations.

Run the repeatable smoke test with:

DEMO_BASE_URL=http://localhost:8787 npm run smoke

Where to take it next

For production, add operator authentication, webhook signature validation at ingress, idempotency for commands and notifications, monitoring-system integrations, and authorization rules around customer-impact data.

The core pattern remains simple: give the long-lived entity a durable actor identity, then make AI and communication channels capabilities of that entity.

Build a Multi-Agent Workflow with Shared CloudFS and the Agent SDK

Harpreet Singh Seehra — Tue, 25 Aug 2026 15:04:54 +0000

Multi-agent applications need more than independent reasoning. They need a reliable way to hand durable artifacts from one specialist to the next.

This sample creates five Telnyx Agent SDK actors: writer, analyst, reviewer, summarizer, and publisher. They share one CloudFS workspace and produce a chain of Markdown and JSON files.

Architecture

CloudFS /shared/runs/<runId>/
├── report.md
├── analysis.json
├── review.md
├── summary.md
└── manifest.json

FleetRegistry
├── agents table
└── files table

CloudFS is mounted with JuiceFS as a POSIX filesystem. The actors use node:fs/promises, not a special storage API. This lets the local demo use /tmp/agentfs while a deployed environment uses the real mount path.

Why combine files and SQL?

CloudFS stores the artifacts themselves. Embedded SQL stores searchable operational metadata: actor role, current status, last artifact, file size, operation type, and timestamp. GET /fleet combines that information into a fleet-wide view.

Every actor also uses AgentSocketServer to broadcast its state. Connected dashboards receive reading, writing, done, and error transitions in real time.

Safe shared writes

The workspace helper validates every relative path and prevents it from escaping the shared directory. It writes content to a temporary file and atomically renames it into place, preventing readers from seeing a partial artifact.

Run the guided demo

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/agent-fleet-shared-workspace
npm install
cp .env.example .env
npm start

Set CLOUDFS_MOUNT_PATH=/tmp/agentfs for local development and open http://localhost:8787. Select Run agent fleet to watch all five stages, generated artifacts, SQL history, and previews update through the handoff.

For an isolated paced run:

curl -X POST http://localhost:8787/demo \
  -H 'Content-Type: application/json' \
  -d '{"runId":"recording-take-01","paceMs":1100}'

Use cases

This pattern works well for content production, document processing, research pipelines, and software delivery. Each agent owns one responsibility, while the shared workspace makes its output durable and inspectable.

Source: https://github.com/team-telnyx/telnyx-code-examples/tree/main/agent-fleet-shared-workspace

Video: https://www.youtube.com/watch?v=ZvEgg_CA_aM

Build a KV-Backed Rate Limiter with Edge Compute and the Agent SDK

Harpreet Singh Seehra — Mon, 24 Aug 2026 12:55:17 +0000

Rate limiting is a foundational API pattern — every production endpoint needs it. But the traditional approach means standing up Redis, wiring it to your API gateway, and building a separate alerting pipeline. That's three services before you've written a single line of business logic.

This sample shows a different approach: a sliding-window rate limiter that runs entirely on Telnyx Edge Compute using the Agent SDK. KV counters track request counts per key, TTL-based windows auto-expire without cleanup code, over-limit requests get rejected with HTTP 429, and SMS alerts fire automatically when rejections cross a threshold — all from a single actor with zero external dependencies.

What We're Building

A rate limiter that:

Tracks request counts per key (phone number, IP address, tenant ID) in Edge KV
Uses sliding time windows with TTL-based auto-expiry — no cleanup jobs
Returns HTTP 429 with metadata when a key exceeds its limit
Sends SMS alerts via a zero-credential [telnyx] binding when rejections sustain
Isolates each key in its own Agent SDK actor — no contention between keys
Provides a /simulate endpoint for testing without real traffic

Architecture

                    Client Request
                           │
                           ▼
              POST /check
                           │
                           ▼
              ┌────────────────────┐
              │   index.ts         │
              │   (HTTP router)     │
              └────┬───────────────┘
                   │
                   ▼
         ┌──────────────────────┐
         │  RateLimitAgent      │  (one actor per key)
         │                      │
         │  1. checkLimit()    │──► KV: GET current window count
         │     if under limit   │
         │  2a. allow()         │──► KV: PUT incremented count (TTL) → 200
         │     if over limit    │
         │  2b. reject()        │──► 429 + track rejection count
         │     if rejections    │
         │     >= threshold     │
         │  3. sendAlert()      │──► SMS via [telnyx] binding
         └──────────────────────┘

The Sliding Window Pattern

The key insight is how KV keys are structured. Each key encodes the rate-limited identifier and the window start time:

rate:+18005551234:1718928000   ← count for the 14:00–14:01 window
rate:+18005551234:1718928060   ← count for the 14:01–14:02 window
rate:+18005559876:1718928000   ← different key, same window

Each KV entry has a TTL of WINDOW_SECONDS (default: 60). When the window passes, the entry auto-expires. No cron jobs, no cleanup code, no Redis eviction policies to tune. This is a fixed-window approximation of a sliding window — simpler than a true sliding window log, but it uses exactly one KV get and one KV put per request.

The Agent SDK Pipeline

Each rate-limited key gets its own RateLimitAgent actor instance. When /check is called, the agent queues a non-blocking pipeline:

Stage 1: checkLimit()

async checkLimit(): Promise<void> {
  const state = await this.getState();
  const windowStart = Math.floor(Date.now() / 1000 / state.windowSeconds) * state.windowSeconds;
  const kvKey = `rate:${state.key}:${windowStart}`;
  const currentStr = await this.env.RATE_KV.get(kvKey);
  const currentCount = currentStr ? parseInt(currentStr, 10) : 0;

  await this.setState({
    ...state,
    currentCount,
    totalRequests: state.totalRequests + 1,
    lastRequestAt: Date.now(),
  });

  if (currentCount < state.limit) {
    await this.queue("allow");
  } else {
    await this.queue("reject");
  }
}

The agent reads the current window count from KV, updates its state, and then queues the next stage based on whether the key is under or over the limit.

Stage 2a: allow()

async allow(): Promise<void> {
  const state = await this.getState();
  const windowStart = Math.floor(Date.now() / 1000 / state.windowSeconds) * state.windowSeconds;
  const kvKey = `rate:${state.key}:${windowStart}`;
  const newCount = state.currentCount + 1;
  await this.env.RATE_KV.put(kvKey, String(newCount), {
    expirationTtl: state.windowSeconds,
  });

  await this.setState({
    ...state,
    currentCount: newCount,
    allowedRequests: state.allowedRequests + 1,
    status: "allowed",
  });
  await this.queue("finalize");
}

The counter is incremented and written back to KV with a TTL. The request is marked as allowed.

Stage 2b: reject()

async reject(): Promise<void> {
  const state = await this.getState();
  const newRejectionCount = state.rejectionCount + 1;

  await this.setState({
    ...state,
    rejectedRequests: state.rejectedRequests + 1,
    rejectionCount: newRejectionCount,
    status: newRejectionCount >= state.alertThreshold ? "alerting" : "rejected",
  });

  if (newRejectionCount >= state.alertThreshold && !state.alertTriggered) {
    await this.queue("sendAlert");
  } else {
    await this.queue("finalize");
  }
}

When a request is rejected, the agent increments its rejection counter. If the rejection count crosses ALERT_THRESHOLD and no alert has been sent yet, the alert stage is queued.

Stage 3: sendAlert()

async sendAlert(): Promise<void> {
  const state = await this.getState();
  const smsText =
    `Rate limit alert: key "${state.key}" has been rejected ${state.rejectionCount} times. ` +
    `Limit is ${state.limit} requests per ${state.windowSeconds}s window. ` +
    `Total requests: ${state.totalRequests}. Allowed: ${state.allowedRequests}. Rejected: ${state.rejectedRequests}.`;

  await this.env.TELNYX.messages.send({
    from: this.env.SENDER_PHONE,
    to: this.env.ALERT_PHONE,
    text: smsText,
  });

  await this.setState({ ...state, alertTriggered: true, status: "done" });
}

The SMS is sent via the [telnyx] binding — no API key in code, no HTTP client, no auth header. The binding handles credentials at the platform level.

Quick Start

1. Clone and install

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/kv-backed-rate-limiter
npm install

2. Configure environment

Copy .env.example and fill in your values:

cp .env.example .env

Variable	Description	Example
`TELNYX_API_KEY`	Telnyx API key	`KEY019...`
`SENDER_PHONE`	Telnyx number sending alerts	`+18005551234`
`ALERT_PHONE`	Ops phone receiving alerts	`+18005559876`
`RATE_LIMIT`	Max requests per window per key	`10`
`WINDOW_SECONDS`	Sliding window duration	`60` (1 minute)
`ALERT_THRESHOLD`	Rejections before SMS fires	`5`

3. Deploy

telnyx-edge secret set TELNYX_API_KEY KEY0123456789ABCDEF
telnyx-edge ship

4. Test the rate limit

# Check a single request
curl -X POST https://your-deployment.telnyxcompute.com/check \
  -H "Content-Type: application/json" \
  -d '{"key":"+18005551234"}'

# Simulate a burst of 15 requests (limit is 10 → 10 allowed, 5 rejected)
curl -X POST https://your-deployment.telnyxcompute.com/simulate \
  -H "Content-Type: application/json" \
  -d '{"key":"+18005551234","count":15}'

The simulate response shows each request, whether it was allowed or rejected, and the current count:

{
  "key": "+18005551234",
  "simulated": 15,
  "allowed": 10,
  "rejected": 5,
  "alertTriggered": true,
  "results": [
    { "request": 1, "action": "allowed", "currentCount": 1 },
    { "request": 2, "action": "allowed", "currentCount": 2 },
    ...
    { "request": 11, "action": "rejected", "currentCount": 10 },
    { "request": 12, "action": "rejected", "currentCount": 10 },
    ...
  ]
}

API Endpoints

Method	Path	Description
`POST`	`/check`	Check a request against the rate limit
`GET`	`/status/:key`	Get agent state for a specific key
`GET`	`/count/:key`	Get current window count for a key
`GET`	`/keys`	List all tracked keys with aggregate stats
`POST`	`/reset/:key`	Reset a key's counter
`POST`	`/simulate`	Simulate a burst of requests for testing
`GET`	`/config`	Get current rate limit configuration
`GET`	`/health/liveness`	Liveness probe
`GET`	`/health/readiness`	Readiness probe

Why This Matters

No Redis, no external services

The traditional rate limiter needs Redis for the counter store and a separate alerting service (SNS, PagerDuty, etc.) for threshold notifications. This sample replaces both with Edge KV (built-in binding) and the [telnyx] messaging binding (zero-credential SMS). One deploy, zero external dependencies.

TTL-based cleanup

The sliding window is implemented as a fixed-window approximation: each window gets its own KV key with a TTL. When the window expires, the key auto-expires. No cleanup cron jobs, no eviction policy tuning, no memory management. The window boundary is computed from Math.floor(Date.now() / 1000 / windowSeconds) * windowSeconds.

Per-key actor isolation

Each rate-limited key gets its own RateLimitAgent actor instance. This means:

No contention between concurrent keys
Per-key state (allowed/rejected counts, alert status) survives across requests
The RateLimitRegistry singleton actor aggregates across keys for the /keys endpoint

Non-blocking pipeline

The this.queue() pattern from the Agent SDK means each stage runs asynchronously. The HTTP request that triggers checkRequest() returns immediately — the pipeline stages execute in the background. This is critical for rate limiting: you don't want the client to wait for the alert SMS to send before getting their 429.

Use Cases

API rate limiting — Limit requests per API key, IP, or tenant
Call Control protection — Limit inbound call rate per phone number before they hit your IVR
SMS flood prevention — Rate limit outbound SMS per user before sending
Webhook throttling — Limit incoming webhook rate per source to protect downstream services
Tenant quotas — Per-tenant request caps on multi-tenant platforms

Frequently Asked Questions

How is this different from a Redis-based rate limiter?

Redis requires a separate service to provision, manage, and pay for. Edge KV is a built-in binding — no provisioning, no connection pooling, no network hops. The TTL-based cleanup is native to KV. The trade-off: Edge KV is eventually consistent, so there's a small window where concurrent requests might both read the same count. For most rate limiting use cases, this is acceptable.

Can I use a true sliding window instead of a fixed window?

Yes. Replace the single counter per window with a sorted set of timestamps (one per request) and count entries within the window. This is more accurate but uses more KV storage and requires listing keys within a range. The fixed-window approximation is simpler and uses exactly one KV get and one KV put per request.

What happens if the SMS alert fails?

The agent catches the error and sets its status to "error" with the error message. The alert is not retried automatically, but the alertTriggered flag is only set on success, so the next rejection will queue another alert attempt. You can monitor for the "error" status via the /status/:key endpoint.

How do I reset a key's counter?

Use the POST /reset/:key endpoint. This deletes the current window's KV entry for that key, effectively resetting the counter to zero.

Resources

Code sample: kv-backed-rate-limiter
Agent SDK docs: developers.telnyx.com/docs/agent-sdk
Edge KV docs: developers.telnyx.com/docs/edge-compute/kv
Messaging docs: developers.telnyx.com/docs/messaging
Telnyx Portal: portal.telnyx.com

Build a Chunked SMS Conversation Exporter with Edge SQL, Cloud Storage, and the Agent SDK

Harpreet Singh Seehra — Fri, 21 Aug 2026 13:01:24 +0000

Export SMS conversation history from Edge SQL to Cloud Storage as chunked JSON files, with a 4-stage Agent SDK pipeline that handles 10k+ messages — and texts you when the export is done. All on Telnyx Edge Compute, in under 350 lines of TypeScript.

What You'll Build

An SMS conversation exporter that:

Ingests SMS messages via Telnyx Messaging webhooks into a per-actor SQL database
Counts and chunks messages for export (configurable chunk size, default 500)
Uploads each chunk as a separate JSON file to Cloud Storage
Writes a manifest file listing all chunks with metadata
Sends an SMS notification when the export is complete (zero-credential binding)
Handles 10,000+ messages via non-blocking, self-requeuing pipeline stages

The entire system runs on Telnyx Edge Compute — no external database, no separate upload service, no cross-cloud latency.

Architecture

                    POST /export
                         │
                         ▼
         ┌──────────────────────┐
         │  ExportAgent         │  (one actor per export job)
         │                      │
         │  1. countMessages()  │──► SQL DB: SELECT COUNT(*)
         │  2. exportChunk()    │──► SQL DB: SELECT chunk
         │                      │──► Cloud Storage: PUT JSON chunk
         │     (re-queues       │    (repeats until all chunks done)
         │      until done)     │
         │  3. writeManifest()  │──► Cloud Storage: PUT manifest.json
         │  4. notifyComplete() │──► SMS via [telnyx] binding
         └──────────────────────┘

Quick Start

1. Clone and install

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/sms-conversation-exporter
npm install

2. Configure

Edit telnyx.toml:

[storage.cloudstorage.EXPORT_STORAGE]
bucket_name = "<your-storage-bucket-name>"
region = "us-central-1"

[env_vars]
ALERT_PHONE = "+18005559876"
SENDER_PHONE = "+18005551234"
CHUNK_SIZE = "500"

3. Deploy

telnyx-edge secret set TELNYX_API_KEY KEY0123456789ABCDEF
telnyx-edge ship

4. Test the export

# Simulate 10,000 messages
curl -X POST https://your-deployment.telnyxcompute.com/simulate-bulk \
  -H "Content-Type: application/json" \
  -d '{"count": 10000}'

# Start the export
curl -X POST https://your-deployment.telnyxcompute.com/export

# Check progress
curl https://your-deployment.telnyxcompute.com/export/{exportId}

How the Pipeline Works

Agent SDK `queue()` — Non-Blocking Stages

Each export job gets its own ExportAgent actor instance. The pipeline uses this.queue() to chain stages without blocking the HTTP request:

async start(params: { exportId: string; conversationFilter: string | null }): Promise<void> {
  await this.setState({ exportId: params.exportId, status: "counting" });
  await this.queue("countMessages");
}

Self-Requeuing Chunks

The key pattern is the self-requeuing exportChunk() stage:

async exportChunk(): Promise<void> {
  const state = await this.getState();
  const offset = state.chunkIndex * chunkSize;

  const rows = this.ctx.storage.sql.exec(
    "SELECT * FROM messages ORDER BY timestamp ASC LIMIT ? OFFSET ?",
    chunkSize, offset
  ).toArray();

  await this.env.EXPORT_STORAGE.put(
    `exports/${state.exportId}/chunk-${state.chunkIndex}.json`,
    JSON.stringify({ messages: rows })
  );

  if (state.chunkIndex + 1 < state.totalChunks) {
    await this.queue("exportChunk");  // schedule next chunk
  } else {
    await this.queue("writeManifest");
  }
}

Each chunk runs as a separate, non-blocking stage. The actor isn't blocked, progress is visible at any time via GET /export/:id, and a chunk failure is captured in state.

Zero-Credential SMS Notification

async notifyComplete(): Promise<void> {
  await this.env.TELNYX.messages.send({
    from: this.env.SENDER_PHONE,
    to: this.env.ALERT_PHONE,
    text: `Export complete: ${state.exportedMessages} messages in ${state.uploadedChunks.length} chunk(s).`
  });
}

No API key in code. The [telnyx] binding in telnyx.toml gives the actor access to the Messaging API — the platform handles authentication.

Chunked Output in Cloud Storage

exports/export-1234567890-abc123/
├── chunk-0000.json    (messages 0–499)
├── chunk-0001.json    (messages 500–999)
├── chunk-0002.json    (messages 1000–1499)
├── ...
└── manifest.json      (metadata: total count, chunk list)

Each chunk is self-contained with metadata — exportId, chunkIndex, totalChunks, totalMessages. Process chunks independently in Spark, BigQuery, or any data pipeline.

API Endpoints

Method	Path	Description
POST	`/webhooks/messaging`	Messaging webhook (ingests SMS into SQL)
POST	`/export`	Start a chunked export job
GET	`/export/:id`	Get export status and progress
GET	`/messages`	List messages in SQL DB
GET	`/messages/count`	Get total message count
POST	`/seed`	Add a single test message
POST	`/simulate-bulk`	Bulk insert test messages (default 10k)

Use Cases

Compliance archival — Export SMS conversation history for regulatory compliance
Data migration — Move SMS data from Edge SQL to a data warehouse via Cloud Storage
Backup — Periodic JSON exports of all conversations to Cloud Storage
Analytics — Export conversation data for offline analysis

Resources

Build a Geo-Distributed Call Logger with Per-Region KV Counters and SMS Alerting on Edge Compute

Harpreet Singh Seehra — Thu, 20 Aug 2026 14:02:35 +0000

Build a Geo-Distributed Call Logger on Edge Compute

Log Telnyx Call Control events to a shared SQL database, track per-region call volume in Edge KV counters, and trigger SMS alerts when a region exceeds a threshold — all on Telnyx Edge Compute with the Agent SDK, in under 400 lines of TypeScript.

What You'll Build

A geo-distributed call logger that:

Receives Telnyx Call Control webhooks (call.initiated, call.answered, call.hangup)
Logs every call to a per-call SQL database
Increments per-region counters in Edge KV with rolling-window TTL
Detects the caller's region from the E.164 country code prefix
Sends an SMS alert to your ops phone when a region's call count exceeds a threshold
Exposes HTTP endpoints for querying call status, recent calls, and region statistics

The entire system runs on Telnyx Edge Compute — no external database, no separate alerting service, no cross-cloud latency.

Why This Architecture Matters

If you've ever tried to build call analytics across multiple regions, you've probably wired together:

A webhook receiver (Lambda function or small server)
A time-series database (InfluxDB, TimescaleDB, or DynamoDB)
A separate counter service (Redis)
An alerting pipeline (PagerDuty + SNS + a Lambda)
A dashboard (Grafana or a custom React app)

That's five services, five deploys, five billing lines, and cross-region latency at every hop. On Edge Compute with the Agent SDK, this collapses to one deploy.

Architecture

                    Telnyx Call Control
                           │
                           ▼
              POST /webhooks/voice
                           │
                           ▼
              ┌────────────────────┐
              │   index.ts         │
              │   (webhook router)  │
              └────┬───────────────┘
                   │
                   ▼
         ┌──────────────────────┐
         │  GeoLoggerAgent      │  (one actor per call)
         │                      │
         │  1. logCall()         │──► SQL DB: INSERT call record
         │                      │──► KV:     INCR region counter
         │  2. checkThreshold()  │──► compare count vs threshold
         │  3. alert()           │──► SMS via [telnyx] binding
         └──────────────────────┘
                   │
                   ▼
         ┌──────────────────────┐
         │  CallRegistry        │  (singleton actor)
         │  cross-call listing  │──► GET /calls
         └──────────────────────┘

Each call gets its own GeoLoggerAgent actor instance with isolated state and its own SQL database. When call.hangup fires, the agent queues a 3-stage pipeline:

logCall() — Inserts the call record into the actor's SQL DB and increments the region counter in KV
checkThreshold() — Compares the post-increment region count against REGION_THRESHOLD
alert() — Sends an SMS via the zero-credential [telnyx] binding — no API key needed in code

Prerequisites

Node.js 18+
Telnyx CLI (npm i -g @telnyx/cli)
A Telnyx account with a Call Control number, an API key, and a messaging-enabled number

Step 1: Configure the Edge Compute Bindings

The telnyx.toml file declares the bindings your agent needs:

name = "geo-distributed-call-logger"
main = "src/index.ts"
compatibility_date = "2026-05-01"

[[actors]]
binding = "GEO_LOGGER"
type = "GeoLoggerAgent"

[[actors]]
binding = "REGISTRY"
type = "CallRegistry"

[telnyx]
binding = "TELNYX"

[storage.kv.REGION_KV]
id = "<kv-namespace-uuid>"

[env_vars]
ALERT_PHONE = "+18005551234"
SENDER_PHONE = "+18005551234"
REGION_THRESHOLD = "100"
WINDOW_SECONDS = "3600"

[[secrets]]
binding = "TELNYX_API_KEY"
name = "TELNYX_API_KEY"

Four bindings: two actor namespaces (GEO_LOGGER, REGISTRY), the zero-credential messaging binding (TELNYX), and the KV namespace (REGION_KV).

Step 2: Region Detection from E.164 Prefixes

const COUNTRY_TO_REGION: Record<string, string> = {
  "1": "us-east-1",      // US/Canada (+1)
  "44": "eu-west-1",      // UK (+44)
  "33": "eu-west-1",      // France (+33)
  "49": "eu-central-1",  // Germany (+49)
  "31": "eu-west-1",     // Netherlands (+31)
  "81": "ap-northeast-1", // Japan (+81)
  "82": "ap-northeast-1", // South Korea (+82)
  "86": "ap-east-1",     // China (+86)
  "91": "ap-south-1",    // India (+91)
  "61": "ap-southeast-1", // Australia (+61)
  "55": "sa-east-1",     // Brazil (+55)
};

export function detectRegion(phoneNumber: string): string {
  const digits = phoneNumber.replace(/^\+/, "");
  const cc2 = digits.slice(0, 2);
  const cc1 = digits.slice(0, 1);
  if (COUNTRY_TO_REGION[cc2]) return COUNTRY_TO_REGION[cc2];
  if (COUNTRY_TO_REGION[cc1]) return COUNTRY_TO_REGION[cc1];
  return "unknown";
}

Country-code prefix to named region. 10 lines of code. For production, swap in a carrier lookup or Number Insight API for city-level precision.

Step 3: The GeoLoggerAgent Pipeline

The GeoLoggerAgent extends the Agent SDK's Agent class. Each call gets its own actor instance:

export class GeoLoggerAgent extends Agent<GeoLoggerEnv, GeoLoggerState> {
  protected override initialState(): GeoLoggerState {
    return {
      callControlId: "",
      fromNumber: "",
      toNumber: "",
      direction: "inbound",
      region: "unknown",
      status: "ringing",
      startedAt: 0,
      answeredAt: 0,
      endedAt: 0,
      durationSec: 0,
      logged: false,
      alertTriggered: false,
      regionCount: 0,
      threshold: 0,
      error: "",
    };
  }
}

When call.hangup fires, the agent queues the 3-stage pipeline:

async onHangup(): Promise<void> {
  const state = await this.getState();
  const endedAt = Date.now();
  const durationSec = state.answeredAt
    ? Math.round((endedAt - state.answeredAt) / 1000)
    : 0;
  await this.setState({ ...state, status: "hungup", endedAt, durationSec });
  await this.queue("logCall");
}

Stage 1: logCall — SQL INSERT + KV INCR

async logCall(): Promise<void> {
  const state = await this.getState();
  try {
    // SQL: insert call record
    this.ctx.storage.sql.exec(
      `CREATE TABLE IF NOT EXISTS calls (
        call_control_id TEXT PRIMARY KEY,
        from_number     TEXT NOT NULL,
        to_number       TEXT NOT NULL,
        direction       TEXT NOT NULL,
        region          TEXT NOT NULL,
        duration_sec    INTEGER NOT NULL,
        started_at      INTEGER NOT NULL,
        ended_at        INTEGER NOT NULL,
        status          TEXT NOT NULL
      )`
    );
    this.ctx.storage.sql.exec(
      `INSERT OR REPLACE INTO calls
        (call_control_id, from_number, to_number, direction, region,
         duration_sec, started_at, ended_at, status)
       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
      state.callControlId, state.fromNumber, state.toNumber,
      state.direction, state.region, state.durationSec,
      state.startedAt, state.endedAt, "completed"
    );

    // KV: increment region counter (rolling window)
    const windowSec = parseInt(this.env.WINDOW_SECONDS, 10) || 3600;
    const windowStart = Math.floor(Date.now() / 1000 / windowSec) * windowSec;
    const kvKey = `region:${state.region}:${windowStart}`;
    const currentStr = await this.env.REGION_KV.get(kvKey);
    const current = currentStr ? parseInt(currentStr, 10) : 0;
    const newCount = current + 1;
    await this.env.REGION_KV.put(kvKey, String(newCount), {
      expirationTtl: windowSec,
    });

    await this.setState({ ...state, logged: true, regionCount: newCount, status: "logged" });
    await this.queue("checkThreshold");
  } catch (e: unknown) {
    const msg = e instanceof Error ? e.message : String(e);
    await this.setState({ ...state, status: "error", error: `logCall: ${msg}` });
  }
}

Two things in one stage: SQL INSERT via ctx.storage.sql.exec() and KV INCR with rolling-window TTL.

Stage 2: checkThreshold

async checkThreshold(): Promise<void> {
  const state = await this.getState();
  if (state.regionCount >= state.threshold) {
    await this.setState({ ...state, status: "alerting", alertTriggered: true });
    await this.queue("alert");
  } else {
    await this.setState({ ...state, status: "done" });
  }
}

Stage 3: alert — Zero-Credential SMS

async alert(): Promise<void> {
  const state = await this.getState();
  const smsText =
    `Geo Call Alert: region "${state.region}" hit ${state.regionCount} calls ` +
    `in the current window (threshold: ${state.threshold}). ` +
    `Last call: ${state.fromNumber} → ${state.toNumber}, ${state.durationSec}s. ` +
    `Check the dashboard.`;

  await this.env.TELNYX.messages.send({
    from: this.env.SENDER_PHONE,
    to: this.env.ALERT_PHONE,
    text: smsText,
  });

  await this.setState({ ...state, status: "done" });
}

The SMS is sent via this.env.TELNYX.messages.send() — the zero-credential [telnyx] binding. No API key in code, no auth header. The binding carries auth from the TELNYX_API_KEY secret.

Step 4: The Webhook Handler

async function handleCallWebhook(req: Request, env: Env): Promise<Response> {
  const event = (await req.json()) as CallControlEvent;
  const data = event.data;
  const callControlId = data.call_control_id;
  const agent = env.GEO_LOGGER.idFromName(actorName(callControlId));

  switch (data.event_type) {
    case "call.initiated":
      await agent.onCallStart({
        callControlId,
        fromNumber: data.from,
        toNumber: data.to,
        direction: data.direction,
      });
      break;
    case "call.answered":
      await agent.onAnswered();
      break;
    case "call.hangup":
      await agent.onHangup(); // queues logCall → checkThreshold → alert
      break;
  }

  return Response.json({ received: true, eventType: data.event_type });
}

Step 5: Run and Test

Install and configure

cd geo-distributed-call-logger
npm install
cp .env.example .env
# Fill in TELNYX_API_KEY, SENDER_PHONE, ALERT_PHONE

Run locally

npm start

Simulate a call (no real phone needed)

curl -X POST http://localhost:3000/simulate \
  -H "Content-Type: application/json" \
  -d '{"from":"+31612345678","to":"+18005551234","duration":42}'

Check region statistics

curl http://localhost:3000/regions/stats

Response:

{
  "threshold": 100,
  "windowSeconds": 3600,
  "regions": [
    { "region": "us-east-1", "count": 47, "windowStart": 1718928000 },
    { "region": "eu-west-1", "count": 103, "windowStart": 1718928000 }
  ]
}

Trigger an alert

Set REGION_THRESHOLD=2, simulate 3 calls from the same region, and the third triggers an SMS to your phone.

Rolling-Window KV Counters

The KV key scheme is the key insight:

region:eu-west-1:1718928000   ← count for 14:00–15:00 window
region:eu-west-1:1718931600   ← count for 15:00–16:00 window

Each key has expirationTtl: WINDOW_SECONDS. When the window expires, the key auto-deletes. No cron job, no cleanup Lambda, no stale counters. The windowStart is Math.floor(Date.now() / 1000 / windowSec) * windowSec — floor to the window boundary, so all calls in the same hour share the same key.

Why Actors?

Each call is isolated in its own GeoLoggerAgent actor instance:

No contention — 100 simultaneous calls get 100 actor instances
State survives retries — if a webhook is retried, the actor state persists
Per-call SQL DB — each actor has its own SQL instance, no database locking
The CallRegistry singleton aggregates across calls for the /calls endpoint

This is the Agent SDK's core value: stateful, durable execution without a database server.

API Endpoints

Method	Path	Description
`POST`	`/webhooks/voice`	Call Control webhook receiver
`GET`	`/status/:callId`	Get agent state for a specific call
`GET`	`/calls`	List recent calls (from registry actor)
`GET`	`/regions/stats`	Per-region call counts in the current window
`GET`	`/regions`	List supported regions and country codes
`POST`	`/simulate`	Simulate a call webhook (for testing)

Frequently Asked Questions

How does region detection work?

The sample maps E.164 country code prefixes to named regions. For production-grade geo-routing, replace detectRegion() with a carrier lookup or Number Insight API.

What happens when the rolling window expires?

The KV key auto-deletes via TTL. No cleanup code. The next call in the new window starts a fresh counter at 1.

Can I change the threshold at runtime?

Yes — REGION_THRESHOLD is an environment variable. Update it via telnyx-edge secret set and the next call picks up the new value.

Do I need a database server?

No. The SQL DB is built into the Edge Compute actor runtime — this.ctx.storage.sql.exec(). No connection string, no pool, no server to manage.

Resources

If you found this helpful, follow me for more Edge Compute and Voice API content. The full code is on GitHub — clone it, run it, break it. The /simulate endpoint means you don't even need a real call to see it work.

Build an Edge Cache Invalidation Agent with the Telnyx Agent SDK

Harpreet Singh Seehra — Thu, 20 Aug 2026 09:04:45 +0000

Trigger cache invalidation across edge locations via webhook, mark caches dirty with KV, update a shared manifest in Cloud Storage, and notify ops via SMS. All orchestrated by the Agent SDK on Telnyx Edge Compute with zero-credential messaging.

The Multi-Region Cache Problem

When you run a content-heavy application across multiple edge locations, cache invalidation becomes a critical operational concern. A CMS publishes a new page version. A deployment pushes updated static assets. An emergency requires immediate content removal. In each case, every edge location needs to know that its cached copy is stale.

The standard approach involves calling a CDN purge API, then running a separate notification service to alert the ops team that the purge happened. Two services, two API keys, two integration points, and a gap between the purge and the notification where nobody knows whether the invalidation actually worked.

The Edge Cache Invalidation Agent does all of it on one network. Telnyx Edge Compute hosts the webhook and orchestrates the pipeline. Telnyx KV stores per-location dirty flags. Telnyx Cloud Storage holds the shared invalidation manifest. Telnyx Messaging delivers the SMS notification. One API key for KV and Cloud Storage. Zero credentials in code for messaging. One deploy command.

What It Does

You send a webhook to an Edge Compute endpoint. The pipeline runs three stages in sequence:

Invalidate — Mark the cache as dirty for each edge location via KV writes.
Update Manifest — Append the invalidation event to a shared manifest in Cloud Storage.
Notify — Send an SMS to the ops team with a summary of what changed and where.

Each stage is a queued job on a stateful actor. State survives across stages, so the list of invalidated locations from stage 1 is available to stage 2 for the manifest entry, and the summary from stage 2 is available to stage 3 for the SMS text. You can check the status of any pipeline run via a REST endpoint and get back the full state, including which locations were invalidated, whether the manifest was updated, and whether the SMS was sent.

Stage	Action	API
Trigger	Receive webhook, validate input, create actor	`POST /invalidate`
Start	Store metadata, queue invalidate	`CacheAgent.start()`
1. Invalidate	KV.put per location with dirty flag and TTL	`CACHE_KV.put("cache:{location}:{contentId}", value, { expirationTtl: 3600 })`
2. Update Manifest	Cloud Storage read/append/write shared manifest	`CACHE_STORAGE.put("cache-manifest.json", updated)`
3. Notify	SMS summary to ops team	`this.env.TELNYX.messages.send()`
Status	Check pipeline progress	`GET /status/:agentId`
Cache Status	Check if a location's cache is dirty	`GET /cache-status/:location/:contentId`

The Architecture

The application is a TypeScript Agent SDK project deployed to Telnyx Edge Compute. It uses three Telnyx products: Edge Compute (hosting and actor runtime), KV (per-location cache flags), Cloud Storage (shared manifest), and Messaging (SMS notifications).

  Content update webhook (POST /invalidate)
        │
        ▼
  ┌──────────────────────────────────────────┐
  │ CacheAgent.start()                        │
  │  → this.queue("invalidate")               │
  │  → this.queue("updateManifest")           │
  │  → this.queue("notify")                   │
  └────────┬─────────────────────────────────┘
           │
           ▼
  Stage 1: invalidate()
    → KV.put("cache:{location}:{contentId}", { dirty: true, version })
    → for each edge location
           │
           ▼
  Stage 2: updateManifest()
    → CloudStorage.get("cache-manifest.json")
    → append invalidation entry
    → CloudStorage.put("cache-manifest.json", updated)
           │
           ▼
  Stage 3: notify()
    → this.env.TELNYX.messages.send({ from, to, text })
    → SMS: "Cache invalidated: {contentId} v{version} — N locations updated"

The webhook handler receives a JSON payload with a content ID, content version, and array of edge locations. It validates the input, generates a Dapr-safe actor name, creates a CacheAgent actor instance, and calls start(). The agent stores the parameters in durable state and queues the first pipeline stage.

The Agent SDK runtime handles the rest. Each queued stage runs as a separate invocation of the actor, with state automatically loaded from the previous invocation. If a stage fails, the error is captured in state and the pipeline stops. You can inspect the status at any time.

Zero-Credential Messaging

The most important design decision in this pipeline is the [telnyx] binding in telnyx.toml:

[telnyx]
binding = "TELNYX"

This single line gives the Agent SDK runtime a pre-authenticated Telnyx client. At runtime, the actor accesses it through this.env.TELNYX. The SMS call in the notify() stage needs no API key, no Authorization header, no environment variable:

await this.env.TELNYX.messages.send({
  from: state.senderPhone,
  to: state.alertPhone,
  text: smsText,
});

Only TELNYX_API_KEY is needed as a secret, and only for KV and Cloud Storage, which use the API key directly. Messaging is zero-credential via the binding. Rotate keys in the portal and the binding picks up the new credentials automatically. No code changes. No redeploy.

Durable State Across Pipeline Stages

The Agent SDK provides durable state through this.setState() and this.getState(). State is scoped to the actor instance and survives across queued invocations:

// In CacheAgent.start()
await this.setState({
  contentId: params.contentId,
  contentVersion: params.contentVersion,
  locations: params.locations,
  senderPhone: this.env.SENDER_PHONE,
  alertPhone: this.env.ALERT_PHONE,
  status: "invalidating",
  createdAt: Date.now(),
});
this.queue("invalidate");

// In invalidate()
const state = await this.getState();
// ... mark caches dirty ...
await this.setState({
  invalidatedLocations: invalidated,
  status: "updating_manifest",
});
this.queue("updateManifest");

// In updateManifest()
const state = await this.getState();
// ... update manifest ...
await this.setState({
  manifestUpdated: true,
  status: "notifying",
});
this.queue("notify");

State is not held in memory between stages. It is persisted to the actor store and reloaded on the next invocation. This means the pipeline is resilient to runtime restarts, scaling events, or delays between stages. If the notify() stage fails (for example, the SMS gateway is temporarily down), the agent retries and reads the summary from state. It doesn't re-invalidate caches or re-update the manifest.

KV-Based Cache Invalidation

Each edge location gets a KV key with the pattern cache:{location}:{contentId}. The value is a JSON object with dirty: true, the new contentVersion, and a timestamp:

const kvKey = `cache:${location}:${state.contentId}`;
const value = JSON.stringify({
  dirty: true,
  contentVersion: state.contentVersion,
  invalidatedAt: Date.now(),
});
await this.env.CACHE_KV.put(kvKey, value, { expirationTtl: 3600 });

KV keys expire after 1 hour (TTL). If a location hasn't refreshed its cache by then, the flag disappears and the location serves stale content until the next invalidation. Adjust the TTL based on your cache refresh window.

You can check whether a specific location's cache is dirty at any time:

curl https://edge-cache-invalidation-agent-<id>.telnyxcompute.com/cache-status/us-east-1/blog/how-to-build-x

Response:

{
  "location": "us-east-1",
  "contentId": "/blog/how-to-build-x",
  "dirty": true,
  "contentVersion": "2026-08-19-v2"
}

Clear the dirty flag to simulate a cache refresh:

curl -X POST https://edge-cache-invalidation-agent-<id>.telnyxcompute.com/cache-clear/us-east-1/blog/how-to-build-x

Cloud Storage Manifest

The shared manifest (cache-manifest.json) in Cloud Storage is an append-only log of invalidation events. Each entry records what changed, the new version, which locations were invalidated, and when:

const manifestEntry = {
  contentId: state.contentId,
  contentVersion: state.contentVersion,
  invalidatedLocations: state.invalidatedLocations,
  updatedAt: Date.now(),
};

manifest.entries.push(manifestEntry);
await this.env.CACHE_STORAGE.put(MANIFEST_KEY, JSON.stringify(manifest, null, 2), {
  contentType: "application/json",
});

This gives ops a durable audit trail across all invalidations. Production should rotate or partition the manifest by date to avoid unbounded growth.

The Pipeline Queue Primitives

The Agent SDK uses this.queue() to schedule non-blocking work. Each call adds a stage to the actor's internal queue and returns immediately. The runtime invokes the queued method when capacity is available:

this.queue("invalidate");
this.queue("updateManifest");
this.queue("notify");

These three calls do not run in parallel. They run in the order queued, with state persisting between each. If invalidate() fails, updateManifest() never runs. If updateManifest() succeeds but notify() fails, the manifest update is still saved in state and you can retry the notify stage manually or inspect the error.

The queue is durable. If the Edge Compute worker restarts between stages, the queue is restored and processing resumes from the next pending stage.

Setup

Prerequisites

Telnyx Edge CLI v0.2.2+
Node.js 18+
A Telnyx account with:
- API key (Portal → API Keys)
- A phone number with SMS enabled (Portal → Numbers)
- KV namespace (Portal → Storage → KV)
- Cloud Storage bucket (Portal → Storage → Buckets)

1. Clone and install

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/edge-cache-invalidation-agent
npm install

2. Set secrets

telnyx-edge secret set TELNYX_API_KEY KEY0123456789ABCDEF
telnyx-edge secret set ALERT_PHONE +18005551234
telnyx-edge secret set SENDER_PHONE +18005551234

TELNYX_API_KEY is your Telnyx API v2 key from the Portal. ALERT_PHONE is the ops number that receives SMS notifications. SENDER_PHONE is a Telnyx number with SMS enabled.

3. Update `telnyx.toml`

Replace the placeholder values:

[storage.kv.CACHE_KV]
id = "your-kv-namespace-uuid"

[storage.cloudstorage.CACHE_STORAGE]
bucket_name = "your-bucket-name"
region = "us-central-1"

4. Deploy

telnyx-edge ship

The ship command builds the TypeScript project, uploads the bundle, and prints a URL like edge-cache-invalidation-agent-<id>.telnyxcompute.com.

5. Test

Health check:

curl https://edge-cache-invalidation-agent-<id>.telnyxcompute.com/health/liveness

Trigger an invalidation:

curl -X POST https://edge-cache-invalidation-agent-<id>.telnyxcompute.com/invalidate \
  -H "Content-Type: application/json" \
  -d '{
    "content_id": "/blog/how-to-build-x",
    "content_version": "2026-08-19-v2",
    "locations": ["us-east-1", "us-west-1", "eu-central-1", "ap-southeast-1"]
  }'

Check pipeline status:

curl https://edge-cache-invalidation-agent-<id>.telnyxcompute.com/status/<agentId>

API Reference

`POST /invalidate`

Trigger a cache invalidation across edge locations.

Request body:

{
  "content_id": "/blog/how-to-build-x",
  "content_version": "2026-08-19-v2",
  "locations": ["us-east-1", "us-west-1", "eu-central-1", "ap-southeast-1"]
}

Response (200):

{
  "action": "queued",
  "agentId": "bloghowtobuildx-20260819v2-1724080800000",
  "contentId": "/blog/how-to-build-x",
  "contentVersion": "2026-08-19-v2",
  "locations": ["us-east-1", "us-west-1", "eu-central-1", "ap-southeast-1"],
  "statusUrl": "/status/bloghowtobuildx-20260819v2-1724080800000"
}

`GET /status/:agentId`

Check the status of an invalidation pipeline.

Response:

{
  "contentId": "/blog/how-to-build-x",
  "contentVersion": "2026-08-19-v2",
  "status": "done",
  "invalidatedLocations": ["us-east-1", "us-west-1", "eu-central-1", "ap-southeast-1"],
  "manifestUpdated": true,
  "smsSent": true,
  "createdAt": 1724080800000,
  "completedAt": 1724080805000
}

`GET /cache-status/:location/:contentId`

Check if a specific location's cache is dirty for a content ID.

Response:

{
  "location": "us-east-1",
  "contentId": "/blog/how-to-build-x",
  "dirty": true,
  "contentVersion": "2026-08-19-v2"
}

`POST /cache-clear/:location/:contentId`

Clear the dirty flag for a location (simulates cache refresh).

Response:

{
  "action": "cleared",
  "location": "us-east-1",
  "contentId": "/blog/how-to-build-x"
}

`GET /locations`

List demo edge locations.

`GET /health/liveness` / `GET /health/readiness`

Health check endpoints.

Frequently Asked Questions

Q: Why does messaging need zero credentials while KV and Cloud Storage need an API key?

The [telnyx] binding in telnyx.toml is a platform-managed authentication channel. Edge Compute injects a pre-authenticated client into this.env.TELNYX at runtime. This client can call Telnyx Messaging because it is part of the same platform identity. KV and Cloud Storage, however, use direct API calls that require an explicit API key. The Telnyx API key serves as that credential for storage operations. One key covers both KV and Cloud Storage. Messaging is zero-credential via the binding.

Q: What happens if a pipeline stage fails?

The error is caught, stored in the actor's state under the error field, and the pipeline stops. The status endpoint returns the error message and the stage that failed. You can inspect the state, fix the issue (for example, set a missing secret), and retry the stage manually if your agent implementation supports it. Because state is durable, a retry reads progress from the previous successful stage instead of re-running it.

Q: Is the pipeline synchronous or asynchronous?

The invalidate endpoint returns immediately with a queued status. The pipeline runs asynchronously in the background. For a typical invalidation across four locations, the full pipeline (invalidate + updateManifest + notify) completes in under a second. You poll the status endpoint or build a webhook callback if you need real-time notification.

Q: Can I invalidate multiple content IDs at once?

The example triggers one invalidation per request. To support batch invalidation, modify the upload handler to accept an array of content IDs and loop over them in the invalidate stage. Each content ID gets its own agent instance, so batching is a matter of orchestration, not architecture.

Q: How do I know if an edge location actually refreshed its cache?

The KV dirty flag is a signal, not a guarantee. The edge location must poll or subscribe to the KV key to know when to refresh. The GET /cache-status/:location/:contentId endpoint lets you check whether the flag is still set. The POST /cache-clear/:location/:contentId endpoint simulates a refresh by clearing the flag. In production, your edge cache layer should read the KV key on each request and refresh when dirty is true.

Q: What is the carrier-edge advantage?

The webhook, KV writes, manifest update, and SMS delivery all happen inside the same Telnyx network boundary. No egress fees, no cross-cloud latency, no third-party CDN API. This is the Physics pillar of Telnyx's three-pillar thesis: carrier-edge compute beats cloud-edge on latency and cost.

Resources

Related Examples

Try It Yourself

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/edge-cache-invalidation-agent
npm install
telnyx-edge secret set TELNYX_API_KEY YOUR_API_KEY
telnyx-edge secret set ALERT_PHONE +18005551234
telnyx-edge secret set SENDER_PHONE +18005551234
telnyx-edge ship

Then trigger an invalidation and watch the pipeline run.

Telnyx is AI Communications Infrastructure — voice, messaging, SIP, AI, IoT, and Cloud Storage on one private, global network. This sample uses four Telnyx products through a single API key and one deploy command. The Agent SDK's [telnyx] binding means SMS delivery needs zero credentials in code, while KV and Cloud Storage use the same key for per-location cache flags and shared manifest storage.

Build an Audio Transcription AI Summary SMS Pipeline with the Telnyx Agent SDK

Harpreet Singh Seehra — Wed, 19 Aug 2026 10:10:19 +0000

Build an Audio Transcription → AI Summary → SMS Pipeline with the Telnyx Agent SDK

Upload a voicemail audio file → transcribe it via speech-to-text → summarize with an LLM → text the summary to any phone number. An end-to-end pipeline on Telnyx Edge Compute with zero-credential inference and messaging.

The Voicemail Problem

Voicemails pile up. You get one after a missed call, listen to it once, and then it sits in your inbox. If you need the details later, you have to listen again. If you want to share the content, you have to paraphrase it or forward the audio. For teams that handle high volumes of voice messages, this friction compounds into real operational drag.

The standard approach to solving this involves stitching together multiple services. A cloud storage provider holds the audio. A transcription API converts speech to text. An LLM provider summarizes the text. An SMS gateway delivers the result. Four vendors, four API keys, four bills, four integration points, and four places where something can break.

The Audio Transcribe → Summarize → SMS example does all of it on one network. Telnyx Cloud Storage holds the audio. Telnyx AI Inference runs speech-to-text and summarization. Telnyx Messaging delivers the summary as an SMS. Telnyx Edge Compute orchestrates the entire pipeline. One API key for storage and STT. Zero credentials in code for inference and messaging. One deploy command.

What It Does

You upload a voicemail audio file to an Edge Compute endpoint. The pipeline runs three stages in sequence:

Transcribe — Download the audio from Cloud Storage and send it to the Telnyx speech-to-text API.
Summarize — Send the transcript to an LLM with a prompt that produces an SMS-friendly summary.
Notify — Send the summary as an SMS to the recipient phone number you provided at upload time.

Each stage is a queued job on a stateful actor. State survives across stages, so the transcript from stage 1 is available to stage 2, and the summary from stage 2 is available to stage 3. You can check the status of any pipeline run via a REST endpoint and get back the full transcript, summary, and completion status.

Stage	Action	API
Upload	Receive multipart form, upload to Cloud Storage (S3 PUT, SigV4)	`POST /upload`
Start	Create actor, store metadata, queue transcribe	`VoicemailAgent.start()`
1. Transcribe	S3 GET download → STT → store transcript	`POST /v2/ai/audio/transcriptions`
2. Summarize	LLM chat completion → store summary	`this.env.TELNYX.ai.openai.chat.createCompletion()`
3. Notify	SMS summary to recipient	`this.env.TELNYX.messages.send()`
Status	Check pipeline progress	`GET /status/:agentId`

The Architecture

The application is a TypeScript Agent SDK project deployed to Telnyx Edge Compute. It uses four Telnyx products: Edge Compute (hosting and actor runtime), AI Inference (STT and LLM), Messaging (SMS), and Cloud Storage (audio file persistence).

  Upload audio file (POST /upload)
        │
        ▼
  ┌──────────────────────────────────────────┐
  │ Upload to Cloud Storage (S3 PUT, SigV4)    │
  └────────┬─────────────────────────────────┘
           │
           ▼
  ┌──────────────────────────────────────────┐
  │ VoicemailAgent.start()                     │
  │  → this.queue("transcribe")                │
  │  → this.queue("summarize")                 │
  │  → this.queue("notify")                    │
  └────────┬─────────────────────────────────┘
           │
           ▼
  Stage 1: transcribe()
    → Download audio from Cloud Storage (S3 GET)
    → POST /v2/ai/audio/transcriptions (STT)
    → Store transcript in agent state
           │
           ▼
  Stage 2: summarize()
    → this.env.TELNYX.ai.openai.chat.createCompletion()
    → Store summary in agent state
           │
           ▼
  Stage 3: notify()
    → this.env.TELNYX.messages.send()
    → SMS summary delivered to recipient

The upload handler receives a multipart form with the audio file and a recipient phone number. It uploads the file to Telnyx Cloud Storage via S3 PUT with AWS SigV4 signing using the Web Crypto API. The S3 key includes a timestamp prefix so uploads are naturally ordered. Once the file is stored, the handler creates a VoicemailAgent actor instance, stores the audio key, bucket name, sender phone, and recipient phone in the actor's durable state, and queues the first pipeline stage.

Zero-Credential Inference and Messaging

The most important design decision in this pipeline is the [telnyx] binding in telnyx.toml:

[telnyx]
binding = "TELNYX"

This single line gives the Agent SDK runtime a pre-authenticated Telnyx client. At runtime, the actor accesses it through this.env.TELNYX. Two critical operations use this binding with no API key in code:

LLM summarization:

const completion = await this.env.TELNYX.ai.openai.chat.createCompletion({
  model: this.env.AI_MODEL || "zai-org/GLM-5.2",
  messages: [
    { role: "system", content: "Summarize this voicemail transcript into a concise SMS-friendly message." },
    { role: "user", content: transcript }
  ],
  max_tokens: 200,
  temperature: 0.3
});
const summary = completion.choices[0].message.content;

SMS delivery:

await this.env.TELNYX.messages.send({
  from: this.env.SENDER_PHONE,
  to: this.state.recipientPhone,
  text: summary
});

No Authorization header. No TELNYX_API_KEY in the source. The binding carries the authentication context from the Edge Compute platform. This means the inference and messaging credentials are managed by the platform, not by the developer. Rotate keys in the portal and the binding picks up the new credentials automatically. No code changes. No redeploy.

Cloud Storage and the speech-to-text API still need an explicit API key because they use S3-compatible SigV4 signing and direct HTTP calls. But that is the same TELNYX_API_KEY you already have. One key for storage and STT. Zero keys for inference and SMS.

Durable State Across Pipeline Stages

The Agent SDK provides durable state through this.setState() and this.getState(). State is scoped to the actor instance and survives across queued invocations:

// In VoicemailAgent.start()
await this.setState({
  audioKey: "voicemails/1724359200000-voicemail.wav",
  bucket: this.env.STORAGE_BUCKET,
  recipientPhone: "+17177247292",
  senderPhone: this.env.SENDER_PHONE,
  status: "queued",
  transcript: "",
  summary: "",
  error: ""
});
this.queue("transcribe");

// In transcribe()
const state = await this.getState();
// ... download audio, run STT ...
state.transcript = transcriptText;
state.status = "transcribed";
await this.setState(state);
this.queue("summarize");

// In summarize()
const state = await this.getState();
// ... run LLM ...
state.summary = summaryText;
state.status = "summarized";
await this.setState(state);
this.queue("notify");

State is not held in memory between stages. It is persisted to the actor store and reloaded on the next invocation. This means the pipeline is resilient to runtime restarts, scaling events, or delays between stages. The state schema is a plain object, so you can store whatever the pipeline needs.

The Pipeline Queue Primitives

this.queue("transcribe");
this.queue("summarize");
this.queue("notify");

These three calls do not run in parallel. They run in the order queued, with state persisting between each. If transcribe() fails, summarize() never runs. If summarize() succeeds but notify() fails, the summary is still saved in state and you can retry the notify stage manually or inspect the error.

The queue is durable. If the Edge Compute worker restarts between stages, the queue is restored and processing resumes from the next pending stage.

Setup

Prerequisites

Telnyx Edge CLI v0.2.2+
Node.js 18+
A Telnyx Cloud Storage bucket (create in the Portal)
A Telnyx phone number with SMS capability

1. Install dependencies

npm install

2. Set secrets

telnyx-edge secret set TELNYX_API_KEY KEY0123456789ABCDEF
telnyx-edge secret set STORAGE_BUCKET my-voicemail-bucket
telnyx-edge secret set SENDER_PHONE +18005551234

TELNYX_API_KEY is your Telnyx API v2 key from the Portal. STORAGE_BUCKET is the name of the Cloud Storage bucket you created. SENDER_PHONE is a Telnyx number with SMS enabled.

Optional secrets:

telnyx-edge secret set STORAGE_REGION us-central-1
telnyx-edge secret set AI_MODEL zai-org/GLM-5.2

3. Deploy

telnyx-edge ship

The ship command builds the TypeScript project, uploads the bundle, and prints a URL like audio-transcribe-summarize-sms-<id>.telnyxcompute.com.

4. Test

Health check:

curl https://audio-transcribe-summarize-sms-<id>.telnyxcompute.com/health/liveness

Upload a voicemail and trigger the pipeline:

curl -X POST https://audio-transcribe-summarize-sms-<id>.telnyxcompute.com/upload \
  -F "file=@voicemail.wav" \
  -F "recipient_phone=+17177247292"

Check pipeline status:

curl https://audio-transcribe-summarize-sms-<id>.telnyxcompute.com/status/<agentId>

API Reference

`POST /upload`

Upload an audio file and trigger the transcribe → summarize → SMS pipeline.

curl -X POST https://audio-transcribe-summarize-sms-<id>.telnyxcompute.com/upload \
  -F "file=@voicemail.wav" \
  -F "recipient_phone=+17177247292"

Response:

{
  "action": "queued",
  "audioKey": "voicemails/1724359200000-voicemail.wav",
  "agentId": "voicemails-1724359200000-voicemail.wav",
  "recipientPhone": "+17177247292",
  "statusUrl": "/status/voicemails-1724359200000-voicemail.wav"
}

`GET /status/:agentId`

Check the pipeline status for a given upload.

curl https://audio-transcribe-summarize-sms-<id>.telnyxcompute.com/status/voicemails-1724359200000-voicemail.wav

Response:

{
  "audioKey": "voicemails/1724359200000-voicemail.wav",
  "bucket": "my-voicemail-bucket",
  "recipientPhone": "+17177247292",
  "senderPhone": "+18005551234",
  "transcript": "Hi, this is John. I'm calling about the invoice from last week...",
  "summary": "John called about an invoice from last week. He wants a callback to discuss it.",
  "status": "done",
  "error": "",
  "createdAt": 1724359200000,
  "completedAt": 1724359210000
}

`GET /health/{liveness,readiness}`

Health checks for load balancers and monitoring.

curl https://audio-transcribe-summarize-sms-<id>.telnyxcompute.com/health/liveness

Frequently Asked Questions

Q: Why does inference and messaging need zero credentials while Cloud Storage needs an API key?

The [telnyx] binding in telnyx.toml is a platform-managed authentication channel. Edge Compute injects a pre-authenticated client into this.env.TELNYX at runtime. This client can call Telnyx AI Inference and Messaging because those services are part of the same platform identity. Cloud Storage, however, uses S3-compatible SigV4 signing, which requires an explicit access key and secret. The Telnyx API key serves as that credential for S3 operations. One key covers both Cloud Storage and the speech-to-text API.

Q: What audio formats are supported for transcription?

The Telnyx Audio Transcriptions API accepts common formats including WAV, MP3, and OGG. The example uses WAV for simplicity. The API reference has the full list of supported codecs and sample rates.

Q: Can I change the LLM model for summarization?

Yes. Set the AI_MODEL secret to any model available on Telnyx AI Inference. The default is zai-org/GLM-5.2, a general-purpose model that works well for summarization. If you switch to a reasoning model, you may need to adjust the prompt because some reasoning models return chain-of-thought text before the final answer.

Q: What happens if a pipeline stage fails?

Q: Is the pipeline synchronous or asynchronous?

The upload endpoint returns immediately with a queued status. The pipeline runs asynchronously in the background. For a typical 30-second voicemail, the full pipeline (transcribe + summarize + SMS) completes in a few seconds. You poll the status endpoint or build a webhook callback if you need real-time notification.

Q: Can I send the summary to multiple phone numbers?

The example sends to a single recipient per upload. To support multiple recipients, modify the upload handler to accept an array of phone numbers and loop over them in the notify stage. The this.env.TELNYX.messages.send() call is idempotent enough that you can invoke it once per recipient.

AI Voice Agent with Function Calling — Calling Real APIs Mid-Conversation in Python

Harpreet Singh Seehra — Fri, 07 Aug 2026 13:22:25 +0000

A phone-call voice agent that calls real APIs (weather, orders, balances) in the middle of a conversation. Speech recognition, LLM tool-calling, and Telnyx text-to-speech in one Flask file. One API key for calls, AI inference, and TTS.

The Problem: Voice Agents That Lie

You have an AI voice agent. It picks up calls. It greets callers. It responds in natural language.

But the moment someone asks, "What's the weather in San Francisco?", "Where is my order 12345?", or "What's my account balance?" — the agent hallucinates. It guesses. It invents a plausible-sounding answer with zero real data.

Voice agents without function calling are LLMs talking to thin air. They have no way to reach out for live information mid-conversation. They just pattern-match.

The ai-voice-agent-with-function-calling-python example fixes this in one Flask file. It wires three Telnyx capabilities — speech recognition via gather_using_ai, LLM tool-calling via AI Inference, and text-to-speech via speak — into a single webhook-driven loop. Callers ask questions in plain speech. The agent calls real functions. It speaks real answers back.

What It Does

Call your agent number. It picks up, greets you, and waits. You speak a request: "What's the weather in San Francisco?" The voice agent transcribes your speech via gather_using_ai, sends the transcript to Telnyx AI Inference with three tool definitions (check_weather, lookup_order, check_account_balance), and the model decides whether to call a tool or respond directly.

If the model calls a tool — say check_weather — your Python function runs, returns its JSON result to the model, and the model synthesizes a one-sentence spoken answer. speak() reads it aloud in natural voice. The conversation loop continues. You ask about an order. The agent calls lookup_order. You ask about your balance. The agent calls check_account_balance. No hallucinations, no guessing — every tool-backed answer is backed by real function output.

Step	What happens	Telnyx API
1	Caller dials agent number	Inbound call → `call.initiated` webhook
2	Agent answers + greets	`answer()` + `speak()` (TTS)
3	TTS ends → start listening	`call.speak.ended` → `gather_using_ai()`
4	Caller speaks a request	Speech-to-text via `gather_using_ai`
5	Transcript returned	`call.ai_gather.ended` webhook
6	Transcript → AI Inference (with tools)	`POST /v2/ai/chat/completions`
7	Model returns `tool_calls`	Loop: execute functions, re-infer
8	Model returns final text	`speak()` reads it aloud
9	Caller hangs up	`call.hangup` → cleanup

The Architecture

One Flask file. One webhook endpoint. In-memory conversation state keyed by call_control_id. No database, no Redis, no background workers. Telnyx owns the telephony, speech recognition, and TTS layers. Your code owns the conversation state and the function implementations.

Inbound call → Telnyx webhook → /webhooks/voice
                              ↓
                 call.initiated → answer() + greet (speak)
                              ↓
                 call.speak.ended → gather_using_ai()
                              ↓
                 call.ai_gather.ended → transcript
                              ↓
                 Transcript → AI Inference (with TOOLS array)
                              ↓
                 Model returns tool_calls?
                  ├── yes → execute_function() → re-infer → loop
                  └── no  → final text response
                              ↓
                 speak() reads response aloud
                              ↓
                 call.speak.ended → gather_using_ai() → loop
                              ↓
                 call.hangup → cleanup

The Tools: OpenAI-Style Function Calling

The agent has three tools, defined in the OpenAI function-calling schema:

TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "check_weather",
            "description": "Get current weather for a city",
            "parameters": {
                "type": "object",
                "properties": {"city": {"type": "string"}},
                "required": ["city"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "lookup_order",
            "description": "Look up order status by order number",
            "parameters": {
                "type": "object",
                "properties": {"order_id": {"type": "string"}},
                "required": ["order_id"],
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "check_account_balance",
            "description": "Check account balance by account number",
            "parameters": {
                "type": "object",
                "properties": {"account_id": {"type": "string"}},
                "required": ["account_id"],
            },
        },
    },
]

The model sees these tools every inference call. When the user asks "What's the weather in San Francisco?", the model returns a tool_calls array containing check_weather with {"city": "San Francisco"}. Your code runs execute_function("check_weather", {"city": "San Francisco"}), returns the JSON result, and re-sends the conversation to the model. The model then synthesizes a spoken answer: "The weather in San Francisco is 72°F and partly cloudy with 45% humidity."

The mock implementations in execute_function are intentionally simple — replace them with real API calls to your weather provider, order management system, or billing platform:

def execute_function(name, args):
    if name == "check_weather":
        return json.dumps({"city": args.get("city"), "temp": "72F",
                           "condition": "Partly cloudy", "humidity": "45%"})
    elif name == "lookup_order":
        return json.dumps({"order_id": args.get("order_id"), "status": "shipped",
                           "eta": "June 20", "carrier": "FedEx"})
    elif name == "check_account_balance":
        return json.dumps({"account_id": args.get("account_id"), "balance": "$1,234.56",
                           "due_date": "July 1"})
    return json.dumps({"error": "Unknown function"})

The Conversation Loop

The webhook handler is the heartbeat. Each event transitions the state machine:

@app.route("/webhooks/voice", methods=["POST"])
def handle_voice():
    # Verify the Telnyx Ed25519 signature before trusting the event.
    try:
        client.webhooks.unwrap(request.get_data(as_text=True), headers=dict(request.headers))
    except Exception:
        return jsonify({"error": "invalid signature"}), 401

    payload = request.get_json()
    data = payload.get("data", {})
    p = data.get("payload", {})
    event_type = data.get("event_type")
    ccid = p.get("call_control_id")
    call = active_calls.get(ccid)

    if event_type == "call.initiated" and p.get("direction") == "incoming":
        active_calls[ccid] = {
            "caller": p.get("from"),
            "conversation": [{"role": "system", "content": SYSTEM_PROMPT}],
            "_ts": time.time(),
        }
        client.calls.actions.answer(ccid)
        return jsonify({"status": "answering"}), 200

    elif event_type == "call.answered":
        client.calls.actions.speak(ccid, payload=GREETING, voice=VOICE, language="en-US")
        return jsonify({"status": "greeting"}), 200

    elif event_type == "call.speak.ended" and call:
        if call.get("processed"):
            call["processed"] = False
        client.calls.actions.gather_using_ai(
            ccid,
            parameters={
                "type": "object",
                "properties": {
                    "user_request": {
                        "type": "string",
                        "description": "What the caller said — their full spoken request.",
                    }
                },
                "required": ["user_request"],
            },
            voice=VOICE,
            language="en-US",
            user_response_timeout_ms=15000,
        )
        return jsonify({"status": "listening"}), 200

    elif event_type == "call.ai_gather.ended" and call:
        if call.get("processed"):
            return jsonify({"status": "ok"}), 200
        call["processed"] = True

        result = p.get("result", {})
        speech = result.get("user_request", "") if isinstance(result, dict) else ""

        if not speech:
            for msg in reversed(p.get("message_history", [])):
                if msg.get("role") == "user":
                    speech = msg.get("content", "")
                    break

        if not speech:
            client.calls.actions.speak(ccid, payload=REPROMPT, voice=VOICE, language="en-US")
            return jsonify({"status": "reprompting"}), 200

        call["conversation"].append({"role": "user", "content": speech})
        response = call_inference(call["conversation"])
        call["conversation"].append({"role": "assistant", "content": response})

        client.calls.actions.speak(ccid, payload=response, voice=VOICE, language="en-US")
        return jsonify({"status": "responding"}), 200

    elif event_type == "call.hangup":
        active_calls.pop(ccid, None)
        return jsonify({"status": "ended"}), 200

    return jsonify({"status": "ok"}), 200

Three things to note. First, the signature verification via client.webhooks.unwrap() — never trust an unverified webhook. Second, the call["processed"] dedup guard — Telnyx retries webhooks, and without it you would speak the same response twice. Third, the call.ai_gather.ended handler extracts speech from result.user_request with a message_history fallback, because the gather result shape varies by SDK version.

The Inference Function: Tool-Calling Loop

The call_inference function is the AI brain. It sends the conversation to the model with the TOOLS array. If the model returns tool_calls, the function executes each one, appends the results to the conversation, and recurses. If the model returns a plain text response, that's the spoken answer.

def call_inference(messages, max_tokens=300, _depth=0, _max_depth=5):
    if _depth >= _max_depth:
        return "I'm having trouble processing that request right now."

    payload = {
        "model": AI_MODEL,
        "messages": messages,
        "temperature": 0.5,
        "tools": TOOLS,
    }
    try:
        resp = requests.post(
            INFERENCE_URL,
            headers={"Authorization": f"Bearer {TELNYX_API_KEY}",
                     "Content-Type": "application/json"},
            json=payload,
            timeout=30,
        )
    except Exception as e:
        app.logger.error("Inference request failed: %s", e)
        return "I couldn't reach the AI service just now. Please try again."

    try:
        resp.raise_for_status()
    except Exception as e:
        app.logger.error("Inference HTTP error: %s — %s", e, resp.text[:200])
        return "The AI service returned an error. Please try again."

    choice = resp.json()["choices"][0]
    msg = choice["message"]

    if msg.get("tool_calls"):
        for tc in msg["tool_calls"]:
            fn = tc["function"]
            try:
                fn_args = json.loads(fn.get("arguments", "{}"))
            except json.JSONDecodeError:
                fn_args = {}
            result = execute_function(fn["name"], fn_args)
            messages.append(msg)
            messages.append({"role": "tool", "tool_call_id": tc["id"], "content": result})
        return call_inference(messages, max_tokens, _depth=_depth + 1, _max_depth=_max_depth)

    return _strip_fences(msg["content"])

The _max_depth=5 recursion guard prevents infinite loops if the model keeps calling tools forever. The _strip_fences() helper strips

```json

code fences from the response so TTS doesn't read them aloud — without this, your voice agent would say "triple backslash triple json" before every answer.

What the Original Sample Got Wrong

This example was the most broken upstream sample in the set. The original app.py had six critical bugs that prevented the app from working at all. We fixed all of them in the fix PR.

Bug 1: gather() is DTMF-only. The original code called gather(input_type="speech", end_silence_timeout_secs=3, language_code="en-US"). None of those parameters are valid — gather() only collects DTMF keypresses. Passing speech params causes a TypeError. The fix is gather_using_ai(), which transcribes free-form speech and fires call.ai_gather.ended instead of call.gather.ended.

Bug 2: voice="female" is invalid. The speak() action requires voice in <Provider>.<Model>.<VoiceId> format (e.g., Telnyx.KokoroTTS.af). Passing "female" causes an API error.

Bug 3: call_inference() UnboundLocalError. If requests.post() raised an exception, the variable resp was never assigned. The next line, resp.raise_for_status(), then crashed with UnboundLocalError: local variable 'resp' referenced before assignment. The fix wraps both calls in try/except blocks and returns a spoken error string.

Bug 4: No recursion depth guard. The original call_inference() recursed on tool_calls with no max depth. A misbehaving model could loop forever. The fix adds _depth and _max_depth=5.

Bug 5: base_url not overridden. The Telnyx Python SDK reads the TELNYX_BASE_URL environment variable, which in some environments routes API calls to a proxy. The fix is explicit base_url="https://api.telnyx.com/v2" in the client constructor.

Bug 6: Markdown fences in TTS. Some models wrap JSON in

```json

fences. Without stripping, speak() reads the fences aloud. The fix is _strip_fences().

One API Key for Voice, AI, and TTS

The entire app uses a single TELNYX_API_KEY:

Voice (Call Control) — answer(), speak(), gather_using_ai() via the Telnyx SDK
AI Inference — POST /v2/ai/chat/completions via requests with Bearer auth
Text-to-speech — handled by speak() (uses the same API key, no separate TTS provider)
Webhook signature verification — client.webhooks.unwrap() validates Ed25519 signatures

No third-party speech-to-text provider. No separate LLM API key. No separate TTS provider. One network, one key, one bill.

The `gather_using_ai` vs `gather` Distinction

This is the most subtle and important distinction in the Call Control API. Two methods with similar names, completely different capabilities:

Method	Input type	Webhook event	Use case
`gather()`	DTMF only (keypad digits)	`call.gather.ended`	"Press 1 for sales, 2 for support" IVR menus
`gather_using_ai()`	Free-form speech	`call.ai_gather.ended`	Natural language voice agents
`gather_using_speak()`	DTMF + TTS prompt	`call.gather.ended`	Spoken IVR prompts with DTMF response

The upstream sample used gather() with speech params that don't exist — a classic copy-paste-from-docs mistake. The fix is gather_using_ai(), which returns transcribed speech in the result.user_request field of the call.ai_gather.ended payload.

Environment Variables

TELNYX_API_KEY=your_api_key_here
TELNYX_PUBLIC_KEY=your_public_key_here
AI_MODEL=moonshotai/Kimi-K2.6
AGENT_NUMBER=+15551234567
CONNECTION_ID=your_connection_id
PORT=5000

TELNYX_API_KEY — your Telnyx API v2 key (Portal → API Keys)
TELNYX_PUBLIC_KEY — your Telnyx public key (used for webhook signature verification)
AI_MODEL — any model on Telnyx AI Inference (default: moonshotai/Kimi-K2.6)
AGENT_NUMBER — the phone number callers dial
CONNECTION_ID — your Call Control Application ID (Portal → Call Control → Applications)
PORT — HTTP port for the Flask server

Try It Yourself

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/ai-voice-agent-with-function-calling-python
cp .env.example .env   # add TELNYX_API_KEY, TELNYX_PUBLIC_KEY, AGENT_NUMBER, CONNECTION_ID
pip install -r requirements.txt
python app.py           # starts on http://localhost:5000

Expose your local server with ngrok and configure the webhook URL in your Call Control Application:

ngrok http 5000
# Copy the HTTPS URL → Portal → Call Control → Application → Webhook URL
# Set to: https://<id>.ngrok.io/webhooks/voice

Call your agent number. You'll hear the greeting. Ask: "What's the weather in San Francisco?" The agent will call the check_weather tool and speak the result. Ask: "Where is order 12345?" The agent will call lookup_order. Ask: "What's my account balance for account 67890?" The agent will call check_account_balance. Every answer is backed by real function output, not LLM hallucination.

Key links:

Repo: https://github.com/team-telnyx/telnyx-code-examples/tree/main/ai-voice-agent-with-function-calling-python
Telnyx Portal: https://portal.telnyx.com