Python SDK 参考

Speech Engine Python SDK 的类、方法和事件。

本页介绍 Speech Engine Python SDK(elevenlabs)的公开 API。

获取 Speech Engine 资源

通过引擎 ID 获取 SpeechEngineResource。返回的对象可用于启动服务器、验证请求或创建单个会话。

from elevenlabs import AsyncElevenLabs
elevenlabs = AsyncElevenLabs()
engine = await elevenlabs.speech_engine.get("seng_8k3m9xr4hjnfg983brhmhkd98n6")

SpeechEngineResource

属性

属性类型描述
engine_idstr语音引擎的 ID。

serve

启动独立的 WebSocket 服务器,直到停止前会一直阻塞。

await engine.serve(
port=3001,
path="/ws",
debug=True,
on_transcript=handle_transcript,
)
参数类型默认值描述
portint3001监听端口。
pathstrNone将连接限制为此路径。None 接受所有连接。
debugboolFalse启用输出到 stdout 的调试日志。
disable_authboolFalse跳过传入连接的 JWT 验证。请参阅禁用身份验证。
on_initcallable会话初始化时调用。
on_transcriptcallable收到用户转录文本时调用。
on_closecallable正常断开连接时调用。
on_disconnectcallableWebSocket 意外断开时调用。
on_errorcallable发生协议或 WebSocket 错误时调用。

禁用身份验证

默认情况下,serve() 会验证每个传入连接的 X-Elevenlabs-Speech-Engine-Authorization 标头。如果服务器位于已将传入流量限制为 ElevenLabs 的基础设施层之后(通常是限定为 ElevenLabs 出站 IP 范围的 IP 允许列表),可通过传入 disable_auth=True 跳过 JWT 验证:

# No api_key required when disable_auth is True
await engine.serve(port=3001, disable_auth=True, on_transcript=on_transcript)
# Or directly on SpeechEngineServer
from elevenlabs.speech_engine import SpeechEngineServer
server = SpeechEngineServer(port=3001, disable_auth=True, on_transcript=on_transcript)
await server.serve()

禁用身份验证后,服务器会接受任何能够访问它的客户端,并在启动时发出 UserWarning。

仅当服务器前方设有 IP 允许列表、自定义标头值或等效的网络级访问限制时,才使用 disable_auth=True。 否则,任何互联网用户都能打开会话,并消耗计算资源和下游 LLM 配额。

verify_request

验证传入请求是否来自 ElevenLabs Speech Engine API。检查 X-Elevenlabs-Speech-Engine-Authorization 标头中是否包含使用 API 密钥 SHA-256 哈希签名的有效 JWT。

仅当自行管理 WebSocket 升级时才需要使用。使用 serve() 时,系统会自动进行验证(除非设置了 disable_auth=True)。

is_valid = engine.verify_request(headers)
参数类型描述
headersdict请求标头字典。

返回: bool — 请求有效时为 True。

create_session

将已接受的 WebSocket 包装为 SpeechEngineSession。用于自定义服务器集成(例如 FastAPI、Starlette 或手动处理 WebSocket)。

session = engine.create_session(websocket, debug=True)
session.on("user_transcript", handle_transcript)
await session.run()
参数类型默认值描述
wsWebSocket已接受的 WebSocket 连接。
debugboolFalse启用调试日志。

返回: SpeechEngineSession

SpeechEngineSession

包装单个 WebSocket 连接。每个连接代表一段对话。会话会为转录文本和生命周期变更触发事件,并提供将 LLM 响应发送回去的方法。

收到新的转录文本时,之前的转录处理程序会自动取消,从而中断正在进行的 LLM 调用。

属性

属性类型描述
conversation_idOptional[str]API 分配的对话 ID。在 init 后可用。
is_openbool会话是否仍处于打开状态。

on

为事件注册处理程序。返回会话以便链式调用。

session.on("user_transcript", handler)

off

移除之前注册的处理程序。

session.off("user_transcript", handler)

once

注册一个仅触发一次后便自行移除的处理程序。

session.once("init", handler)

send_response

将 LLM 响应发送回 Speech Engine API 进行文本转语音合成。必须在 on_transcript 处理程序内调用。在处理程序外调用会发出警告并直接返回,不会发送响应。

# String response
await session.send_response("Hello, how can I help?")
# Streamed response (OpenAI, Anthropic, or Gemini)
stream = await openai_client.responses.create(model="gpt-4o", input=messages, stream=True)
await session.send_response(stream)
参数类型描述
responsestr | async iterable完整字符串,或文本片段 / LLM 流事件的异步可迭代对象。

SDK 会自动检测并从以下 LLM 流格式中提取文本:

提供商事件格式
OpenAI Responses API{ type: "response.output_text.delta", delta: "text" }
OpenAI Chat Completions{ choices: [{ delta: { content: "text" } }] }
Anthropic Messages API{ type: "content_block_delta", delta: { type: "text_delta", text: "text" } }
Google Gemini API{ candidates: [{ content: { parts: [{ text: "text" }] } }] }

run

运行接收循环,直至 WebSocket 关闭。通过 create_session() 手动创建会话后,这是主要入口点。

session = engine.create_session(websocket)
session.on("user_transcript", handle_transcript)
await session.run()

close

关闭会话及底层 WebSocket 连接。

session.close()

回调

传递给 serve() 的关键字参数。所有回调都是可选的。处理程序可以是同步或异步(协程)函数。

回调签名描述
on_init(conversation_id: str, session) -> None使用对话 ID 初始化会话。
on_transcript(transcript: list, session) -> None用户语音已转录。
on_close(session) -> None与 ElevenLabs 正常断开连接。
on_disconnect(session) -> NoneWebSocket 意外断开。
on_error(error: Exception, session) -> None协议或 WebSocket 错误。

事件

直接使用 session.on() 而非回调时,以下为事件名称及其处理程序签名。

事件处理程序签名
user_transcript(transcript: list[ConversationMessage])
init(conversation_id: str)
close()
disconnected()
error(error: Exception)

提供事件名称常量,支持类型安全的使用方式:

from elevenlabs.speech_engine import USER_TRANSCRIPT, INIT, CLOSE, DISCONNECTED, ERROR
session.on(USER_TRANSCRIPT, handle_transcript)

ConversationMessage

对话历史中的单条消息。每轮都会将完整转录文本传递给 on_transcript。

属性类型描述
role"user" | "agent"消息发送者。
contentstr消息的文本内容。

线协议

以下是通过 WebSocket 连接交换的 JSON 消息,供参考。SDK 会自动处理序列化和反序列化。

传入消息(ElevenLabs API 到开发者服务器)

消息类型字段描述
initconversation_id: string会话已初始化。
user_transcriptuser_transcript: TranscriptMessage[], event_id: number用户语音已转录。
ping保活。SDK 会响应 pong。
close正常断开连接。
errormessage: string来自 API 的错误。

传出消息(开发者服务器到 ElevenLabs API)

消息类型字段描述
agent_responsecontent: string, event_id: number, is_final: boolean用于 TTS 合成的 LLM 响应片段。
pong对 ping 的响应。