telnyx-voice-streaming-python

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Telnyx Voice Streaming - Python

Telnyx 语音流传输 - Python

Installation

安装

bash

pip install telnyx

bash

pip install telnyx

Setup

环境配置

python

import os
from telnyx import Telnyx

client = Telnyx(
    api_key=os.environ.get("TELNYX_API_KEY"),  # This is the default and can be omitted
)

All examples below assume

client

is already initialized as shown above.

python

import os
from telnyx import Telnyx

client = Telnyx(
    api_key=os.environ.get("TELNYX_API_KEY"),  # 这是默认配置，可省略
)

以下所有示例均默认

client

已按上述方式完成初始化。

Error Handling

错误处理

All API calls can fail with network errors, rate limits (429), validation errors (422), or authentication errors (401). Always handle errors in production code:

python

import telnyx

try:
    result = client.messages.send(to="+13125550001", from_="+13125550002", text="Hello")
except telnyx.APIConnectionError:
    print("Network error — check connectivity and retry")
except telnyx.RateLimitError:
    # 429: rate limited — wait and retry with exponential backoff
    import time
    time.sleep(1)  # Check Retry-After header for actual delay
except telnyx.APIStatusError as e:
    print(f"API error {e.status_code}: {e.message}")
    if e.status_code == 422:
        print("Validation error — check required fields and formats")

Common error codes:

invalid API key,

insufficient permissions,

resource not found,

validation error (check field formats),

rate limited (retry with exponential backoff).

所有API调用都可能失败，原因包括网络错误、速率限制(429)、校验错误(422)或鉴权错误(401)。生产环境代码中请务必做好错误处理：

python

import telnyx

try:
    result = client.messages.send(to="+13125550001", from_="+13125550002", text="Hello")
except telnyx.APIConnectionError:
    print("网络错误 — 检查网络连接并重试")
except telnyx.RateLimitError:
    # 429: 触发速率限制 — 等待后采用指数退避策略重试
    import time
    time.sleep(1)  # 可查看Retry-After响应头获取准确的等待时长
except telnyx.APIStatusError as e:
    print(f"API错误 {e.status_code}: {e.message}")
    if e.status_code == 422:
        print("校验错误 — 检查必填字段和格式是否正确")

常见错误码：

API密钥无效，

权限不足，

资源未找到，

校验错误（检查字段格式），

触发速率限制（采用指数退避策略重试）。

Forking start

启动媒体分发

Call forking allows you to stream the media from a call to a specific target in realtime. This stream can be used to enable realtime audio analysis to support a variety of use cases, including fraud detection, or the creation of AI-generated audio responses. Requests must specify either the

target

attribute or the

rx

and

tx

attributes.

POST /calls/{call_control_id}/actions/fork_start

Optional:

client_state

(string),

command_id

(string),

rx

(string),

stream_type

(enum: decrypted),

tx

(string)

python

response = client.calls.actions.start_forking(
    call_control_id="550e8400-e29b-41d4-a716-446655440000",
)
print(response.data)

Returns:

result

(string)

通话媒体分发功能允许你将通话中的媒体流实时发送到指定目标，该流可用于实现实时音频分析，支持欺诈检测、AI生成音频响应等多种场景。请求必须指定

target

属性，或

rx

和

tx

属性。

POST /calls/{call_control_id}/actions/fork_start

可选参数：

client_state

(字符串),

command_id

(字符串),

rx

(字符串),

stream_type

(枚举值: decrypted),

tx

(字符串)

python

response = client.calls.actions.start_forking(
    call_control_id="550e8400-e29b-41d4-a716-446655440000",
)
print(response.data)

返回值：

result

(字符串)

Forking stop

停止媒体分发

Stop forking a call. Expected Webhooks:

```
call.fork.stopped
```

POST /calls/{call_control_id}/actions/fork_stop

Optional:

client_state

(string),

command_id

(string),

stream_type

(enum: raw, decrypted)

python

response = client.calls.actions.stop_forking(
    call_control_id="550e8400-e29b-41d4-a716-446655440000",
)
print(response.data)

Returns:

result

(string)

停止通话媒体分发。预期触发的Webhooks：

```
call.fork.stopped
```

POST /calls/{call_control_id}/actions/fork_stop

可选参数：

client_state

(字符串),

command_id

(字符串),

stream_type

(枚举值: raw, decrypted)

python

response = client.calls.actions.stop_forking(
    call_control_id="550e8400-e29b-41d4-a716-446655440000",
)
print(response.data)

返回值：

result

(字符串)

Streaming start

启动流传输

Start streaming the media from a call to a specific WebSocket address or Dialogflow connection in near-realtime. Audio will be delivered as base64-encoded RTP payload (raw audio), wrapped in JSON payloads. Please find more details about media streaming messages specification under the link.

POST /calls/{call_control_id}/actions/streaming_start

Optional:

client_state

(string),

command_id

(string),

custom_parameters

(array[object]),

dialogflow_config

(object),

enable_dialogflow

(boolean),

stream_auth_token

(string),

stream_bidirectional_codec

(enum: PCMU, PCMA, G722, OPUS, AMR-WB, L16),

stream_bidirectional_mode

(enum: mp3, rtp),

stream_bidirectional_sampling_rate

(enum: 8000, 16000, 22050, 24000, 48000),

stream_bidirectional_target_legs

(enum: both, self, opposite),

stream_codec

(enum: PCMU, PCMA, G722, OPUS, AMR-WB, L16, default),

stream_track

(enum: inbound_track, outbound_track, both_tracks),

stream_url

(string)

python

response = client.calls.actions.start_streaming(
    call_control_id="550e8400-e29b-41d4-a716-446655440000",
    stream_url="wss://example.com/audio-stream",
)
print(response.data)

Returns:

result

(string)

近乎实时地将通话媒体流发送到指定WebSocket地址或Dialogflow连接。音频将以base64编码的RTP负载（原始音频）形式封装在JSON payload中传输。你可以通过链接查看媒体流消息规范的更多详情。

POST /calls/{call_control_id}/actions/streaming_start

可选参数：

client_state

(字符串),

command_id

(字符串),

custom_parameters

(对象数组),

dialogflow_config

(对象),

enable_dialogflow

(布尔值),

stream_auth_token

(字符串),

stream_bidirectional_codec

(枚举值: PCMU, PCMA, G722, OPUS, AMR-WB, L16),

stream_bidirectional_mode

(枚举值: mp3, rtp),

stream_bidirectional_sampling_rate

(枚举值: 8000, 16000, 22050, 24000, 48000),

stream_bidirectional_target_legs

(枚举值: both, self, opposite),

stream_codec

(枚举值: PCMU, PCMA, G722, OPUS, AMR-WB, L16, default),

stream_track

(枚举值: inbound_track, outbound_track, both_tracks),

stream_url

(字符串)

python

response = client.calls.actions.start_streaming(
    call_control_id="550e8400-e29b-41d4-a716-446655440000",
    stream_url="wss://example.com/audio-stream",
)
print(response.data)

返回值：

result

(字符串)

Streaming stop

停止流传输

Stop streaming a call to a WebSocket. Expected Webhooks:

```
streaming.stopped
```

POST /calls/{call_control_id}/actions/streaming_stop

Optional:

client_state

(string),

command_id

(string),

stream_id

(uuid)

python

response = client.calls.actions.stop_streaming(
    call_control_id="550e8400-e29b-41d4-a716-446655440000",
)
print(response.data)

Returns:

result

(string)

停止向WebSocket发送通话流。预期触发的Webhooks：

```
streaming.stopped
```

POST /calls/{call_control_id}/actions/streaming_stop

可选参数：

client_state

(字符串),

command_id

(字符串),

stream_id

(uuid)

python

response = client.calls.actions.stop_streaming(
    call_control_id="550e8400-e29b-41d4-a716-446655440000",
)
print(response.data)

返回值：

result

(字符串)

Transcription start

启动实时转写

Start real-time transcription. Transcription will stop on call hang-up, or can be initiated via the Transcription stop command. Expected Webhooks:

```
call.transcription
```

POST /calls/{call_control_id}/actions/transcription_start

Optional:

client_state

(string),

command_id

(string),

transcription_engine

(enum: Google, Telnyx, Deepgram, Azure, A, B),

transcription_engine_config

(object),

transcription_tracks

(string)

python

response = client.calls.actions.start_transcription(
    call_control_id="550e8400-e29b-41d4-a716-446655440000",
)
print(response.data)

Returns:

result

(string)

启动实时语音转写，转写将在通话挂断时自动停止，也可通过停止转写命令主动终止。预期触发的Webhooks：

```
call.transcription
```

POST /calls/{call_control_id}/actions/transcription_start

可选参数：

client_state

(字符串),

command_id

(字符串),

transcription_engine

(枚举值: Google, Telnyx, Deepgram, Azure, A, B),

transcription_engine_config

(对象),

transcription_tracks

(字符串)

python

response = client.calls.actions.start_transcription(
    call_control_id="550e8400-e29b-41d4-a716-446655440000",
)
print(response.data)

返回值：

result

(字符串)

Transcription stop

停止实时转写

Stop real-time transcription.

POST /calls/{call_control_id}/actions/transcription_stop

Optional:

client_state

(string),

command_id

(string)

python

response = client.calls.actions.stop_transcription(
    call_control_id="550e8400-e29b-41d4-a716-446655440000",
)
print(response.data)

Returns:

result

(string)

停止实时语音转写。

POST /calls/{call_control_id}/actions/transcription_stop

可选参数：

client_state

(字符串),

command_id

(字符串)

python

response = client.calls.actions.stop_transcription(
    call_control_id="550e8400-e29b-41d4-a716-446655440000",
)
print(response.data)

返回值：

result

(字符串)

Webhooks

Webhook Verification

Webhook验证

Telnyx signs webhooks with Ed25519. Each request includes

telnyx-signature-ed25519

and

telnyx-timestamp

headers. Always verify signatures in production:

python

undefined

Telnyx使用Ed25519对Webhooks进行签名，每个请求都包含

telnyx-signature-ed25519

和

telnyx-timestamp

头。生产环境请务必验证签名：

python

undefined

In your webhook handler (e.g., Flask — use raw body, not parsed JSON):

在你的Webhook处理逻辑中（例如Flask — 请使用原始请求体，不要使用解析后的JSON）：

@app.route("/webhooks", methods=["POST"]) def handle_webhook(): payload = request.get_data(as_text=True) # raw body as string headers = dict(request.headers) try: event = client.webhooks.unwrap(payload, headers=headers) except Exception as e: print(f"Webhook verification failed: {e}") return "Invalid signature", 400 # Signature valid — event is the parsed webhook payload print(f"Received event: {event.data.event_type}") return "OK", 200


The following webhook events are sent to your configured webhook URL.
All webhooks include `telnyx-timestamp` and `telnyx-signature-ed25519` headers for Ed25519 signature verification. Use `client.webhooks.unwrap()` to verify.

| Event | Description |
|-------|-------------|
| `callForkStarted` | Call Fork Started |
| `callForkStopped` | Call Fork Stopped |
| `callStreamingFailed` | Call Streaming Failed |
| `callStreamingStarted` | Call Streaming Started |
| `callStreamingStopped` | Call Streaming Stopped |
| `transcription` | Transcription |

@app.route("/webhooks", methods=["POST"]) def handle_webhook(): payload = request.get_data(as_text=True) # 字符串格式的原始请求体 headers = dict(request.headers) try: event = client.webhooks.unwrap(payload, headers=headers) except Exception as e: print(f"Webhook验证失败: {e}") return "Invalid signature", 400 # 签名验证通过 — event为解析后的Webhook payload print(f"收到事件: {event.data.event_type}") return "OK", 200


以下Webhook事件会发送到你配置的Webhook地址，所有Webhooks都包含`telnyx-timestamp`和`telnyx-signature-ed25519`头用于Ed25519签名验证，可使用`client.webhooks.unwrap()`完成验证。

| 事件 | 描述 |
|-------|-------------|
| `callForkStarted` | 通话媒体分发已启动 |
| `callForkStopped` | 通话媒体分发已停止 |
| `callStreamingFailed` | 通话流传输失败 |
| `callStreamingStarted` | 通话流传输已启动 |
| `callStreamingStopped` | 通话流传输已停止 |
| `transcription` | 转写结果 |

Webhook payload fields

Webhook payload字段

callForkStarted

Field	Type	Description
`data.record_type`	enum: event	Identifies the type of the resource.
`data.event_type`	enum: call.fork.started	The type of event being delivered.
`data.id`	uuid	Identifies the type of resource.
`data.occurred_at`	date-time	ISO 8601 datetime of when the event occurred.
`data.payload.connection_id`	string	Call Control App ID (formerly Telnyx connection ID) used in the call.
`data.payload.call_control_id`	string	Unique ID for controlling the call.
`data.payload.call_leg_id`	string	ID that is unique to the call and can be used to correlate webhook events.
`data.payload.call_session_id`	string	ID that is unique to the call session and can be used to correlate webhook events.
`data.payload.client_state`	string	State received from a command.
`data.payload.stream_type`	enum: decrypted	Type of media streamed.

callForkStopped

Field	Type	Description
`data.record_type`	enum: event	Identifies the type of the resource.
`data.event_type`	enum: call.fork.stopped	The type of event being delivered.
`data.id`	uuid	Identifies the type of resource.
`data.occurred_at`	date-time	ISO 8601 datetime of when the event occurred.
`data.payload.connection_id`	string	Call Control App ID (formerly Telnyx connection ID) used in the call.
`data.payload.call_control_id`	string	Unique ID for controlling the call.
`data.payload.call_leg_id`	string	ID that is unique to the call and can be used to correlate webhook events.
`data.payload.call_session_id`	string	ID that is unique to the call session and can be used to correlate webhook events.
`data.payload.client_state`	string	State received from a command.
`data.payload.stream_type`	enum: decrypted	Type of media streamed.

callStreamingFailed

Field	Type	Description
`data.record_type`	enum: event	Identifies the resource.
`data.event_type`	enum: streaming.failed	The type of event being delivered.
`data.id`	uuid	Identifies the type of resource.
`data.occurred_at`	date-time	ISO 8601 datetime of when the event occurred.
`data.payload.call_control_id`	string	Call ID used to issue commands via Call Control API.
`data.payload.connection_id`	string	Call Control App ID (formerly Telnyx connection ID) used in the call.
`data.payload.call_leg_id`	string	ID that is unique to the call and can be used to correlate webhook events.
`data.payload.call_session_id`	string	ID that is unique to the call session and can be used to correlate webhook events.
`data.payload.client_state`	string	State received from a command.
`data.payload.failure_reason`	string	A short description explaning why the media streaming failed.
`data.payload.stream_id`	uuid	Identifies the streaming.
`data.payload.stream_type`	enum: websocket, dialogflow	The type of stream connection the stream is performing.

callStreamingStarted

Field	Type	Description
`data.record_type`	enum: event	Identifies the type of the resource.
`data.event_type`	enum: streaming.started	The type of event being delivered.
`data.id`	uuid	Identifies the type of resource.
`data.occurred_at`	date-time	ISO 8601 datetime of when the event occurred.
`data.payload.call_control_id`	string	Call ID used to issue commands via Call Control API.
`data.payload.connection_id`	string	Call Control App ID (formerly Telnyx connection ID) used in the call.
`data.payload.call_leg_id`	string	ID that is unique to the call and can be used to correlate webhook events.
`data.payload.call_session_id`	string	ID that is unique to the call session and can be used to correlate webhook events.
`data.payload.client_state`	string	State received from a command.
`data.payload.stream_url`	string	Destination WebSocket address where the stream is going to be delivered.

callStreamingStopped

Field	Type	Description
`data.record_type`	enum: event	Identifies the type of the resource.
`data.event_type`	enum: streaming.stopped	The type of event being delivered.
`data.id`	uuid	Identifies the type of resource.
`data.occurred_at`	date-time	ISO 8601 datetime of when the event occurred.
`data.payload.call_control_id`	string	Call ID used to issue commands via Call Control API.
`data.payload.connection_id`	string	Call Control App ID (formerly Telnyx connection ID) used in the call.
`data.payload.call_leg_id`	string	ID that is unique to the call and can be used to correlate webhook events.
`data.payload.call_session_id`	string	ID that is unique to the call session and can be used to correlate webhook events.
`data.payload.client_state`	string	State received from a command.
`data.payload.stream_url`	string	Destination WebSocket address where the stream is going to be delivered.

transcription

Field	Type	Description
`data.record_type`	enum: event	Identifies the type of the resource.
`data.event_type`	enum: call.transcription	The type of event being delivered.
`data.id`	uuid	Identifies the type of resource.
`data.occurred_at`	date-time	ISO 8601 datetime of when the event occurred.
`data.payload.call_control_id`	string	Unique identifier and token for controlling the call.
`data.payload.call_leg_id`	string	ID that is unique to the call and can be used to correlate webhook events.
`data.payload.call_session_id`	string	ID that is unique to the call session and can be used to correlate webhook events.
`data.payload.client_state`	string	Use this field to add state to every subsequent webhook.
`data.payload.connection_id`	string	Call Control App ID (formerly Telnyx connection ID) used in the call.

callForkStarted

字段	类型	描述
`data.record_type`	枚举值: event	标识资源类型
`data.event_type`	枚举值: call.fork.started	触发的事件类型
`data.id`	uuid	标识资源类型
`data.occurred_at`	日期时间	事件发生的ISO 8601格式时间
`data.payload.connection_id`	字符串	通话使用的Call Control App ID（原Telnyx connection ID）
`data.payload.call_control_id`	字符串	用于控制通话的唯一ID
`data.payload.call_leg_id`	字符串	通话唯一ID，可用于关联Webhook事件
`data.payload.call_session_id`	字符串	通话会话唯一ID，可用于关联Webhook事件
`data.payload.client_state`	字符串	从命令中获取的状态
`data.payload.stream_type`	枚举值: decrypted	传输的媒体流类型

callForkStopped

字段	类型	描述
`data.record_type`	枚举值: event	标识资源类型
`data.event_type`	枚举值: call.fork.stopped	触发的事件类型
`data.id`	uuid	标识资源类型
`data.occurred_at`	日期时间	事件发生的ISO 8601格式时间
`data.payload.connection_id`	字符串	通话使用的Call Control App ID（原Telnyx connection ID）
`data.payload.call_control_id`	字符串	用于控制通话的唯一ID
`data.payload.call_leg_id`	字符串	通话唯一ID，可用于关联Webhook事件
`data.payload.call_session_id`	字符串	通话会话唯一ID，可用于关联Webhook事件
`data.payload.client_state`	字符串	从命令中获取的状态
`data.payload.stream_type`	枚举值: decrypted	传输的媒体流类型

callStreamingFailed

字段	类型	描述
`data.record_type`	枚举值: event	标识资源类型
`data.event_type`	枚举值: streaming.failed	触发的事件类型
`data.id`	uuid	标识资源类型
`data.occurred_at`	日期时间	事件发生的ISO 8601格式时间
`data.payload.call_control_id`	字符串	用于通过Call Control API发送命令的通话ID
`data.payload.connection_id`	字符串	通话使用的Call Control App ID（原Telnyx connection ID）
`data.payload.call_leg_id`	字符串	通话唯一ID，可用于关联Webhook事件
`data.payload.call_session_id`	字符串	通话会话唯一ID，可用于关联Webhook事件
`data.payload.client_state`	字符串	从命令中获取的状态
`data.payload.failure_reason`	字符串	媒体流传输失败的简短说明
`data.payload.stream_id`	uuid	流传输的唯一标识
`data.payload.stream_type`	枚举值: websocket, dialogflow	流连接的类型

callStreamingStarted

字段	类型	描述
`data.record_type`	枚举值: event	标识资源类型
`data.event_type`	枚举值: streaming.started	触发的事件类型
`data.id`	uuid	标识资源类型
`data.occurred_at`	日期时间	事件发生的ISO 8601格式时间
`data.payload.call_control_id`	字符串	用于通过Call Control API发送命令的通话ID
`data.payload.connection_id`	字符串	通话使用的Call Control App ID（原Telnyx connection ID）
`data.payload.call_leg_id`	字符串	通话唯一ID，可用于关联Webhook事件
`data.payload.call_session_id`	字符串	通话会话唯一ID，可用于关联Webhook事件
`data.payload.client_state`	字符串	从命令中获取的状态
`data.payload.stream_url`	字符串	流传输的目标WebSocket地址

callStreamingStopped

字段	类型	描述
`data.record_type`	枚举值: event	标识资源类型
`data.event_type`	枚举值: streaming.stopped	触发的事件类型
`data.id`	uuid	标识资源类型
`data.occurred_at`	日期时间	事件发生的ISO 8601格式时间
`data.payload.call_control_id`	字符串	用于通过Call Control API发送命令的通话ID
`data.payload.connection_id`	字符串	通话使用的Call Control App ID（原Telnyx connection ID）
`data.payload.call_leg_id`	字符串	通话唯一ID，可用于关联Webhook事件
`data.payload.call_session_id`	字符串	通话会话唯一ID，可用于关联Webhook事件
`data.payload.client_state`	字符串	从命令中获取的状态
`data.payload.stream_url`	字符串	流传输的目标WebSocket地址

transcription

字段	类型	描述
`data.record_type`	枚举值: event	标识资源类型
`data.event_type`	枚举值: call.transcription	触发的事件类型
`data.id`	uuid	标识资源类型
`data.occurred_at`	日期时间	事件发生的ISO 8601格式时间
`data.payload.call_control_id`	字符串	用于控制通话的唯一标识和令牌
`data.payload.call_leg_id`	字符串	通话唯一ID，可用于关联Webhook事件
`data.payload.call_session_id`	字符串	通话会话唯一ID，可用于关联Webhook事件
`data.payload.client_state`	字符串	可通过该字段为后续所有Webhook添加状态信息
`data.payload.connection_id`	字符串	通话使用的Call Control App ID（原Telnyx connection ID）