DEV Community: Harpreet Singh Seehra

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

Browser Phone Calls with Real-Time AI Coaching

Harpreet Singh Seehra — Thu, 06 Aug 2026 13:39:27 +0000

Click-to-Call WebRTC with AI Assist — a split-screen browser app where you make real phone calls from your browser and get live AI coaching tips in a sidebar. WebRTC telephony credentials, browser SpeechRecognition, and Telnyx AI Inference. One API key for calls and AI. No phone needed.

The Problem: Sales Calls Without Feedback

You are on a sales call. You are pitching, handling objections, trying to close. You have no idea if you are talking too fast, missing buying signals, or failing to ask for the next step. Your manager is not on the line. There is no coach whispering in your ear. You are flying blind.

The existing solutions are clunky. Call recording platforms store the audio for later review, but "later" means after the deal is lost. AI note-takers join as a third party on the call, which creeps out prospects and sometimes breaks compliance. Real-time coaching tools exist, but they require integrations with your dialer, your CRM, and your conferencing platform. Three vendors, three API keys, three bills.

The Click-to-Call WebRTC with AI Assist example fixes this in 75 lines of Python. Open a browser tab. Enter a phone number. Click Call. You are talking to a real phone number from your browser, and an AI coach is watching your transcript and feeding you actionable tips every 8 seconds in a live sidebar.

What It Does

Open http://localhost:5000 in your browser. You see a split-screen UI: a call panel on the left and an AI coaching sidebar on the right.

Enter a phone number and click Call. The backend creates a WebRTC telephony credential via POST /v2/telephony_credentials. The frontend uses the Telnyx WebRTC SDK to connect via SIP. The WebRTC call connects — you are now talking to a real phone number from your browser, no desk phone, no softphone app, just the browser.

While you talk, the browser SpeechRecognition API transcribes your speech in real time. Every 8 seconds, the transcript is sent to the /coaching endpoint, which calls Telnyx AI Inference and returns one actionable coaching tip. The tip appears in the right sidebar with a timestamp — live AI coaching during the call.

Click Hang Up to end. The call disconnects, transcription stops, coaching stops.

Step	What happens	Tech
1	User clicks Call	Frontend sends number to backend
2	Backend creates WebRTC credential	`POST /v2/telephony_credentials`
3	Frontend connects via WebRTC	`@telnyx/webrtc` SDK, SIP registration
4	Call connects	WebRTC audio to PSTN
5	Browser transcribes speech	`SpeechRecognition` API, real-time
6	Every 8s, transcript sent to AI	`POST /v2/ai/chat/completions`
7	AI returns coaching tip	Model: `moonshotai/Kimi-K2.6`
8	Tip appears in sidebar	DOM update with timestamp
9	User clicks Hang Up	WebRTC disconnect, cleanup

The Architecture

Everything lives in one Flask file (~75 lines) and one HTML template. No database, no Redis, no WebSocket server. The browser handles speech recognition. The backend handles WebRTC credentials and AI inference.

Browser opens http://localhost:5000
        ↓
User enters number, clicks Call
        ↓
Frontend POST /webrtc/token → Backend creates telephony credential
        ↓
Backend returns SIP username, password, caller_number
        ↓
Frontend TelnyxRTC client.connect() via WebRTC
        ↓
WebRTC call connects to PSTN number
        ↓
Browser SpeechRecognition starts → live transcript in left panel
        ↓
Every 8s: transcript POST /coaching → AI Inference → coaching tip
        ↓
Tip appears in right sidebar with timestamp
        ↓
User clicks Hang Up → client.disconnect() → cleanup

The Backend: Flask in 75 Lines

The backend has four endpoints. Here is the full app.py:

import os, time, requests
from flask import Flask, request, jsonify, render_template

app = Flask(__name__)

TELNYX_API_KEY = os.getenv("TELNYX_API_KEY")
AI_MODEL = os.getenv("AI_MODEL", "moonshotai/Kimi-K2.6")
WEBRTC_CONNECTION_ID = os.getenv("WEBRTC_CONNECTION_ID")
CALLER_NUMBER = os.getenv("CALLER_NUMBER")
INFERENCE_URL = "https://api.telnyx.com/v2/ai/chat/completions"

@app.route("/")
def index():
    return render_template("index.html")

@app.route("/webrtc/token", methods=["POST"])
def webrtc_token():
    """Create a WebRTC telephony credential and return SIP credentials."""
    resp = requests.post(
        "https://api.telnyx.com/v2/telephony_credentials",
        headers={"Authorization": f"Bearer {TELNYX_API_KEY}",
                 "Content-Type": "application/json"},
        json={"connection_id": WEBRTC_CONNECTION_ID,
              "expires_secs": 3600},
        timeout=10)
    resp.raise_for_status()
    data = resp.json()["data"]
    return jsonify({
        "sip_username": data["sip_username"],
        "sip_password": data["sip_password"],
        "caller_number": CALLER_NUMBER
    })

@app.route("/coaching", methods=["POST"])
def coaching():
    """Send transcript to AI Inference and return a coaching tip."""
    transcript = request.json.get("transcript", "")
    if not transcript:
        return jsonify({"tip": "Start speaking to get coaching tips."})

    messages = [
        {"role": "system", "content":
            "You are a sales coach. Review the call transcript and give ONE "
            "actionable tip to improve the call. Be specific, concise, and "
            "practical. Focus on: asking better questions, handling objections, "
            "closing techniques, or tone adjustments. Return only the tip, no preamble."},
        {"role": "user", "content": f"Call transcript so far:\n{transcript}"}
    ]

    resp = requests.post(INFERENCE_URL,
        headers={"Authorization": f"Bearer {TELNYX_API_KEY}",
                 "Content-Type": "application/json"},
        json={"model": AI_MODEL, "messages": messages,
              "max_tokens": 150, "temperature": 0.7},
        timeout=15)
    resp.raise_for_status()
    tip = resp.json()["choices"][0]["message"]["content"]
    return jsonify({"tip": tip, "timestamp": time.strftime("%H:%M:%S")})

@app.route("/health")
def health():
    return jsonify({"status": "ok"})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=int(os.getenv("PORT", 5000)))

That is the entire backend. Four endpoints, one inference call, one credential creation call. No session management, no database, no background workers.

WebRTC Telephony Credentials

The /webrtc/token endpoint creates a temporary SIP credential via POST /v2/telephony_credentials:

resp = requests.post(
    "https://api.telnyx.com/v2/telephony_credentials",
    headers={"Authorization": f"Bearer {TELNYX_API_KEY}",
             "Content-Type": "application/json"},
    json={"connection_id": WEBRTC_CONNECTION_ID,
          "expires_secs": 3600},
    timeout=10)

The connection_id links the credential to your Telnyx WebRTC connection (configured in the portal). The credential expires in 1 hour. The response contains sip_username and sip_password, which the frontend uses to register with Telnyx's SIP server over WebRTC.

This is the key abstraction: Telnyx handles the SIP infrastructure, the STUN/TURN servers, the NAT traversal, and the PSTN gateway. Your code just asks for a credential and passes it to the WebRTC SDK.

AI Coaching Endpoint

The /coaching endpoint sends the transcript to Telnyx AI Inference and returns a coaching tip:

messages = [
    {"role": "system", "content":
        "You are a sales coach. Review the call transcript and give ONE "
        "actionable tip to improve the call. Be specific, concise, and "
        "practical. Focus on: asking better questions, handling objections, "
        "closing techniques, or tone adjustments. Return only the tip, no preamble."},
    {"role": "user", "content": f"Call transcript so far:\n{transcript}"}
]

resp = requests.post(INFERENCE_URL,
    headers={"Authorization": f"Bearer {TELNYX_API_KEY}",
             "Content-Type": "application/json"},
    json={"model": AI_MODEL, "messages": messages,
          "max_tokens": 150, "temperature": 0.7},
    timeout=15)

The system prompt is specific: one tip, actionable, no preamble. Temperature is 0.7 — high enough for creative, varied coaching tips, low enough that they stay relevant. Max tokens 150 — enough for a sentence or two, not enough for the model to ramble.

The default model is moonshotai/Kimi-K2.6, but you can override via the AI_MODEL environment variable. Any model available on Telnyx AI Inference works.

The Frontend: Browser WebRTC + SpeechRecognition

The frontend is a single HTML template served by Flask. It loads the Telnyx WebRTC SDK from CDN and uses the browser's native SpeechRecognition API.

WebRTC Connection with TelnyxRTC

<script src="https://unpkg.com/@telnyx/webrtc@2/dist/telnyx.js"></script>
<script>
const client = new TelnyxRTC({
    login: sipUsername,
    password: sipPassword,
    ringtoneFile: null,
    ringbackFile: null
});

client.on('telnyx.ready', () => {
    statusBadge.textContent = 'Ready';
    statusBadge.className = 'badge ready';
});

client.on('telnyx.error', (error) => {
    statusBadge.textContent = 'Error: ' + error.message;
});

async function startCall() {
    const resp = await fetch('/webrtc/token', {method: 'POST'});
    const creds = await resp.json();
    sipUsername = creds.sip_username;
    sipPassword = creds.sip_password;
    callerNumber = creds.caller_number;

    client.login = sipUsername;
    client.password = sipPassword;
    client.connect();

    client.on('telnyx.ready', () => {
        const call = client.newCall({
            destinationNumber: document.getElementById('phoneNumber').value,
            callerNumber: callerNumber
        });
        activeCall = call;
        startTranscription();
        startCoachingLoop();
        startTimer();
    });
}
</script>

The TelnyxRTC client handles SIP registration, ICE negotiation, and media setup. You pass it the SIP username and password from /webrtc/token, call connect(), and wait for the telnyx.ready event. Then you create a new call with newCall({destinationNumber, callerNumber}).

Browser SpeechRecognition for Live Transcript

const recognition = new (window.SpeechRecognition || window.webkitSpeechRecognition)();
recognition.continuous = true;
recognition.interimResults = true;
recognition.lang = 'en-US';

let transcript = '';

recognition.onresult = (event) => {
    let interim = '';
    for (let i = event.resultIndex; i < event.results.length; i++) {
        const result = event.results[i];
        if (result.isFinal) {
            transcript += result[0].transcript + ' ';
            document.getElementById('transcript').textContent = transcript;
        } else {
            interim += result[0].transcript;
        }
    }
    document.getElementById('interim').textContent = interim;
};

function startTranscription() {
    transcript = '';
    recognition.start();
}

The SpeechRecognition API is built into Chrome and Safari. It streams partial results in real time (interimResults: true) and finalizes phrases when the speaker pauses. The transcript accumulates in a string that is displayed in the left panel and sent to the coaching endpoint every 8 seconds.

The AI Coaching Loop

let coachingInterval;

function startCoachingLoop() {
    coachingInterval = setInterval(async () => {
        if (!transcript.trim()) return;
        const resp = await fetch('/coaching', {
            method: 'POST',
            headers: {'Content-Type': 'application/json'},
            body: JSON.stringify({transcript})
        });
        const data = await resp.json();
        addCoachingTip(data.tip, data.timestamp);
    }, 8000);
}

function addCoachingTip(tip, timestamp) {
    const sidebar = document.getElementById('coachingSidebar');
    const div = document.createElement('div');
    div.className = 'coaching-tip';
    div.innerHTML = `<span class="timestamp">${timestamp}</span> ${tip}`;
    sidebar.appendChild(div);
    sidebar.scrollTop = sidebar.scrollHeight;
}

Every 8 seconds, the current transcript is sent to /coaching. The AI reviews the conversation so far and returns one tip. The tip is appended to the sidebar with a timestamp. The sidebar auto-scrolls so the latest tip is always visible.

The 8-second interval is a tradeoff. Shorter intervals mean more responsive coaching but more API calls and higher latency (the AI needs context to give good tips). Longer intervals mean fewer calls but delayed feedback. Eight seconds hits the sweet spot for sales calls: enough context for meaningful tips, frequent enough to feel live.

Call Timer and Status Badges

let callStartTime;
let timerInterval;

function startTimer() {
    callStartTime = Date.now();
    timerInterval = setInterval(() => {
        const elapsed = Math.floor((Date.now() - callStartTime) / 1000);
        const mins = String(Math.floor(elapsed / 60)).padStart(2, '0');
        const secs = String(elapsed % 60).padStart(2, '0');
        document.getElementById('timer').textContent = `${mins}:${secs}`;
    }, 1000);
}

The call timer starts when the call connects and updates every second. Status badges show Ready, Connecting, On Call, and Ended states. These are small UX touches that make the app feel like a real dialer.

One API Key for WebRTC and AI

The entire app uses a single TELNYX_API_KEY:

WebRTC credentials — POST /v2/telephony_credentials (via requests.post with Bearer token)
AI Inference — POST /v2/ai/chat/completions (via requests.post with Bearer token)
Call delivery — handled by Telnyx's PSTN gateway (no additional API calls)

The SDK handles the WebRTC media layer. The two requests.post calls handle credentials and AI directly. No third-party transcription service, no separate LLM provider, no conferencing platform. One network, one key, one bill.

Environment Variables

TELNYX_API_KEY=your_api_key_here
AI_MODEL=moonshotai/Kimi-K2.6
WEBRTC_CONNECTION_ID=your_connection_id
CALLER_NUMBER=+12155551234
PORT=5000

TELNYX_API_KEY — your Telnyx API v2 key
AI_MODEL — any model on Telnyx AI Inference (default: moonshotai/Kimi-K2.6)
WEBRTC_CONNECTION_ID — the connection ID from your Telnyx Portal WebRTC configuration
CALLER_NUMBER — the caller ID for outbound calls (must be a Telnyx number on your account)
PORT — HTTP port (default: 5000)

Try It Yourself

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/click-to-call-webrtc-with-ai-assist-python
cp .env.example .env   # add TELNYX_API_KEY, WEBRTC_CONNECTION_ID, CALLER_NUMBER
pip install -r requirements.txt
python app.py           # starts on http://localhost:5000

Open http://localhost:5000 in Chrome or Safari. Enter a phone number. Click Call. Start talking. Watch the coaching tips appear.

Key links:

Repo: https://github.com/team-telnyx/telnyx-code-examples/tree/main/click-to-call-webrtc-with-ai-assist-python
Telnyx Portal: https://portal.telnyx.com
WebRTC docs: https://developers.telnyx.com/docs/voice/webrtc
AI Inference docs: https://developers.telnyx.com/docs/inference

From Phone Call to Formatted Email in 80 Lines of Python — AI Voice Memo Cleanup with Telnyx

Harpreet Singh Seehra — Wed, 05 Aug 2026 10:44:52 +0000

AI Voice Memo to Email — an 80-line Flask webhook that answers a phone call, gathers a spoken memo, runs it through AI Inference to clean up grammar and extract structure, and delivers a formatted email. One API key for voice, AI, and messaging. No third-party services.

The Voice Memo Problem

Voice memos are the fastest way to capture a thought — you speak, you're done. But what you get is a rambling audio blob that nobody (including you) wants to read later. The raw transcript is worse: no punctuation, false starts, filler words, and no structure. You still have to manually clean it up before it's useful as an email, a status update, or a meeting summary.

The existing solutions split the problem across multiple services. A transcription service converts audio to text. An LLM API cleans up the text. An email service sends the result. Three vendors, three API keys, three bills, three points of failure.

The AI Voice Memo to Email example does all of it on one network — Telnyx Call Control handles the phone call, Telnyx AI Inference cleans up the transcript, and Telnyx Messaging delivers the email. One API key. One Flask file. About 80 lines of Python.

What It Does

You call a Telnyx number. The app answers, speaks a greeting, and starts listening. You dictate your memo — a status update, a meeting summary, a bug report, whatever — and press # when you're done. The app sends the transcript to AI Inference with a prompt that returns structured JSON: a subject line, a formatted body, and a list of action items. The app sends that as an email to your default address and confirms back on the call: "Memo saved and emailed. Subject: [inferred subject]. Goodbye!"

Step	Event	Action
1	`call.initiated` (incoming)	Answer the call, create session
2	`call.answered`	TTS: "Voice memo. Speak your memo after the tone. Press pound when finished."
3	`call.speak.ended`	Start speech gather (120s timeout, `#` terminates)
4	`call.gather.ended`	Send transcript to AI Inference → get structured JSON → send email → TTS confirmation
5	`call.hangup`	Clean up session

The memo is also stored in memory and accessible via GET /memos — so even if email delivery isn't configured, the formatted memo is still retrievable.

The Architecture

Everything lives in one Flask file. No database, no Redis, no Celery. Call state is tracked in an in-memory dict keyed by call_control_id. Memos are stored in a list. A background thread cleans up expired sessions every 5 minutes (1-hour TTL).

Caller dials your Telnyx number
        ↓
Telnyx sends call.initiated webhook → /webhooks/voice
        ↓
app calls answer() → creates session in active_calls[ccid]
        ↓
Telnyx sends call.answered → app calls speak() with greeting
        ↓
Telnyx sends call.speak.ended → app calls gather(input_type="speech", terminating_digit="#")
        ↓
Caller dictates memo, presses #
        ↓
Telnyx sends call.gather.ended with speech transcript
        ↓
app sends transcript to AI Inference → gets JSON {subject, body, action_items}
        ↓
app sends email via Telnyx Messaging API
        ↓
app calls speak() with confirmation: "Memo saved and emailed. Subject: X. Goodbye!"
        ↓
Telnyx sends call.hangup → app removes session

The Call Flow State Machine

The webhook handler is a state machine driven by Telnyx events. Each event triggers the next action:

@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()
    event_type = payload.get("data", {}).get("event_type")
    data = payload.get("data", {})
    p = data.get("payload", {})
    ccid = p.get("call_control_id")

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

    elif event_type == "call.answered":
        client.calls.actions.speak(ccid,
            payload="Voice memo. Speak your memo after the tone. Press pound when finished.",
            voice="female", language_code="en-US")
        return jsonify({"status": "greeting"}), 200

    elif event_type == "call.speak.ended":
        client.calls.actions.gather(ccid,
            input_type="speech", end_silence_timeout_secs=5, timeout_secs=120,
            language_code="en-US", terminating_digit="#")
        return jsonify({"status": "recording"}), 200

    elif event_type == "call.gather.ended":
        call = active_calls.get(ccid)
        speech = p.get("speech", {}).get("result", "")
        if call and speech:
            call["raw_text"].append(speech)
            # ... AI cleanup + email + confirmation
        return jsonify({"status": "processed"}), 200

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

The state machine has five transitions, one per event. The call.initiated handler checks direction == "incoming" to avoid processing outbound call legs. The call.speak.ended handler is what advances from greeting to gathering — Telnyx fires this event when TTS playback finishes, so you know the caller has heard the greeting before the gather starts.

The gather uses end_silence_timeout_secs=5 — if the caller stops speaking for 5 seconds, the gather ends automatically. The timeout_secs=120 caps the total gather at 2 minutes. The terminating_digit="#" lets the caller explicitly signal "I'm done" by pressing pound.

AI-Powered Memo Cleanup

The core of the app is a single inference call that turns rambling speech into structured JSON:

def call_inference(messages, max_tokens=400):
    resp = requests.post(INFERENCE_URL,
        headers={"Authorization": f"Bearer {TELNYX_API_KEY}", "Content-Type": "application/json"},
        json={"model": AI_MODEL, "messages": messages,
              "max_tokens": max_tokens, "temperature": 0.3},
        timeout=15)
    resp.raise_for_status()
    return resp.json()["choices"][0]["message"]["content"]

The system prompt asks for three fields — subject, body, and action_items:

formatted = call_inference([
    {"role": "system", "content":
        "Clean up this voice memo into a well-formatted email. "
        "Fix grammar, add structure (paragraphs, bullets if needed). "
        "Return JSON: subject (string, inferred from content), "
        "body (string, the formatted memo), "
        "action_items (list of strings)."},
    {"role": "user", "content": speech}
])
memo = json.loads(formatted)

Temperature is 0.3 — low enough that the same memo produces roughly the same output every time, but high enough that the AI can infer a reasonable subject line from the content. The max_tokens=400 cap is sufficient for a typical voice memo.

If the AI response isn't valid JSON, the except block saves the raw speech and speaks a simpler confirmation — the caller still gets their memo saved, just without the email:

except Exception:
    memos.append({"raw": speech, "caller": call["caller"],
                  "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ")})
    client.calls.actions.speak(ccid, payload="Memo saved. Goodbye!",
        voice="female", language_code="en-US")

Graceful degradation — the call is never wasted. If AI fails, the raw transcript is preserved. If email fails, the formatted memo is preserved. The caller always gets a confirmation.

Email Delivery via Telnyx Messaging

After the memo is formatted, the app sends it as an email through the Telnyx Messaging API:

def send_email(to, subject, body):
    try:
        requests.post("https://api.telnyx.com/v2/messages",
            headers={"Authorization": f"Bearer {TELNYX_API_KEY}",
                     "Content-Type": "application/json"},
            json={"from": {"email_address": f"memo@{MEMO_NUMBER.replace('+','')}.telnyx.com"},
                  "to": [{"email_address": to}],
                  "subject": subject, "body": body, "type": "email"},
            timeout=15)
    except Exception as e:
        app.logger.error("Email send failed: %s", e)

The same TELNYX_API_KEY that answers the call and runs the AI inference also sends the email — one key, one bill, one network. The email send is wrapped in a try/except because email delivery may require additional Telnyx setup. If it fails, the memo is still saved and retrievable via GET /memos.

Webhook Signature Verification

Every Telnyx webhook is signed with an Ed25519 key. The app verifies the signature before processing the event:

try:
    client.webhooks.unwrap(request.get_data(as_text=True), headers=dict(request.headers))
except Exception:
    return jsonify({"error": "invalid signature"}), 401

The webhooks.unwrap() method from the Telnyx Python SDK handles the Ed25519 verification internally — it reads the telnyx-signature-ed25519 and telnyx-timestamp headers, reconstructs the signed payload, and verifies the signature against the public key. The raw body is verified, not the parsed JSON — because JSON parsing is not canonical, and the signature would fail.

Try It Yourself

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/ai-voice-memo-to-email-python
cp .env.example .env   # add TELNYX_API_KEY, MEMO_NUMBER, DEFAULT_EMAIL
pip install -r requirements.txt
python app.py           # starts on http://localhost:5000

Then:

ngrok http 5000

Configure your Call Control Application webhook URL to https://<id>.ngrok.io/webhooks/voice in the Telnyx Portal.

Call your Telnyx number. Speak your memo. Press #. Check your email.

Check saved memos:

curl http://localhost:5000/memos | python3 -m json.tool

Key links:

Repo: https://github.com/team-telnyx/telnyx-code-examples/tree/main/ai-voice-memo-to-email-python
Telnyx Portal: https://portal.telnyx.com
Call Control docs: https://developers.telnyx.com/docs/voice/call-control
AI Inference docs: https://developers.telnyx.com/docs/inference
Messaging docs: https://developers.telnyx.com/docs/messaging

Build a Real-Time AI Translation Bridge for Phone Calls in 141 Lines of Python

Harpreet Singh Seehra — Tue, 28 Jul 2026 10:04:01 +0000

Connect two callers who speak different languages on the same phone call. Each caller speaks in their own language and hears the other in their own language — live, on the same call, no interpreter in the loop.

No Google Translate. No AWS Translate. No DeepL. No third-party translation API. Just Telnyx Voice Call Control (for the phone call) and Telnyx AI Inference (for the translation). Same API key, same platform, same billing.

141 lines of Python. Here's how it works.

What It Does

You POST two phone numbers and two languages:

curl -X POST http://localhost:5000/bridge \
  -H "Content-Type: application/json" \
  -d '{"number_a": "+13125550001", "lang_a": "English",
       "number_b": "+5215550002", "lang_b": "Spanish"}'

The app calls caller A. When they answer, it calls caller B. When B answers, the translation bridge goes live:

Caller A speaks in English
Telnyx transcribes the speech → sends to your webhook
Your app sends the transcript to AI Inference: "Translate from English to Spanish"
The translated text is spoken to caller B via TTS in Spanish
Caller B speaks in Spanish → transcribed → translated to English → spoken to caller A
Loop until either caller hangs up

Both callers stay on the same phone call. Neither installs anything. Translation happens server-side, transparently.

The Architecture

Everything lives in one Flask file. No database, no Redis, no message queue. Bridge state lives in an in-memory dict, keyed by a bridge ID that travels in Telnyx's client_state field on every webhook event.

POST /bridge → call A → A answers → call B → B answers → bridge active
     ↓
gather A speech → translate → TTS to B → gather B speech → translate → TTS to A → loop
     ↓
hangup → hangup other caller → bridge ended

The client_state field on every Telnyx action carries the bridge ID and which side (a or b) the call belongs to. When a webhook fires, you decode the base64 state, look up the bridge, and know exactly which call leg you're handling. No session store — the call is self-describing.

The Language Code Bug

The original sample had all TTS and speech recognition hardcoded to language_code="en-US" — even for Spanish. The call "worked" (audio flowed, no errors) but the translation was useless:

Spanish TTS with en-US: the TTS engine reads Spanish words with English phonemes. Unintelligible.
Spanish STT with en-US: the STT engine listens for English sounds. Spanish speech recognition drops.

The fix — a language name → BCP-47 code mapping:

LANG_CODES = {
    "english": "en-US", "spanish": "es-US", "french": "fr-FR",
    "german": "de-DE", "italian": "it-IT", "portuguese": "pt-BR",
    "hindi": "hi-IN", "arabic": "ar-SA", "chinese": "zh-CN",
    "japanese": "ja-JP", "korean": "ko-KR", "russian": "ru-RU",
}

def lang_code(lang):
    return LANG_CODES.get(lang.lower(), "en-US")

Now TTS uses the target language (Spanish text → es-US pronunciation) and STT uses the speaker's language (Spanish speech → es-US recognition).

The Translation Call

Uses Telnyx AI Inference — the OpenAI-compatible /v2/ai/chat/completions endpoint:

def translate(text, from_lang, to_lang):
    resp = requests.post(INFERENCE_URL,
        headers={"Authorization": f"Bearer {TELNYX_API_KEY}"},
        json={
            "model": AI_MODEL,
            "messages": [
                {"role": "system",
                 "content": f"Translate from {from_lang} to {to_lang}. "
                           "Return ONLY the translation, nothing else."},
                {"role": "user", "content": text}
            ],
            "max_tokens": 200, "temperature": 0.1
        })
    return resp.json()["choices"][0]["message"]["content"]

Temperature 0.1 for deterministic translations. 200 max tokens for conversational phrases. The model defaults to moonshotai/Kimi-K2.6 — swap to any model in the Telnyx catalog via the AI_MODEL env var.

No separate translation API account. The same Telnyx API key that makes the phone call does the translation.

Webhook Signature Verification

Telnyx signs every webhook with an Ed25519 key. The app verifies before trusting:

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

Without this, anyone who knows your webhook URL can inject fake call events. The telnyx SDK's webhooks.unwrap() handles the Ed25519 verification.

The Call Loop

Each Telnyx event triggers the next action:

# Caller finished speaking → we have a transcript
elif event_type == "call.gather.ended" and bridge:
    speech = p.get("speech", {}).get("result", "")
    if speech and bridge.get("state") == "active":
        from_lang = bridge[f"lang_{side}"]
        to_lang = bridge["lang_b"] if side == "a" else bridge["lang_a"]
        other_ccid = bridge["ccids"].get("b" if side == "a" else "a")
        translated = translate(speech, from_lang, to_lang)
        if other_ccid:
            client.calls.actions.speak(other_ccid, payload=translated,
                voice="female", language_code=lang_code(to_lang))
        client.calls.actions.gather(ccid, input_type="speech",
            language_code=lang_code(bridge[f"lang_{side}"]))

When either caller hangs up, the other is hung up too — a translation bridge with one caller is pointless:

elif event_type == "call.hangup" and bridge:
    other_side = "b" if side == "a" else "a"
    other_ccid = bridge.get("ccids", {}).get(other_side)
    if other_ccid:
        requests.post(f".../calls/{other_ccid}/actions/hangup", ...)
    bridge["state"] = "ended"

Try It Yourself

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/ai-real-time-translation-bridge-python
cp .env.example .env   # TELNYX_API_KEY, TELNYX_PUBLIC_KEY, BRIDGE_NUMBER, CONNECTION_ID
pip install -r requirements.txt
python app.py

Then:

ngrok http 5000

Point your Call Control Application webhook URL to https://<id>.ngrok.io/webhooks/voice in the Telnyx Portal.

Trigger a bridge:

curl -X POST http://localhost:5000/bridge \
  -H "Content-Type: application/json" \
  -d '{"number_a": "+13125550001", "lang_a": "English",
       "number_b": "+5215550002", "lang_b": "Spanish"}'

Call from two phones. Speak in different languages. Hear real-time translation.

Key links:

Repo: https://github.com/team-telnyx/telnyx-code-examples/tree/main/ai-real-time-translation-bridge-python
Telnyx Portal: https://portal.telnyx.com
Call Control docs: https://developers.telnyx.com/docs/voice/call-control
AI Inference docs: https://developers.telnyx.com/docs/inference

Route Phone Calls by Geography at the Carrier Edge — One Number, Three Regions, Three Compliant Experiences

Harpreet Singh Seehra — Mon, 27 Jul 2026 13:51:05 +0000

Route Phone Calls by Geography at the Carrier Edge

Edge Geo Smart Router — a 256-line Flask webhook that detects caller region from the E.164 number prefix and applies different language, voice, recording, and GDPR consent policy per region. US callers get English AI, LATAM gets Spanish, EU gets a consent prompt before any recording starts.

The Problem With Global Phone Numbers

A single phone number can ring from anywhere. A US toll-free number might receive calls from New York, São Paulo, London, and Berlin in the same hour. Traditional IVRs handle this with a single language menu — "Press 1 for English, 2 for Spanish" — which puts the burden on the caller and guarantees nothing about compliance.

The compliance part is the hard one. If you record a call from an EU caller without explicit consent, you've violated GDPR. If you don't record a US call, you lose your quality assurance trail. If you greet a LATAM caller in English, you've already lost them.

The Edge Geo Smart Router solves this by making the routing decision at the carrier edge — before the first word of greeting is spoken. The caller's number prefix (+1 for US, +55 for Brazil, +44 for UK) is the geo signal. No geoip lookup, no external database, no round-trip to a backend. The webhook receives call.initiated, reads the from field, classifies the region, and applies the right config for the rest of the call.

What It Does

One phone number. Three regions. Three compliant experiences.

Region	Language	Voice	Recording	Consent
US	en-US	AWS.Polly.Joanna-Neural	✅ auto-start	❌ not required
LATAM	es-MX	AWS.Polly.Lupe-Neural	✅ auto-start	❌ not required
EU	en-GB	AWS.Polly.Amy-Neural	⏸️ after consent	✅ DTMF prompt
DEFAULT	en-US	AWS.Polly.Joanna-Neural	✅ auto-start	❌ not required

US and LATAM callers get straight into a conversation with AI in their language. EU callers hear: "This call will be recorded for quality purposes. Press 1 to consent and continue, or press 2 to proceed without recording." If they press 1, recording starts. If they press 2, the call proceeds unrecorded. Either way, GDPR is satisfied.

The Architecture

Everything lives in one 256-line Flask file. No database, no Redis, no message queue. Call state travels in Telnyx's client_state field — a base64-encoded JSON blob that Telnyx passes back to you on every webhook event for the same call.

Caller dials your number
        ↓
Telnyx sends call.initiated webhook → /webhooks/voice
        ↓
app detects region from `from` number prefix
        ↓
app calls /actions/answer with client_state = {region, step: "greeting"}
        ↓
Telnyx sends call.answered → app reads client_state
        ↓
  ┌─────────────┬─────────────┬─────────────┐
  │ US / LATAM  │ EU          │ DEFAULT     │
  │             │             │             │
  │ record_start│ gather DTMF │ record_start│
  │ gather_speak│ (consent)   │ gather_speak│
  │ (greeting)  │             │ (greeting)  │
  └─────┬───────┴──────┬──────┴──────┬──────┘
        ↓              ↓             ↓
  call.gather.ended → app reads step from client_state
        ↓
  "consent" step → did they press 1?
        ↓            → record_start + gather (greeting)
  "conversation" step → call_inference(speech, language)
        ↓                        → gather_speak (AI response)
  loop until hangup
        ↓
  call.hangup → clean up session

Region Detection: The Prefix Match

The routing logic is 12 lines of Python. No geoip, no database, no external API call.

EU_PREFIXES = ["+33", "+34", "+39", "+44", "+49", "+31", "+32", "+43",
               "+45", "+46", "+47", "+48", "+351", "+353", "+358",
               "+352", "+356", "+370", "+371", "+372", "+420", "+421"]
LATAM_PREFIXES = ["+52", "+55", "+54", "+56", "+57", "+51", "+58",
                  "+53", "+506", "+507", "+502", "+503", "+504", "+505"]

def detect_region(phone):
    if not phone:
        return "DEFAULT"
    for prefix in EU_PREFIXES:
        if phone.startswith(prefix):
            return "EU"
    for prefix in LATAM_PREFIXES:
        if phone.startswith(prefix):
            return "LATAM"
    if phone.startswith("+1"):
        return "US"
    return "DEFAULT"

EU is checked first because some European country codes share the +1 prefix with NANP. By checking EU first, we avoid misrouting through the US path.

Per-Region Config

Each region has its own voice, language, greeting, recording policy, and consent requirement.

REGION_CONFIG = {
    "US":     {"language": "en-US", "voice": "AWS.Polly.Joanna-Neural",
               "greeting": "Welcome. How can I help you today?",
               "requires_consent": False, "record": True},
    "LATAM":  {"language": "es-MX", "voice": "AWS.Polly.Lupe-Neural",
               "greeting": "Bienvenido. Como puedo ayudarle hoy?",
               "requires_consent": False, "record": True},
    "EU":     {"language": "en-GB", "voice": "AWS.Polly.Amy-Neural",
               "greeting": "This call will be recorded for quality purposes. "
                           "Press 1 to consent and continue, or press 2 to "
                           "proceed without recording.",
               "requires_consent": True, "record": False},
    "DEFAULT":{"language": "en-US", "voice": "AWS.Polly.Joanna-Neural",
               "greeting": "Welcome. How can I help you?",
               "requires_consent": False, "record": True}
}

The voice ID gotcha

Bare "female" and "male" voice IDs only work with service_level: "basic", which is en-US only. If you try "voice": "female" with "language": "es-MX", the call silently fails. The fix is to use full neural voice IDs — AWS.Polly.Lupe-Neural for es-MX, AWS.Polly.Amy-Neural for en-GB. Each voice supports its respective language.

The GDPR Consent Flow

EU callers get a DTMF (touch-tone) consent prompt before any recording starts. The flow is:

call.answered fires for an EU caller
App sees requires_consent: True in the region config
App calls gather_using_speak with the consent greeting, minimum_digits: 1, maximum_digits: 1
Caller presses 1 (consent) or 2 (no consent)
call.gather.ended fires with step: "consent" in client_state
If digit is "1": app calls record_start, then gather_using_speak with the actual greeting
If digit is "2": app calls gather_using_speak with the greeting, no recording

if step == "consent":
    digits = ep.get("digits", "")
    if digits == "1":
        requests.post(f".../actions/record_start",
                      headers=HEADERS, json={"format": "mp3", "channels": "dual"})
        call_sessions.get(cc_id, {})["consented"] = True
    requests.post(f".../actions/gather_using_speak",
                  json={"payload": "Thank you. How can I help you today?",
                        "voice": config["voice"], "language": config["language"],
                        "input_type": "speech", "timeout_secs": 30,
                        "client_state": encode_state({"region": region, "step": "conversation"})})

The key insight: recording only starts AFTER the caller presses 1. If they press 2, the call proceeds with AI conversation but no recording is captured. GDPR satisfied by design, not by policy.

State Management Without a Database

Every call carries its own state in Telnyx's client_state field. This is a base64-encoded JSON blob that Telnyx passes back to you on every subsequent webhook event for the same call. You set it when you call an action (answer, gather, record), and you read it when the next event fires.

def encode_state(data):
    return base64.b64encode(json.dumps(data).encode()).decode()

def decode_state(b64):
    try: return json.loads(base64.b64decode(b64).decode())
    except: return {}

The state contains just two fields: region (so every event handler knows which region config to apply) and step (so call.gather.ended knows whether it's handling the consent prompt or the actual conversation). No session store, no Redis, no database. The call is self-describing.

Webhook Signature Verification

Telnyx signs every webhook with an Ed25519 key. The signature is over <timestamp>|<raw body>. Verifying it before you trust the payload is non-optional — without it, anyone can POST to your webhook URL and inject fake call events.

def verify_telnyx_signature(raw_body, headers):
    if _TELNYX_VERIFY_KEY is None:
        return False
    signature = headers.get("telnyx-signature-ed25519", "")
    timestamp = headers.get("telnyx-timestamp", "")
    if not signature or not timestamp:
        return False
    try:
        if abs(time.time() - int(timestamp)) > MAX_SKEW_SECONDS:
            return False
        signed = f"{timestamp}|".encode() + raw_body
        _TELNYX_VERIFY_KEY.verify(base64.b64decode(signature), signed)
        return True
    except (InvalidSignature, ValueError, TypeError):
        return False

The 300-second skew window rejects replay attacks. The raw body is verified, not the parsed JSON — because JSON parsing is not canonical (key order, whitespace), and the signature would fail. Get the public key from Portal → Keys & Credentials.

Recording Archival to Telnyx Storage

When call.recording.saved fires, the app downloads the recording and uploads it to Telnyx Storage (S3-compatible). The key includes the region and call session ID for later retrieval.

elif event_type == "call.recording.saved":
    recording_url = ep.get("recording_urls", {}).get("mp3")
    if not (recording_url and is_telnyx_url(recording_url)):
        return jsonify({"status": "ok"})
    if s3 is None:
        app.logger.info("Recording ready but Storage credentials not configured; skipping.")
        return jsonify({"status": "ok"})
    audio = requests.get(recording_url, headers=HEADERS, timeout=30).content
    key = f"recordings/{region}/{call_session_id}.mp3"
    s3.put_object(Bucket=STORAGE_BUCKET, Key=key, Body=audio, ContentType="audio/mpeg")

Two guards here:

SSRF protection: is_telnyx_url() only fetches URLs on telnyx.com or *.telnyx.com hosts. The recording URL comes from the webhook payload, which is attacker-controllable. Without this check, a forged webhook could make your server download from any URL — including internal services.
Storage credentials ≠ API key: Telnyx Storage is S3-compatible but uses its own access/secret key pair, created under Portal → Storage → Credentials. The original version of this sample reused the Telnyx API key as both the S3 access key AND the secret key — which silently failed. The fix: separate STORAGE_ACCESS_KEY and STORAGE_SECRET_KEY environment variables. If they're not set, archival is gracefully skipped (the call still works, just no recording saved).

AI Inference: Multilingual Replies

When the caller speaks, the app sends the transcript to Telnyx AI Inference (OpenAI-compatible /v2/ai/chat/completions endpoint) with a language instruction derived from the region config.

def call_inference(prompt, language="en-US"):
    lang_instruction = "Respond in Spanish." if "es" in language else "Respond in English."
    resp = requests.post(INFERENCE_URL, headers=HEADERS, timeout=15, json={
        "model": AI_MODEL,
        "messages": [{"role": "system",
                      "content": f"You are a helpful business assistant. Be concise. {lang_instruction}"},
                     {"role": "user", "content": prompt}]
    })
    if resp.ok:
        return resp.json()["choices"][0]["message"]["content"]
    return "I apologize, I'm having trouble right now. Please try again."

The AI response is then spoken back to the caller via gather_using_speak with the region's voice and language, and the loop continues until the caller hangs up.

Try It Yourself

git clone https://github.com/team-telnyx/telnyx-code-examples.git
cd telnyx-code-examples/edge-geo-smart-router-python
cp .env.example .env   # add TELNYX_API_KEY, TELNYX_PUBLIC_KEY, optional Storage creds
pip install -r requirements.txt
python app.py

Then:

ngrok http 5000

Configure your Call Control Application webhook URL to https://<id>.ngrok.io/webhooks/voice in the Telnyx Portal.

Call your Telnyx number from:

A US number (+1) → English AI, auto-recording
A Brazil number (+55) → Spanish AI, auto-recording
A UK number (+44) → English (en-GB) AI, consent prompt first

Check the routing config:

curl http://localhost:5000/regions

Check health:

curl http://localhost:5000/health

Key links:

Repo: https://github.com/team-telnyx/telnyx-code-examples/tree/main/edge-geo-smart-router-python
Telnyx Portal: https://portal.telnyx.com
Call Control docs: https://developers.telnyx.com/docs/voice/call-control
AI Inference docs: https://developers.telnyx.com/docs/inference

DEV Community: Harpreet Singh Seehra

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

Architecture

Validate lifecycle transitions

Notify customers without exposing them

Use the same context for inbound calls

Write the RCA and check for recurrence

Run the demo

Where to take it next

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

Architecture

Why combine files and SQL?

Safe shared writes

Run the guided demo

Use cases

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

What We're Building

Architecture

The Sliding Window Pattern

The Agent SDK Pipeline

Stage 1: checkLimit()

Stage 2a: allow()

Stage 2b: reject()

Stage 3: sendAlert()

Quick Start

1. Clone and install

2. Configure environment

3. Deploy

4. Test the rate limit

API Endpoints

Why This Matters

No Redis, no external services

TTL-based cleanup

Per-key actor isolation

Non-blocking pipeline

Use Cases

Frequently Asked Questions

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

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

What happens if the SMS alert fails?

How do I reset a key's counter?

Resources

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

What You'll Build

Architecture

Quick Start

1. Clone and install

2. Configure

3. Deploy

4. Test the export

How the Pipeline Works

Agent SDK queue() — Non-Blocking Stages

Self-Requeuing Chunks

Zero-Credential SMS Notification

Chunked Output in Cloud Storage

API Endpoints

Use Cases

Resources

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

Build a Geo-Distributed Call Logger on Edge Compute

What You'll Build

Why This Architecture Matters

Architecture

Prerequisites

Step 1: Configure the Edge Compute Bindings

Step 2: Region Detection from E.164 Prefixes

Step 3: The GeoLoggerAgent Pipeline

Stage 1: logCall — SQL INSERT + KV INCR

Stage 2: checkThreshold

Stage 3: alert — Zero-Credential SMS

Step 4: The Webhook Handler

Step 5: Run and Test

Install and configure

Run locally

Simulate a call (no real phone needed)

Check region statistics

Trigger an alert

Rolling-Window KV Counters

Why Actors?

API Endpoints

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

3. Update `telnyx.toml`

`POST /invalidate`

`GET /status/:agentId`

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

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

`GET /locations`

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

`POST /upload`

`GET /status/:agentId`

`GET /health/{liveness,readiness}`

The `gather_using_ai` vs `gather` Distinction