team-telnyx/telnyx-python
GitHub: team-telnyx/telnyx-python
Telnyx 官方 Python SDK,提供同步与异步两种客户端,方便开发者在 Python 应用中便捷地调用 Telnyx 通信 API。
Stars: 184 | Forks: 24
# Telnyx Python API 库
[)](https://pypi.org/project/telnyx/)
Telnyx Python 库提供了从任何 Python 3.9+ 应用程序便捷访问 Telnyx REST API 的能力。该库包含所有请求参数和响应字段的类型定义,并提供由 [httpx](https://github.com/encode/httpx) 驱动的同步和异步客户端。
它是使用 [Stainless](https://www.stainless.com/) 生成的。
## MCP Server
使用 Telnyx MCP Server 可以让 AI 助手与此 API 进行交互,允许它们探索 endpoint、发起测试请求,并利用文档帮助将此 SDK 集成到你的应用程序中。
[](https://cursor.com/en-US/install-mcp?name=telnyx-mcp&config=eyJjb21tYW5kIjoibnB4IiwiYXJncyI6WyIteSIsInRlbG55eC1tY3AiXSwiZW52Ijp7IlRFTE5ZWF9BUElfS0VZIjoiTXkgQVBJIEtleSIsIlRFTE5ZWF9QVUJMSUNfS0VZIjoiTXkgUHVibGljIEtleSIsIlRFTE5ZWF9DTElFTlRfSUQiOiJNeSBDbGllbnQgSUQiLCJURUxOWVhfQ0xJRU5UX1NFQ1JFVCI6Ik15IENsaWVudCBTZWNyZXQifX0)
[](https://vscode.stainless.com/mcp/%7B%22name%22%3A%22telnyx-mcp%22%2C%22command%22%3A%22npx%22%2C%22args%22%3A%5B%22-y%22%2C%22telnyx-mcp%22%5D%2C%22env%22%3A%7B%22TELNYX_API_KEY%22%3A%22My%20API%20Key%22%2C%22TELNYX_PUBLIC_KEY%22%3A%22My%20Public%20Key%22%2C%22TELNYX_CLIENT_ID%22%3A%22My%20Client%20ID%22%2C%22TELNYX_CLIENT_SECRET%22%3A%22My%20Client%20Secret%22%7D%7D)
## 文档
可以在 [api.md](api.md) 中找到本库的完整 API。
## 安装
```
# 从 PyPI 安装
pip install telnyx
```
## 使用方法
可以在 [api.md](api.md) 中找到本库的完整 API。
```
import os
from telnyx import Telnyx
client = Telnyx(
api_key=os.environ.get("TELNYX_API_KEY"), # This is the default and can be omitted
)
response = client.calls.dial(
connection_id="conn12345",
from_="+15557654321",
to="+15551234567",
webhook_url="https://your-webhook.url/events",
)
print(response.data)
```
虽然你可以提供 `api_key` 关键字参数,
但我们建议使用 [python-dotenv](https://pypi.org/project/python-dotenv/)
将 `TELNYX_API_KEY="My API Key"` 添加到你的 `.env` 文件中,
这样你的 API Key 就不会在源码版本控制中被泄露。
## 异步使用
只需导入 `AsyncTelnyx` 而不是 `Telnyx`,并在每次 API 调用时使用 `await`:
```
import os
import asyncio
from telnyx import AsyncTelnyx
client = AsyncTelnyx(
api_key=os.environ.get("TELNYX_API_KEY"), # This is the default and can be omitted
)
async def main() -> None:
response = await client.calls.dial(
connection_id="conn12345",
from_="+15557654321",
to="+15551234567",
webhook_url="https://your-webhook.url/events",
)
print(response.data)
asyncio.run(main())
```
除此之外,同步和异步客户端之间的功能是完全相同的。
### 结合 aiohttp 使用
默认情况下,异步客户端使用 `httpx` 进行 HTTP 请求。但是,为了获得更好的并发性能,你也可以使用 `aiohttp` 作为 HTTP 后端。
你可以通过安装 `aiohttp` 来启用它:
```
# 从 PyPI 安装
pip install telnyx[aiohttp]
```
然后,你可以通过在实例化客户端时传入 `http_client=DefaultAioHttpClient()` 来启用它:
```
import os
import asyncio
from telnyx import DefaultAioHttpClient
from telnyx import AsyncTelnyx
async def main() -> None:
async with AsyncTelnyx(
api_key=os.environ.get("TELNYX_API_KEY"), # This is the default and can be omitted
http_client=DefaultAioHttpClient(),
) as client:
response = await client.calls.dial(
connection_id="conn12345",
from_="+15557654321",
to="+15551234567",
webhook_url="https://your-webhook.url/events",
)
print(response.data)
asyncio.run(main())
```
## 使用类型
嵌套的请求参数是 [TypedDicts](https://docs.python.org/3/library/typing.html#typing.TypedDict)。响应是 [Pydantic models](https://docs.pydantic.dev),它们还提供了一些辅助方法,用于:
- 序列化回 JSON,`model.to_json()`
- 转换为字典,`model.to_dict()`
类型化的请求和响应在你的编辑器中提供了自动补全和文档。如果你想在 VS Code 中看到类型错误以便更早地发现 bug,请将 `python.analysis.typeCheckingMode` 设置为 `basic`。
## 分页
Telnyx API 中的列表方法是分页的。
该库为每个列表响应提供了自动分页的迭代器,因此你不必手动请求连续的页面:
```
from telnyx import Telnyx
client = Telnyx()
all_access_ip_addresses = []
# 根据需要自动获取更多页面。
for access_ip_address in client.access_ip_address.list(
page_number=1,
page_size=50,
):
# Do something with access_ip_address here
all_access_ip_addresses.append(access_ip_address)
print(all_access_ip_addresses)
```
或者,以异步方式:
```
import asyncio
from telnyx import AsyncTelnyx
client = AsyncTelnyx()
async def main() -> None:
all_access_ip_addresses = []
# Iterate through items across all pages, issuing requests as needed.
async for access_ip_address in client.access_ip_address.list(
page_number=1,
page_size=50,
):
all_access_ip_addresses.append(access_ip_address)
print(all_access_ip_addresses)
asyncio.run(main())
```
另外,你可以使用 `.has_next_page()`、`.next_page_info()` 或 `.get_next_page()` 方法来更精细地控制页面处理:
```
first_page = await client.access_ip_address.list(
page_number=1,
page_size=50,
)
if first_page.has_next_page():
print(f"will fetch next page using these details: {first_page.next_page_info()}")
next_page = await first_page.get_next_page()
print(f"number of items we just fetched: {len(next_page.data)}")
# 对于非 async 用法,移除 `await`。
```
或者直接处理返回的数据:
```
first_page = await client.access_ip_address.list(
page_number=1,
page_size=50,
)
print(f"page number: {first_page.meta.page_number}") # => "page number: 1"
for access_ip_address in first_page.data:
print(access_ip_address.id)
# 对于非 async 用法,移除 `await`。
```
## 嵌套参数
嵌套参数是使用 `TypedDict` 定义类型的字典,例如:
```
from telnyx import Telnyx
client = Telnyx()
response = client.calls.dial(
connection_id="7267xxxxxxxxxxxxxx",
from_="+18005550101",
to="+18005550100 or sip:username@sip.telnyx.com;secure=srtp",
answering_machine_detection_config={
"after_greeting_silence_millis": 1000,
"between_words_silence_millis": 1000,
"greeting_duration_millis": 1000,
"greeting_silence_duration_millis": 2000,
"greeting_total_analysis_time_millis": 50000,
"initial_silence_millis": 1000,
"maximum_number_of_words": 1000,
"maximum_word_length_millis": 2000,
"silence_threshold": 512,
"total_analysis_time_millis": 5000,
},
)
print(response.answering_machine_detection_config)
```
## 错误处理
当库无法连接到 API 时(例如,由于网络连接问题或超时),会抛出 `telnyx.APIConnectionError` 的子类。
当 API 返回非成功状态码(即 4xx 或 5xx 响应)时,会抛出 `telnyx.APIStatusError` 的子类,其中包含 `status_code` 和 `response` 属性。
所有错误都继承自 `telnyx.APIError`。
```
import telnyx
from telnyx import Telnyx
client = Telnyx()
try:
client.number_orders.create(
phone_numbers=[{"phone_number": "+15558675309"}],
)
except telnyx.APIConnectionError as e:
print("The server could not be reached")
print(e.__cause__) # an underlying Exception, likely raised within httpx.
except telnyx.RateLimitError as e:
print("A 429 status code was received; we should back off a bit.")
except telnyx.APIStatusError as e:
print("Another non-200-range status code was received")
print(e.status_code)
print(e.response)
```
错误代码如下:
| 状态码 | 错误类型 |
| ----------- | -------------------------- |
| 400 | `BadRequestError` |
| 401 | `AuthenticationError` |
| 403 | `PermissionDeniedError` |
| 404 | `NotFoundError` |
| 422 | `UnprocessableEntityError` |
| 429 | `RateLimitError` |
| >=500 | `InternalServerError` |
| N/A | `APIConnectionError` |
### 重试
默认情况下,某些错误会自动重试 2 次,并使用短暂的指数退避策略。
连接错误(例如,由于网络连接问题)、408 Request Timeout、409 Conflict、429 Rate Limit 和 >=500 Internal 错误默认都会进行重试。
你可以使用 `max_retries` 选项来配置或禁用重试设置:
```
from telnyx import Telnyx
# 为所有请求配置默认值:
client = Telnyx(
# default is 2
max_retries=0,
)
# 或者,按请求进行配置:
client.with_options(max_retries=5).number_orders.create(
phone_numbers=[{"phone_number": "+15558675309"}],
)
```
### 超时
默认情况下,请求会在 1 分钟后超时。你可以使用 `timeout` 选项进行配置,
该选项接受一个浮点数或一个 [`httpx.Timeout`](https://www.python-httpx.org/advanced/timeouts/#fine-tuning-the-configuration) 对象:
```
from telnyx import Telnyx
# 为所有请求配置默认值:
client = Telnyx(
# 20 seconds (default is 1 minute)
timeout=20.0,
)
# 更精细的控制:
client = Telnyx(
timeout=httpx.Timeout(60.0, read=5.0, write=10.0, connect=2.0),
)
# 按请求覆盖:
client.with_options(timeout=5.0).number_orders.create(
phone_numbers=[{"phone_number": "+15558675309"}],
)
```
超时时,会抛出 `APITimeoutError` 异常。
请注意,超时的请求默认会[重试两次](#retries)。
## 进阶
### 日志
我们使用标准库的 [`logging`](https://docs.python.org/3/library/logging.html) 模块。
你可以通过将环境变量 `TELNYX_LOG` 设置为 `info` 来启用日志记录。
```
$ export TELNYX_LOG=info
```
或者设置为 `debug` 以获取更详细的日志。
### 如何判断 `None` 表示 `null` 还是缺失
在 API 响应中,某个字段可能显式为 `null`,也可能完全缺失;在这两种情况下,它在本库中的值都是 `None`。你可以使用 `.model_fields_set` 来区分这两种情况:
```
if response.my_field is None:
if 'my_field' not in response.model_fields_set:
print('Got json like {}, without a "my_field" key present at all.')
else:
print('Got json like {"my_field": null}.')
```
### 访问原始响应数据(例如 headers)
可以通过在任何 HTTP 方法调用前加上 `.with_raw_response.` 前缀来访问“原始” Response 对象,例如:
```
from telnyx import Telnyx
client = Telnyx()
response = client.number_orders.with_raw_response.create(
phone_numbers=[{
"phone_number": "+15558675309"
}],
)
print(response.headers.get('X-My-Header'))
number_order = response.parse() # get the object that `number_orders.create()` would have returned
print(number_order.data)
```
这些方法返回一个 [`APIResponse`](https://github.com/team-telnyx/telnyx-python/tree/master/src/telnyx/_response.py) 对象。
异步客户端返回一个具有相同结构的 [`AsyncAPIResponse`](https://github.com/team-telnyx/telnyx-python/tree/master/src/telnyx/_response.py),唯一的区别是读取响应内容的方法是可 `await` 的。
#### `.with_streaming_response`
```
with client.number_orders.with_streaming_response.create(
phone_numbers=[{"phone_number": "+15558675309"}],
) as response:
print(response.headers.get("X-My-Header"))
for line in response.iter_lines():
print(line)
```
必须使用上下文管理器,以确保响应能被可靠地关闭。
### 发起自定义/未记录的请求
本库为便捷访问已记录的 API 提供了类型定义。
如果你需要访问未记录的 endpoint、参数或响应属性,仍然可以使用该库。
#### 未记录的 endpoint
要向未记录的 endpoint 发起请求,你可以使用 `client.get`、`client.post` 以及其他 HTTP 动词来发起请求。发起此请求时,客户端上的选项(如重试)将得到尊重。
```
import httpx
response = client.post(
"/foo",
cast_to=httpx.Response,
body={"my_param": True},
)
print(response.headers.get("x-foo"))
```
#### 未记录的请求参数
如果你想显式发送额外的参数,可以使用 `extra_query`、`extra_body` 和 `extra_headers` 请求选项。
#### 未记录的响应属性
要访问未记录的响应属性,你可以像 `response.unknown_prop` 这样访问额外字段。你也可以使用
[`response.model_extra`](https://docs.pydantic.dev/latest/api/base_model/#pydantic.BaseModel.model_extra) 将 Pydantic model 上的所有额外字段作为字典获取。
### 配置 HTTP client
你可以直接覆盖 [httpx client](https://www.python-httpx.org/api/#client) 以根据你的使用场景进行自定义,包括:
```
import httpx
from telnyx import Telnyx, DefaultHttpxClient
client = Telnyx(
# Or use the `TELNYX_BASE_URL` env var
base_url="http://my.test.server.example.com:8083",
http_client=DefaultHttpxClient(
proxy="http://my.test.proxy.example.com",
transport=httpx.HTTPTransport(local_address="0.0.0.0"),
),
)
```
你也可以使用 `with_options()` 在每个请求的基础上自定义客户端:
```
client.with_options(http_client=DefaultHttpxClient(...))
```
### 管理 HTTP 资源
默认情况下,每当客户端被[垃圾回收](https://docs.python.org/3/reference/datamodel.html#object.__del__)时,库就会关闭底层的 HTTP 连接。如果需要,你可以使用 `.close()` 方法手动关闭客户端,或者在退出时使用上下文管理器关闭。
```
from telnyx import Telnyx
with Telnyx() as client:
# make requests here
...
# HTTP client 现已关闭
```
## 版本控制
本包通常遵循 [SemVer](https://semver.org/spec/v2.0.0.html) 规范,不过某些向后不兼容的更改可能会作为次要版本发布:
1. 仅影响静态类型而不破坏运行时行为的更改。
2. 对库内部结构的更改,这些内部结构技术上是公开的,但并不打算供外部使用或未在文档中记录。_(如果你依赖于这些内部结构,请提交一个 GitHub issue 告诉我们。)_
3. 我们认为在实践中不会影响绝大多数用户的更改。
我们非常重视向后兼容性,并努力确保你能获得顺畅的升级体验。
我们期待你的反馈;如果有任何问题、Bug 或建议,请提交一个 [issue](https://www.github.com/team-telnyx/telnyx-python/issues)。
### 确定已安装的版本
如果你已经升级到最新版本,但没有看到你期望的任何新功能,那么你的 Python 环境可能仍在使用旧版本。
你可以通过以下方式确定正在使用的版本:
```
import telnyx
print(telnyx.__version__)
```
## 环境要求
Python 3.9 或更高版本。
标签:运行时操纵, 逆向工具