cdn/config/qwencloud-audio-tts-config.json
{
"tts": {
"default_model": "qwen-audio-3.0-tts-plus",
"default_voice": "Cherry",
"ws_models": [
"qwen-audio-3.0-tts-plus",
"qwen-audio-3.0-tts-flash"
],
"ws_default_voices": {
"qwen-audio-3.0-tts-plus": "longanlingxin",
"qwen-audio-3.0-tts-flash": "longanfengyue"
}
},
"cosyvoice": {
"default_model": "cosyvoice-v3-flash",
"default_voice": "longanyang",
"model_voices": {
"cosyvoice-v3-plus": [
"longanyang",
"longanhuan"
],
"cosyvoice-v3-flash": [
"longanyang",
"longanhuan",
"longhuhu_v3",
"longjielidou_v3",
"longshanshan_v3",
"longniuniu_v3",
"longanyue_v3",
"longshange_v3",
"longanmin_v3",
"longyingxiao_v3",
"longyingxun_v3",
"longyingtao_v3",
"longanyun_v3",
"longanwen_v3",
"longanli_v3",
"longanlang_v3",
"longyingmu_v3",
"longhua_v3",
"longwan_v3",
"longanzhi_v3",
"longanya_v3",
"longanqin_v3",
"longwanjun_v3",
"longyichen_v3",
"longlaobo_v3",
"longlaoyi_v3",
"longjiqi_v3",
"longhouge_v3",
"longdaiyu_v3",
"longanxuan_v3"
]
},
"instruction_voices": {
"cosyvoice-v3-plus": [
"longanyang",
"longanhuan"
],
"cosyvoice-v3-flash": [
"longanyang",
"longanhuan",
"longhuhu_v3"
]
}
}
}
cdn/config/qwencloud-token-plan-config.json
{
"personal_models": [
"qwen3.8-max",
"qwen3.8-flash",
"qwen3.7-max",
"qwen3.7-plus",
"qwen3.6-flash",
"qwen-image-3.0-pro",
"qwen-audio-3.0-tts-plus",
"qwen-audio-3.0-realtime-plus",
"qwen-audio-3.0-asr-flash",
"glm-5.3",
"glm-5.2",
"deepseek-v4-pro",
"deepseek-v4-pro-0813",
"deepseek-v4-flash-0731",
"deepseek-v4.1-flash",
"wan2.7-image",
"wan2.7-image-pro",
"happyhorse-1.1-i2v",
"happyhorse-1.1-t2v",
"happyhorse-1.1-r2v"
],
"team_models": [
"qwen3.8-max",
"qwen3.8-flash",
"qwen3.7-max",
"qwen3.7-plus",
"qwen3.6-plus",
"qwen3.6-flash",
"qwen-image-2.0",
"qwen-image-2.0-pro",
"qwen-image-3.0-pro",
"qwen-audio-3.0-tts-plus",
"qwen-audio-3.0-realtime-plus",
"qwen-audio-3.0-asr-flash",
"wan2.7-image",
"wan2.7-image-pro",
"deepseek-v4-pro",
"deepseek-v4-pro-0813",
"deepseek-v4-flash",
"deepseek-v4-flash-0731",
"deepseek-v4.1-flash",
"deepseek-v3.2",
"kimi-k2.7-code",
"glm-5.3",
"glm-5.2",
"glm-5.1",
"happyhorse-1.1-t2v",
"happyhorse-1.1-i2v",
"happyhorse-1.1-r2v"
]
}
cdn/references/qwencloud-audio-tts-models.md
# QwenCloud Audio TTS Models
> Catalog checked: 2026-09-18. Only model families released on or after 2025-08-01 are included.
## Defaults
| Scope | Runtime default | Configured default voice | Protocol and availability |
|-------|-----------------|--------------------------|---------------------------|
| Overall | `qwen-audio-3.0-tts-plus` | `longanlingxin` after WebSocket voice mapping | WebSocket; Token Plan + PAYG |
| Qwen-Audio Flash | `qwen-audio-3.0-tts-flash` | `longanfengyue` after WebSocket voice mapping | WebSocket; PAYG-only |
| Qwen3-TTS HTTP | `qwen3-tts-flash` | `Cherry` | API: HTTP with optional SSE; bundled `tts.py`: non-streaming HTTP; PAYG-only |
| CosyVoice | `cosyvoice-v3-flash` | `longanyang` | WebSocket; PAYG-only |
The generic CLI default voice is `Cherry`. When the voice is omitted for a Qwen-Audio WebSocket model, the runtime selects that model's configured default voice. An explicitly supplied voice—including explicit `Cherry`—is passed through unchanged. Explicit model and voice selections are not replaced.
## Runtime model catalog
| Model | Use | Protocol | Plan scope |
|-------|-----|----------|------------|
| `qwen-audio-3.0-tts-plus` | **Default.** Quality-oriented speech synthesis | WebSocket | Token Plan + PAYG |
| `qwen-audio-3.0-tts-flash` | Low-latency speech synthesis | WebSocket | PAYG-only |
| `qwen3-tts-flash` | Multilingual synthesis | API: HTTP with optional SSE; bundled `tts.py`: non-streaming HTTP | PAYG-only |
| `qwen3-tts-instruct-flash` | Instruction-controlled tone, emotion, pace, and character | API: HTTP with optional SSE; bundled `tts.py`: non-streaming HTTP | PAYG-only |
| `qwen3-tts-vd-2026-01-26` | Voice Design from a text description, followed by synthesis | Customization + HTTP | PAYG-only |
| `qwen3-tts-vc-2026-01-22` | Voice Cloning from an audio sample, followed by synthesis | Customization + HTTP | PAYG-only |
| `cosyvoice-v3-flash` | Fast CosyVoice synthesis | WebSocket through DashScope SDK | PAYG-only |
| `cosyvoice-v3-plus` | Quality-oriented CosyVoice synthesis | WebSocket through DashScope SDK | PAYG-only |
The Qwen3-TTS voice reference also documents real-time aliases and dated snapshots released after the cutoff. They are not listed as runtime models here because the bundled runtime uses the HTTP Qwen3-TTS path rather than the real-time Qwen3-TTS WebSocket protocol.
Qwen3-TTS's HTTP API supports optional SSE streaming, but the bundled `scripts/tts.py` HTTP path currently makes a non-streaming request and returns/downloads the resulting audio URL. Use the direct SSE example in `references/execution-guide.md` when streaming chunks are required.
## Limits and request compatibility
- Qwen3-TTS HTTP models accept at most 600 characters of synthesis text per request.
- `qwen3-tts-instruct-flash` accepts `instructions` of at most 1,600 tokens, in Chinese or English only.
- Qwen3-TTS non-streaming output URLs expire after 24 hours.
- `language_type` applies to the Qwen3-TTS HTTP path. WebSocket models use their own model-specific parameters and voices.
- A cloned or designed voice must be synthesized with the same target model used to create it.
## Qwen-Audio system voices
Voices are model-specific and cannot be interchanged without verification.
| Model | Supported system voices |
|-------|-------------------------|
| `qwen-audio-3.0-tts-plus` | `longanlingxin`, `longanlufeng` |
| `qwen-audio-3.0-tts-flash` | `longanfengyue`, `longanyuanfei`, `longanlingxi`, `longanxiaoxin`, `longanhuan_v3.6`, `longjielidou_v3.6`, `longpaopao_v3.6`, `longhuohuo_v3.6`, `longchuanshu_v3.6`, `loongmary`, `loongeva_v3.6`, `loongjohn` |
Both Qwen-Audio models also provide model-specific base voices generated through voice cloning. Use the exact model prefix documented in the official voice list.
## Qwen3-TTS system voices
The unversioned aliases expose the following current voice sets. Language support is voice-specific, especially for dialect voices.
### `qwen3-tts-instruct-flash` (24)
`Cherry`, `Serena`, `Ethan`, `Chelsie`, `Momo`, `Vivian`, `Moon`, `Maia`, `Kai`, `Nofish`, `Bella`, `Eldric Sage`, `Mia`, `Mochi`, `Bellona`, `Vincent`, `Bunny`, `Neil`, `Elias`, `Arthur`, `Nini`, `Seren`, `Pip`, `Stella`
### `qwen3-tts-flash` (48)
All 24 voices above, plus `Jennifer`, `Ryan`, `Katerina`, `Aiden`, `Bodega`, `Sonrisa`, `Alek`, `Dolce`, `Sohee`, `Ono Anna`, `Lenn`, `Emilien`, `Andre`, `Radio Gol`, `Jada`, `Dylan`, `Li`, `Marcus`, `Roy`, `Peter`, `Sunny`, `Eric`, `Rocky`, and `Kiki`.
## CosyVoice system voices
| Model | Supported system voices |
|-------|-------------------------|
| `cosyvoice-v3-plus` | `longanyang`, `longanhuan` |
| `cosyvoice-v3-flash` | `longanyang`, `longanhuan`, `longhuhu_v3`, `longjielidou_v3`, `longshanshan_v3`, `longniuniu_v3`, `longanyue_v3`, `longshange_v3`, `longanmin_v3`, `longyingxiao_v3`, `longyingxun_v3`, `longyingtao_v3`, `longanyun_v3`, `longanwen_v3`, `longanli_v3`, `longanlang_v3`, `longyingmu_v3`, `longhua_v3`, `longwan_v3`, `longanzhi_v3`, `longanya_v3`, `longanqin_v3`, `longwanjun_v3`, `longyichen_v3`, `longlaobo_v3`, `longlaoyi_v3`, `longjiqi_v3`, `longhouge_v3`, `longdaiyu_v3`, `longanxuan_v3` |
All listed CosyVoice voices support SSML and timestamps. Instruction support is voice-specific:
- `cosyvoice-v3-plus`: `longanyang`, `longanhuan`.
- `cosyvoice-v3-flash`: `longanyang`, `longanhuan`, `longhuhu_v3`.
## Billing and Token Plan compatibility
- PAYG text-to-speech is billed per **10,000 input characters**; output is not charged.
- A Chinese character, Japanese kanji, or Korean hanja counts as 2 characters. Letters, digits, punctuation, spaces, Japanese kana, and Korean letters count as 1 character. SSML tags are excluded.
- The official `qwen-audio-3.0-tts-plus` model page currently lists **$0.20 per 10,000 characters** and an RPM limit of **180**.
- `qwen-audio-3.0-tts-plus` is the only speech-synthesis model in the current Token Plan Individual and Team exact allowlists.
- `qwen-audio-3.0-tts-flash`, Qwen3-TTS, and CosyVoice models require PAYG.
cdn/references/qwencloud-model-pricing.md
# QwenCloud Model Pricing
> Checked: 2026-09-18. Prices are in USD.
>
> This file records the billing structure published by QwenCloud. The official pricing source no longer exposes a complete model-by-model price table, so unverified model lists and inferred prices are intentionally omitted. For current model availability, use the domain catalogs and the exact Token Plan allowlists.
## PAYG billing units
| Workload | Billing unit | Source-backed rules |
|----------|--------------|---------------------|
| Text generation | Per 1 million tokens | Input and output are priced separately. Some models use input-length tiers. Thinking tokens count as output tokens. |
| Image generation | Per successfully generated output image; some editing models also charge per input image | Text-to-image has no input-image charge. Input-image billing is model-specific. Output pricing may vary by resolution. Failed images are not charged. |
| Video generation | Per second | Output video is charged by generated duration. Some models also charge input-video duration. Price may vary by resolution, mode, or aspect ratio. |
| Text to speech | Per 10,000 input characters | Output is not charged. |
| Speech to text | Per second of input audio | Output is not charged. Applies to realtime recognition and file transcription. |
| Speech to speech | Per 1 million tokens | Rates differ by modality. Conversation history is billed again as input on later turns. |
| Embedding and reranking | Per 1 million input tokens | Output is not charged. |
### Text-to-speech character counting
- One Chinese character, Japanese kanji, or Korean hanja counts as 2 characters.
- Letters, digits, punctuation, spaces, Japanese kana, and Korean letters count as 1 character.
- SSML tags are excluded; only their text content is counted.
### Discounts and optimization
- Text-generation Batch API input and output rates are 50% of realtime rates.
- Embedding and reranking models that support Batch API are billed at 50% of realtime input pricing.
- Cached input tokens may receive a model-specific discount.
- Batch and cache discounts cannot be combined on the same text-generation request.
- Failed API calls are not charged and do not consume free quota.
## Built-in tool fees
| Tool | Fee | Note |
|------|-----|------|
| Web Search | $10 per 1,000 calls | In addition to model token costs |
| Image Search | $8 per 1,000 calls | Text-to-image and image-to-image search |
| Web Extractor | Free | Limited time |
| Code Interpreter | Free | Limited time |
Function calling and MCP have no separate tool fee; their tool descriptions count as input tokens.
## Token Plan
Token Plan uses Credits rather than PAYG billing units. Credits consumed by a request are dynamic and depend on the model type, token usage, thinking mode, and tool calls. The exact supported-model catalogs are maintained in `qwencloud-token-plan-models.md`:
- Individual: 20 models.
- Team: 27 models, comprising all 20 Individual models plus 7 Team-only models.
### Individual
| Tier | Standard price | Current promotional price | 7-day limit | Concurrent agents |
|------|----------------|---------------------------|-------------|-------------------|
| Lite | $8/month | $6/month | 2,500 Credits | 1–2 |
| Essential | $16/month | $10/month | 5,625 Credits | 2–3 |
| Standard | $25/month | $18/month | 10,000 Credits | 3–4 |
| Pro | $80/month | $68/month | 40,000 Credits | 6–8 |
| Credit Pack | — | $15/pack/month | 20,000 Credits/pack; not subject to the 7-day limit | — |
### Team
| Seat or pack | Standard price | Current promotional price | Monthly quota |
|--------------|----------------|---------------------------|---------------|
| Standard Seat | $30/seat/month | $20/seat/month | 25,000 Credits/seat/month |
| Pro Seat | $100/seat/month | $75/seat/month | 100,000 Credits/seat/month |
| Max Seat | $200/seat/month | — | 250,000 Credits/seat/month |
| Credit Pack | $700/pack | — | 625,000 Credits/pack |
Promotions, free quotas, prices, and model availability can change. Verify them on the official pages before making a purchase or cost estimate, and do not assume that a user's free quota remains available.
cdn/references/qwencloud-token-plan-models.md
# QwenCloud Token Plan Models
> Catalog checked: 2026-09-18. The tables are exact-string allowlists from the official Individual and Team overview pages. Only models released on or after 2025-08-01 are included.
## Billing and access scope
| Dimension | PAYG | Token Plan |
|-----------|------|------------|
| Key prefix | `sk-` or `sk-ws-` | `sk-sp-` |
| Billing | Model-specific units | Unified Credits |
| Protocol compatibility | Model/API-specific | OpenAI/Anthropic compatibility for chat-capable models; dedicated native or WebSocket endpoints for image, video, and audio models |
| Intended use | API usage | Interactive use in compatible AI programming and agent tools |
Token Plan publishes these Singapore endpoints: OpenAI-compatible `https://token-plan.ap-southeast-1.maas.aliyuncs.com/compatible-mode/v1`, Anthropic-compatible `https://token-plan.ap-southeast-1.maas.aliyuncs.com/apps/anthropic`, native multimodal `https://token-plan.ap-southeast-1.maas.aliyuncs.com/api/v1`, and TTS WebSocket `wss://token-plan.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference`. Obtain the API key from the QwenCloud console and honor a console-provided Base URL override if it differs.
## Token Plan Individual (20 models)
| Brand | Model ID | Capability |
|-------|----------|------------|
| Qwen | `qwen3.8-max` | Reasoning, visual understanding, text generation |
| Qwen | `qwen3.8-flash` | Reasoning, visual understanding, text generation |
| Qwen | `qwen3.7-max` | Reasoning, text generation |
| Qwen | `qwen3.7-plus` | Reasoning, visual understanding, text generation |
| Qwen | `qwen3.6-flash` | Reasoning, visual understanding, text generation |
| Qwen | `qwen-image-3.0-pro` | Image generation |
| Qwen | `qwen-audio-3.0-tts-plus` | Speech synthesis |
| Qwen | `qwen-audio-3.0-realtime-plus` | Realtime voice conversation |
| Qwen | `qwen-audio-3.0-asr-flash` | Speech recognition |
| Zhipu AI | `glm-5.3` | Reasoning, text generation |
| Zhipu AI | `glm-5.2` | Reasoning, text generation |
| DeepSeek | `deepseek-v4-pro` | Reasoning, text generation |
| DeepSeek | `deepseek-v4-pro-0813` | Reasoning, text generation |
| DeepSeek | `deepseek-v4-flash-0731` | Reasoning, text generation |
| DeepSeek | `deepseek-v4.1-flash` | Reasoning, visual understanding, text generation |
| Wan | `wan2.7-image` | Image generation |
| Wan | `wan2.7-image-pro` | Image generation |
| HappyHorse | `happyhorse-1.1-i2v` | Video generation |
| HappyHorse | `happyhorse-1.1-t2v` | Video generation |
| HappyHorse | `happyhorse-1.1-r2v` | Video generation |
## Token Plan Team (27 models)
Team includes all 20 Individual models and these 7 additional models:
| Brand | Model ID | Capability |
|-------|----------|------------|
| Qwen | `qwen3.6-plus` | Reasoning, visual understanding, text generation |
| Qwen | `qwen-image-2.0` | Image generation |
| Qwen | `qwen-image-2.0-pro` | Image generation |
| DeepSeek | `deepseek-v4-flash` | Reasoning, text generation |
| DeepSeek | `deepseek-v3.2` | Reasoning, text generation |
| Moonshot AI | `kimi-k2.7-code` | Reasoning, visual understanding, text generation |
| Zhipu AI | `glm-5.1` | Reasoning, text generation |
## Audio capabilities
- `qwen-audio-3.0-tts-plus`: speech synthesis and the default TTS model because it is available through both PAYG and Token Plan.
- `qwen-audio-3.0-realtime-plus`: realtime voice conversation.
- `qwen-audio-3.0-asr-flash`: speech recognition.
- Other Qwen3-TTS, Qwen-Audio-TTS Flash, and CosyVoice model IDs are not in the Token Plan exact allowlists and require PAYG.
## Plans and quotas
### Individual
| Tier | Standard price | Current promotional price | 7-day limit | Concurrent agents |
|------|----------------|---------------------------|-------------|-------------------|
| Lite | $8/month | $6/month | 2,500 Credits | 1–2 |
| Essential | $16/month | $10/month | 5,625 Credits | 2–3 |
| Standard | $25/month | $18/month | 10,000 Credits | 3–4 |
| Pro | $80/month | $68/month | 40,000 Credits | 6–8 |
| Credit Pack | — | $15/pack/month | 20,000 Credits/pack; not subject to the 7-day limit | — |
Credit Packs require an active Individual subscription, and an account may hold up to five at a time.
### Team
| Seat or pack | Standard price | Current promotional price | Monthly quota |
|--------------|----------------|---------------------------|---------------|
| Standard Seat | $30/seat/month | $20/seat/month | 25,000 Credits/seat/month |
| Pro Seat | $100/seat/month | $75/seat/month | 100,000 Credits/seat/month |
| Max Seat | $200/seat/month | — | 250,000 Credits/seat/month |
| Credit Pack | $700/pack | — | 625,000 Credits/pack |
Promotional pricing is time-sensitive. Check the official pages before purchase.
## Console URLs
- Token Plan Individual: https://home.qwencloud.com/analytics/token-plan/individual
- Token Plan Team: https://home.qwencloud.com/analytics/token-plan/team
- Team Management: https://tokenplan-enterprise.qwencloud.com
- Pricing: https://www.qwencloud.com/pricing/token-plan
references/api-guide.md
# Qwen Audio TTS — API Supplementary Guide
> **Content validity**: 2026-03 | **Sources**: [TTS API](https://docs.qwencloud.com/api-reference/speech-synthesis/qwen-tts) · [TTS Guide](https://docs.qwencloud.com/developer-guides/speech/tts) · [Voice List](https://docs.qwencloud.com/api-reference/speech-synthesis/voice-list)
---
## Definition
Text-to-speech synthesis service that produces natural, human-like voices. Supports system voices, multiple languages, streaming real-time playback, **natural language instruction control** for tone and emotion, and custom voices via **voice cloning** (from audio samples) and **voice design** (from text descriptions). Instruction-control availability and limits are model-specific; check the CDN model catalog before using `instructions`.
---
## Use Cases
> **🚫 CRITICAL — Never override user-specified parameters.** If the user explicitly specifies a model (e.g. `qwen-audio-3.0-tts-plus`), voice, or any parameter, you MUST use it exactly as given. Do NOT:
> - Replace the model with a "better suited" one (e.g. switching to `qwen3-tts-instruct-flash` for audiobooks/poetic/stylized text)
> - Add `instructions`, `optimize_instructions`, or `language_type` the user did not ask for
> - Rewrite the user's `text` (beyond necessary URL/file reference conversion)
>
> The model guidance below applies **only when the user has NOT specified a model**.
Fetch and read the current [QwenCloud audio TTS model catalog](https://alioth-intl.alicdn.com/skills-info/models/references/qwencloud-audio-tts-models.md) for scenario recommendations, defaults, compatibility, and basic model information. If CDN access fails, use the [local fallback](../cdn/references/qwencloud-audio-tts-models.md).
---
## Key Usage
### Non-streaming Synthesis
```python
import os, dashscope
dashscope.base_http_api_url = 'https://dashscope-intl.aliyuncs.com/api/v1'
resp = dashscope.MultiModalConversation.call(
model="qwen3-tts-flash",
api_key=os.getenv("DASHSCOPE_API_KEY"),
text="Today is a wonderful day to build something people love!",
voice="Cherry",
language_type="English",
)
print(resp.output.audio.url) # Audio URL, valid for 24 hours
```
### Streaming Synthesis (real-time playback)
The Qwen3-TTS API supports this SSE mode. The bundled `scripts/tts.py` HTTP path is non-streaming, so use this direct API/SDK pattern (or the curl example in `execution-guide.md`) when streaming chunks are required.
```python
resp = dashscope.MultiModalConversation.call(
model="qwen3-tts-flash",
api_key=os.getenv("DASHSCOPE_API_KEY"),
text="Today is a wonderful day to build something people love!",
voice="Cherry",
language_type="English",
stream=True,
)
for chunk in resp: # Each chunk contains Base64-encoded audio data
print(chunk)
```
### Instruction Control
Use natural language to describe the desired speech style. Before using the following API shape, verify `instructions` compatibility for the selected model in the CDN catalog linked above:
```python
resp = dashscope.MultiModalConversation.call(
model="qwen3-tts-instruct-flash",
api_key=os.getenv("DASHSCOPE_API_KEY"),
text="Welcome to our grand opening! We have amazing deals waiting for you!",
voice="Cherry",
instructions="Fast speech rate, with a clear rising intonation, suitable for introducing fashion products.",
optimize_instructions=True,
)
```
### System Voice List
Fetch the CDN model catalog linked above for the current voice list, descriptions, language support,
and model compatibility.
### Voice Cloning (quick example)
```python
import requests
# Step 1: Upload audio sample and create a voice
url = "https://dashscope-intl.aliyuncs.com/api/v1/services/audio/tts/customization"
resp = requests.post(url,
headers={"Authorization": f"Bearer {os.getenv('DASHSCOPE_API_KEY')}"},
files={"file": open("voice_sample.mp3", "rb")},
data={"model": "qwen-voice-enrollment", "target_model": "qwen3-tts-vc-2026-01-22",
"action": "create", "preferred_name": "my_brand_voice"})
custom_voice = resp.json()["output"]["voice"]
# Step 2: Use the cloned voice for synthesis
resp = dashscope.MultiModalConversation.call(
model="qwen3-tts-vc-2026-01-22", # Must match target_model
api_key=os.getenv("DASHSCOPE_API_KEY"),
text="Hello, this is a test with a cloned voice.",
voice=custom_voice,
)
```
### Key Parameters
| Parameter | Required | Description |
|-----------|----------|-------------|
| `model` | Yes | Model ID. |
| `text` | Yes | Text to synthesize; fetch the CDN model catalog linked above for model-specific limits. |
| `voice` | Yes | System voice ID, or cloned/designed voice name. |
| `language_type` | No | Default: `Auto`. Specifying the exact language significantly improves synthesis quality over `Auto`. Supported: `Chinese`, `English`, `Japanese`, `Korean`, `French`, `German`, `Russian`, `Italian`, `Spanish`, `Portuguese`. **Do not add it on the user's behalf** — if the user didn't set it, either omit it or suggest setting it and let them decide. |
| `instructions` | No | Natural language instructions for speech control; fetch the CDN model catalog linked above for current model compatibility and limits. |
| `optimize_instructions` | No | When true, the system semantically enhances `instructions` for better naturalness. Requires `instructions` to be set. |
| `stream` | No | `false` = returns audio URL. `true` = streams Base64-encoded audio chunks. |
---
## Important Notes
1. **Audio URLs are valid for only 24 hours.** Download immediately after generation.
2. **Specifying `language_type` significantly outperforms `Auto`.** When the text is in a single language, setting the exact language improves pronunciation accuracy and naturalness. However, only set it when the user requested it — otherwise omit it or suggest it to the user.
3. **`instructions` support is model-specific.** Fetch the CDN model catalog linked above before using it.
4. **Voice cloning `target_model` must match the synthesis `model`.** Otherwise synthesis fails.
5. **`SpeechSynthesizer` has been unified to `MultiModalConversation`.** In the DashScope Python SDK, the old `SpeechSynthesizer` interface has been replaced by `MultiModalConversation`. Parameters are fully compatible; only the interface name needs to change.
6. **Model availability is region-specific.** The CDN catalog does not provide region coverage; verify it with `qwencloud models info <model>` or the official model documentation.
7. **Streaming via HTTP** requires the header `X-DashScope-SSE: enable`. The Java SDK uses the `streamCall` interface.
8. **Bundled-script boundary:** `scripts/tts.py` does not currently expose Qwen3-TTS SSE streaming; its HTTP path requests the non-streaming audio URL.
---
## FAQ
**Q: How do I choose between qwen3-tts-flash and qwen3-tts-instruct-flash?**
A: Fetch and read the current CDN model catalog linked above for recommendations and capability differences. If the user already specified a model, keep it regardless of what the text looks like.
**Q: How do I make synthesized speech sound more natural?**
A: (1) Set `language_type` to match the text language only if the user wants it; otherwise suggest it. (2) If the user explicitly asks for style control, check the CDN catalog and use `instructions` only with a compatible model. Do NOT switch the user's model just to apply these tips. (3) Enable `optimize_instructions=True` only when `instructions` is set and the selected model supports it.
**Q: What is the difference between voice cloning and voice design?**
A: Cloning (VC) replicates a voice from an audio sample — suitable for reproducing an existing voice. Design (VD) creates a new voice from a text description — suitable for designing brand voices from scratch.
**Q: What is the maximum text length per synthesis call?**
A: Model limits are listed in the CDN model catalog linked above. For longer text, split it into segments and concatenate the audio files.
**Q: How do I achieve real-time audio playback?**
A: Set `stream=True`. The Python SDK returns a generator; iterate over it to receive Base64-encoded audio chunks for decoding and playback.
references/cosyvoice-guide.md
# CosyVoice TTS Guide
CosyVoice models use **WebSocket API** (not HTTP REST), requiring the DashScope SDK.
## Models
Fetch and read the current [QwenCloud audio TTS model catalog](https://alioth-intl.alicdn.com/skills-info/models/references/qwencloud-audio-tts-models.md) for the CosyVoice model list, defaults, instruction support, and voice compatibility. If CDN access fails, use the [local fallback](../cdn/references/qwencloud-audio-tts-models.md).
## Prerequisites
- **DashScope SDK** (venv recommended):
```bash
python3 -m venv .venv
source .venv/bin/activate # Windows: .venv\Scripts\activate
pip install dashscope>=1.25.17
```
- **API Key**: Same as Qwen TTS (`DASHSCOPE_API_KEY` or `QWEN_API_KEY`) — must be a **standard PAYG key** (`sk-...`); Token Plan keys are rejected (see below)
## Token Plan Compatibility
Fetch and read the current [Token Plan model catalog](https://alioth-intl.alicdn.com/skills-info/models/references/qwencloud-token-plan-models.md) before selecting a model. If CDN access fails, use the [local fallback](../cdn/references/qwencloud-token-plan-models.md).
The bundled `tts_cosyvoice.py` currently rejects Token Plan keys (`sk-sp-...`) and exits immediately with:
```text
Error: cosyvoice models are not available on Token Plan.
Use tts.py with qwen-audio-3.0-tts-plus instead (Token Plan supported).
```
For Token Plan TTS, use `tts.py` with an exact model from the current Token Plan catalog. Never replace an explicitly requested model without the user's confirmation.
## Run Script
**Discovery:** `python3 <this-skill-dir>/scripts/tts_cosyvoice.py --help`
```bash
python3 scripts/tts_cosyvoice.py --text "Hello, world!"
```
| Argument | Description |
|----------|-------------|
| `--text`, `-t` | **Required** — text to synthesize |
| `--model`, `-m` | Model ID; fetch the CDN model catalog linked above for the current default |
| `--voice`, `-v` | Voice ID; fetch the CDN model catalog linked above for the current default |
| `--output`, `-o` | Output file (default: `output/qwencloud-audio-tts/cosyvoice.mp3`) |
| `--format`, `-f` | Audio format: mp3, wav, pcm (default: mp3) |
| `--sample-rate` | Sample rate in Hz: 8000/16000/22050/24000/44100/48000 (default: 24000) |
| `--instruction` | Free-style instruction for speech control; fetch the CDN model catalog for compatibility |
| `--language-hints` | Target language hint (e.g. `zh`, `en`) |
## Available Voices
Fetch the CDN model catalog linked above for the current verified voices, descriptions, defaults, and
model compatibility. See the [official voice list](https://docs.qwencloud.com/api-reference/speech-synthesis/voice-list)
for the authoritative full catalog.
The script validates the exact model/voice pair before connecting. When `--instruction` is supplied, it also requires that exact voice to be instruction-compatible for the selected model; model-level support alone is not sufficient.
## Examples
> Example model/voice choices below are demonstrations only. If the user explicitly specified a model or voice, use exactly their choice — do not switch to `cosyvoice-v3-plus` (or any other model) because the text "sounds like professional narration".
```bash
# Basic synthesis
python3 scripts/tts_cosyvoice.py -t "Hello, world!"
# Chinese with specific voice
python3 scripts/tts_cosyvoice.py -t "你好世界" -v longanhuan
# Explicit model example
python3 scripts/tts_cosyvoice.py -t "Professional narration" -m cosyvoice-v3-plus
# Instruction-guided style control (verify model compatibility in the CDN catalog)
python3 scripts/tts_cosyvoice.py -t "欢迎光临" --instruction "用热情洋溢的声音" --language-hints zh
# WAV output at 48 kHz
python3 scripts/tts_cosyvoice.py -t "Hello" -f wav --sample-rate 48000
# Multiple files (use --output to avoid overwriting)
python3 scripts/tts_cosyvoice.py -t "First sentence" -o output/qwencloud-audio-tts/part1.mp3
python3 scripts/tts_cosyvoice.py -t "Second sentence" -o output/qwencloud-audio-tts/part2.mp3
```
> **Tip**: Default output overwrites previous file. Use `-o` with different filenames for batch tasks.
## Error Handling
| Error Pattern | Resolution |
|---------------|------------|
| `dashscope SDK not installed` | Run `pip install dashscope>=1.25.17` |
| `cosyvoice models are not available on Token Plan` | Token Plan keys are rejected — use `tts.py` with an exact supported model from the Token Plan catalog, or a PAYG key |
| `WebSocket task failed: <code>: <message>` | Structured WebSocket error — check `error_code` / `error_message` (e.g. invalid API key, invalid voice, quota) |
| `WebSocket connection failed` | Check network; verify API key |
| `voice ... is not supported by model` | Select a voice listed for that exact CosyVoice model; Qwen TTS voices such as Cherry/Ethan are invalid here |
| `--instruction is not supported for voice` | Select an instruction-compatible voice for that exact model, or omit `--instruction` |
## Character Counting & Billing
CosyVoice models are billed per character. Fetch the [CDN model-pricing reference](https://alioth-intl.alicdn.com/skills-info/models/references/qwencloud-model-pricing.md) for the current model list and billing units. If CDN access fails, use the [local fallback](../cdn/references/qwencloud-model-pricing.md).
references/execution-guide.md
# Qwen Audio TTS — Execution Guide
Fallback paths when the bundled script (Path 1) fails or is unavailable.
## Path 0 · Environment Fix
When the script fails due to environment issues (not API errors):
1. **`python3` not found**: Try `python --version` or `py -3 --version`. Use whichever returns 3.9+. If none work, help the user install Python 3.9+ from https://www.python.org/downloads/.
2. **Version too low** (`Python 3.9+ required` or `SyntaxError`): Install Python 3.9+ alongside existing Python, then use `python3.9` or `python3.11` explicitly.
3. **SSL errors** (`CERTIFICATE_VERIFY_FAILED`): On macOS, run `/Applications/Python\ 3.x/Install\ Certificates.command`. On Linux/Windows, set `SSL_CERT_FILE` to point to your CA bundle.
4. **Proxy**: Set `HTTPS_PROXY=http://proxy:port` before running the script.
After fixing, retry the script (Path 1). If the environment is unfixable, fall through to **Path 2 (curl)** below — curl is available on most systems without Python.
## Path 2 · Direct API Call (curl)
### Standard TTS
**Step 1 — Synthesize speech:**
```bash
curl -sS -X POST "https://dashscope-intl.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation" \
-H "Authorization: Bearer $DASHSCOPE_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"model": "qwen3-tts-flash",
"input": {
"text": "Hello, this is a test.",
"voice": "Cherry",
"language_type": "English"
}
}'
```
**Step 2 — Extract audio URL** from `output.audio.url`:
```json
{
"output": {
"audio": {
"url": "https://dashscope-result-bj.oss-cn-beijing.aliyuncs.com/...",
"format": "wav",
"sample_rate": 24000
}
},
"usage": {"characters": 25}
}
```
**Step 3 — Download the audio file** (URL valid 24 hours):
```bash
curl -o output.wav "AUDIO_URL_FROM_STEP_2"
```
### With instruction control (qwen3-tts-instruct-flash)
> **⚠️ Only use the instruct model + `instructions` when the user explicitly asked for instruction/style control.** This template is for that specific case — do NOT switch a user-specified model (e.g. `qwen-audio-3.0-tts-plus` or `qwen3-tts-flash`) to `qwen3-tts-instruct-flash`, and do NOT add `instructions` the user didn't request, even if the text "sounds like" an audiobook/radio-drama. Use the user's model as given.
```bash
curl -sS -X POST "https://dashscope-intl.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation" \
-H "Authorization: Bearer $DASHSCOPE_API_KEY" \
-H "Content-Type: application/json" \
-d '{
"model": "qwen3-tts-instruct-flash",
"input": {
"text": "Hello, this is a test.",
"voice": "Cherry",
"instructions": "Warm and calm tone, slightly slower pace.",
"optimize_instructions": true
}
}'
```
### Streaming (Base64 audio chunks via SSE)
This is a direct API fallback. Qwen3-TTS supports SSE at the API level, but the bundled `scripts/tts.py` HTTP path is non-streaming and does not consume its request JSON `stream` field.
```bash
curl -sS --no-buffer -X POST "https://dashscope-intl.aliyuncs.com/api/v1/services/aigc/multimodal-generation/generation" \
-H "Authorization: Bearer $DASHSCOPE_API_KEY" \
-H "Content-Type: application/json" \
-H "X-DashScope-SSE: enable" \
-d '{
"model": "qwen3-tts-flash",
"input": {
"text": "Hello, this is a test.",
"voice": "Cherry"
}
}'
```
Each SSE event contains a Base64-encoded audio chunk. Decode and concatenate to build the full audio file.
### Region endpoints
| Region | Base URL |
|--------|----------|
| Singapore (default) | `https://dashscope-intl.aliyuncs.com/api/v1` |
## Paths 3–5 · Fallback Cascade
When agent-executed paths (1–2) fail or shell is restricted:
**Path 3 — Generate Python script**: Read `scripts/tts.py` to understand the API logic. Write a self-contained Python script (stdlib `urllib`) tailored to the user's task. Present it for the user to save and run. Use `os.environ["DASHSCOPE_API_KEY"]` — never hardcode or expose the key.
**Path 4 — Generate curl commands**: Customize the curl templates from Path 2 with the user's specific parameters. Present as ready-to-copy commands.
**Path 5 — Autonomous resolution**: Read `scripts/tts.py` source and `references/*.md` to understand the full API contract. Reason about alternative approaches and implement.
If dependency errors occur: `python3 -m venv .venv && source .venv/bin/activate && pip install <package>`.
references/prompt-guide.md
# Audio TTS — Prompt Guide
Techniques for optimizing TTS text input and voice instructions.
> **🚫 Never override user-specified parameters.** If the user specified a model, voice, or text, use them exactly as given. The instruction templates below are only for a model currently marked instruction-compatible in the CDN model catalog, and only when the user asks for style control. NEVER switch models or inject instructions on your own initiative.
## Text Formatting for Natural Speech
TTS quality is highly sensitive to punctuation and structure:
| Technique | Effect | Example |
|-----------|--------|---------|
| Commas | Short pause | `Hello, welcome to our event.` |
| Periods | Full stop | `Thank you. Let's begin.` |
| Ellipsis | Dramatic pause | `And the winner is... Team Alpha!` |
| Exclamation | Energy / emphasis | `This is incredible!` |
| Question mark | Rising intonation | `Are you ready?` |
| Short sentences | Clearer delivery | Break 100+ word paragraphs into 1–2 sentence chunks |
For mixed-language text, set `language_type: "Auto"`.
## Instruction Templates
> These templates apply ONLY when: (1) the user did NOT specify a model, AND (2) the user explicitly asks for a specific style/tone. Otherwise, do not add instructions.
Fetch and read the current [QwenCloud audio TTS model catalog](https://alioth-intl.alicdn.com/skills-info/models/references/qwencloud-audio-tts-models.md) before choosing a model or instruction field. If CDN access fails, use the [local fallback](../cdn/references/qwencloud-audio-tts-models.md).
Use the instruction field documented for the selected model; do not interchange `instructions` and `instruction`.
| Scenario | Instructions |
|----------|-------------|
| News anchor | `Professional, authoritative tone. Clear enunciation, moderate pacing.` |
| Audiobook | `Warm, engaging narration. Varied pacing — slow for drama, faster for action.` |
| Children's story | `Cheerful, animated voice. Exaggerated intonation. Slow and playful.` |
| Product ad | `Confident enthusiasm. Dynamic pacing — measured start, building momentum. Emphasize product name.` |
| Meditation | `Very slow, soft, calm. Long pauses between sentences. Almost whispered.` |
| Tutorial | `Clear, patient, instructional. Moderate pace with brief pauses between steps.` |
General pattern:
```
Speak in a [emotion] tone with [speed] pacing. Use a [quality] voice,
as if [character/scenario]. [Additional: emphasis, pauses, accent].
```
## Voice Selection
Fetch the CDN model catalog linked above for the current voice list and model compatibility, then choose
a voice that matches the user's requested tone. Do not infer support for a voice that is absent from the catalog.
references/sources.md
# Qwen Audio TTS — Official Documentation
> **Enriched API guide**: See [api-guide.md](api-guide.md) for complete documentation with code examples and voice list (validity: 2026-03).
## API Reference
- **Qwen TTS API**: https://docs.qwencloud.com/api-reference/speech-synthesis/qwen-tts
## User Guide
- **Qwen TTS guide**: https://docs.qwencloud.com/developer-guides/speech/tts
## Voice List
- **Supported voices**: https://docs.qwencloud.com/api-reference/speech-synthesis/voice-list
## Models
- **Model list**: https://www.qwencloud.com/models
- **Model detail page**: `https://www.qwencloud.com/models/<model-name>` (e.g. https://www.qwencloud.com/models/qwen3-tts-flash)
scripts/gossamer.py
#!/usr/bin/env python3
"""Shared update-check signal emitter for qwencloud/qwencloud-ai.
Detects whether qwencloud-update-check is installed and emits structured
signals to stderr that agents can parse and act on:
[ACTION_REQUIRED] — qwencloud-update-check is not installed
[UPDATE_AVAILABLE] — a newer version is available
Self-contained, stdlib only.
"""
from __future__ import annotations
import json
import subprocess
import sys
import time
from pathlib import Path
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
_SKILLS_REPO = "QwenCloud/qwencloud-ai"
INSTALL_CMD = f"npx skills add {_SKILLS_REPO} --skill qwencloud-update-check -y"
_FATIGUE_INTERVAL = 86400 # 24 hours
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _find_repo_root(start: Path) -> Path | None:
for parent in [start] + list(start.parents):
if (parent / ".agents").is_dir() and parent.name != ".agents":
return parent
return None
def _state_file(repo: Path) -> Path:
return repo / ".agents" / "state.json"
def _read_state(repo: Path) -> dict:
sf = _state_file(repo)
if sf.exists():
try:
return json.loads(sf.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
pass
return {}
def _write_state(state: dict, repo: Path) -> None:
sf = _state_file(repo)
try:
sf.parent.mkdir(parents=True, exist_ok=True)
sf.write_text(
json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8"
)
except OSError:
pass
def _read_installed_skills(repo: Path) -> list[str]:
lf = repo / "skills-lock.json"
if lf.exists():
try:
data = json.loads(lf.read_text(encoding="utf-8"))
skills = data.get("skills", {})
if isinstance(skills, dict):
return sorted(k for k in skills if k.startswith("qwencloud-"))
except (json.JSONDecodeError, OSError):
pass
return []
def _build_update_cmd(repo: Path) -> str:
skills = _read_installed_skills(repo)
skill_args = " ".join(skills) if skills else ""
return f"npx skills add {_SKILLS_REPO} --skill {skill_args} -y"
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
def run(caller: str | Path | None = None) -> None:
"""Check update status and emit structured signals to stderr.
Parameters
----------
caller : str or Path, optional
``__file__`` of the calling script — used to locate the repo root.
Falls back to this module's own ``__file__`` if omitted.
"""
origin = Path(caller).resolve() if caller else Path(__file__).resolve()
repo = _find_repo_root(origin)
if repo is None:
return
state = _read_state(repo)
if state.get("never_install"):
return
ts = state.get("last_interaction", 0)
if (time.time() - ts) < _FATIGUE_INTERVAL:
return
# Look for qwencloud-update-check in known locations
for subdir in ("ops/qwencloud-update-check", "qwencloud-update-check"):
skill_dir = repo / "skills" / subdir
if not skill_dir.is_dir():
skill_dir = repo / ".agents" / "skills" / subdir
if skill_dir.is_dir():
script = skill_dir / "scripts" / "check_update.py"
if not script.is_file():
return
try:
proc = subprocess.run(
[sys.executable, str(script), "--print-response"],
capture_output=True, text=True, timeout=15,
)
if proc.returncode == 0 and proc.stdout.strip():
result = json.loads(proc.stdout)
if result.get("has_update"):
cmd = _build_update_cmd(repo)
print(
f"\n[UPDATE_AVAILABLE] qwencloud/qwencloud-ai has a newer version available.",
file=sys.stderr,
)
print(f" Run: {cmd}\n", file=sys.stderr)
except (subprocess.TimeoutExpired, OSError, json.JSONDecodeError):
pass
return
dismiss_install_prompt()
print(
f"[ACTION_REQUIRED] qwencloud-update-check skill is not installed. "
f"Install: {INSTALL_CMD}",
file=sys.stderr,
)
def dismiss_install_prompt() -> None:
"""Record a user interaction timestamp.
Suppresses all update-check signals for 24 hours.
Called after both "Install now" and "Skip for now".
"""
repo = _find_repo_root(Path(__file__).resolve())
if repo is None:
return
state = _read_state(repo)
state["last_interaction"] = time.time()
_write_state(state, repo)
def never_install_prompt() -> None:
"""Record that the user chose to never install qwencloud-update-check.
Permanently suppresses all update-check signals.
"""
repo = _find_repo_root(Path(__file__).resolve())
if repo is None:
return
state = _read_state(repo)
state["never_install"] = True
_write_state(state, repo)
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="Manage update-check preferences")
parser.add_argument("--never-install", action="store_true",
help="Permanently suppress update-check prompts")
parser.add_argument("--dismiss", action="store_true",
help="Suppress update-check prompts for 24 hours")
args = parser.parse_args()
if args.never_install:
never_install_prompt()
print("Update-check prompts permanently suppressed.", file=sys.stderr)
elif args.dismiss:
dismiss_install_prompt()
print("Update-check prompts suppressed for 24 hours.", file=sys.stderr)
scripts/qwencloud_lib.py
"""Shared cloud AI client infrastructure for qwencloud/qwencloud-ai skills.
Provider-based architecture: generic HTTP, env, and file-handling infrastructure
with pluggable AI provider implementations. DashScope is the built-in default;
additional providers can be registered via ``register_provider()``.
Modules are organised by responsibility:
- Environment & credentials: load_dotenv, find_repo_root, require_api_key
- Provider abstraction: AIProvider, DashScopeProvider, register_provider
- Region & endpoints: compat_base_url, native_base_url
- HTTP client: http_request, http_post, stream_sse
- File I/O: upload_local_file, resolve_file, download_file
- Request/response utilities: load_request, save_result, extract_text
- Update-check signal: run_update_signal
Stdlib only -- no pip install required.
"""
from __future__ import annotations
import sys
if sys.version_info < (3, 9):
print(
f"Error: Python 3.9+ required (found {sys.version}). "
"Install: https://www.python.org/downloads/",
file=sys.stderr,
)
sys.exit(1)
import base64
import http.client
import inspect
import json
import mimetypes
import os
import re
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
from pathlib import Path
from typing import Any, BinaryIO, Callable, Generator, Iterator, Optional
_MODEL_CONFIG_CDN_BASE = (
"https://alioth-intl.alicdn.com/skills-info/models/config/"
)
_MODEL_CONFIG_TIMEOUT_SECONDS = 3
# Keep the symlink path when this shared module is imported from an individual
# skill. Resolving it would incorrectly point at ``skills/_lib/cdn`` in the
# source repository instead of that skill's bundled fallback directory.
_MODEL_CONFIG_LOCAL_DIR = Path(__file__).absolute().parent.parent / "cdn" / "config"
_MODEL_CONFIG_CACHE: dict[tuple[str, str], dict[str, Any]] = {}
# ---------------------------------------------------------------------------
# Section 1: Generic environment & credential loading
# ---------------------------------------------------------------------------
def find_repo_root(start: Path) -> Path | None:
"""Walk up from *start* looking for a repository root marker.
Returns the first directory that contains ``.git`` or a ``skills/``
subdirectory, or ``None`` if no marker is found.
"""
for parent in [start] + list(start.parents):
if (parent / ".git").exists() or (parent / "skills").is_dir():
return parent
return None
def load_dotenv(path: Path) -> None:
"""Parse a simple ``.env`` file and inject into ``os.environ``.
Existing environment variables are **not** overwritten.
"""
if not path.exists():
return
for line in path.read_text(encoding="utf-8").splitlines():
line = line.strip()
if not line or line.startswith("#") or "=" not in line:
continue
key, value = line.split("=", 1)
key = key.strip()
value = value.strip().strip('"').strip("'")
if key and key not in os.environ:
os.environ[key] = value
def load_env(script_file: str | Path | None = None) -> None:
"""Load ``.env`` files with proper priority: current dir > repo root.
Parameters
----------
script_file : str or Path, optional
``__file__`` of the calling script, used to locate the repo root.
"""
load_dotenv(Path.cwd() / ".env")
origin = Path(script_file).resolve() if script_file else None
if origin:
repo = find_repo_root(origin)
if repo:
load_dotenv(repo / ".env")
def load_cdn_model_config(
filename: str,
*,
local_dir: str | Path | None = None,
required_keys: tuple[str, ...] = (),
validator: Callable[[dict[str, Any]], bool | None] | None = None,
) -> dict[str, Any]:
"""Load model configuration from the CDN, then a bundled local fallback.
``local_dir`` lets callers identify their own ``cdn/config`` directory.
This matters in the source tree, where this module is shared through
symlinks, while installed skills may contain a materialized copy.
"""
if (
Path(filename).name != filename
or "/" in filename
or "\\" in filename
or not filename.endswith(".json")
or not filename[:-5]
):
raise ValueError(f"Invalid model configuration filename: {filename!r}")
def validate(config: Any) -> dict[str, Any]:
if not isinstance(config, dict):
raise RuntimeError("expected a JSON object")
missing = [key for key in required_keys if key not in config]
if missing:
raise RuntimeError(f"missing {', '.join(missing)}")
if validator is not None and validator(config) is False:
raise RuntimeError("custom validation failed")
return config
url = urllib.parse.urljoin(_MODEL_CONFIG_CDN_BASE, filename)
fallback_dir = Path(local_dir) if local_dir is not None else _MODEL_CONFIG_LOCAL_DIR
local_path = fallback_dir / filename
if local_dir is None and not local_path.is_file():
repository_copy = (
Path(__file__).resolve().parents[2]
/ "skills-info"
/ "models"
/ "config"
/ filename
)
if repository_copy.is_file():
local_path = repository_copy
cache_key = (filename, str(local_path.absolute()))
cached = _MODEL_CONFIG_CACHE.get(cache_key)
if cached is not None:
try:
return validate(cached)
except (RuntimeError, TypeError, ValueError):
# A later caller may require a stricter schema for the same file.
# Re-resolve it so a valid bundled fallback can still recover.
_MODEL_CONFIG_CACHE.pop(cache_key, None)
request = urllib.request.Request(
url,
headers={"User-Agent": "qwencloud-skills"},
)
try:
with urllib.request.urlopen(request, timeout=_MODEL_CONFIG_TIMEOUT_SECONDS) as response:
config = validate(json.loads(response.read().decode("utf-8")))
except (
OSError,
TimeoutError,
UnicodeDecodeError,
json.JSONDecodeError,
http.client.HTTPException,
RuntimeError,
TypeError,
ValueError,
) as cdn_exc:
try:
config = validate(json.loads(local_path.read_text(encoding="utf-8")))
except (
OSError,
UnicodeDecodeError,
json.JSONDecodeError,
RuntimeError,
TypeError,
ValueError,
) as local_exc:
raise RuntimeError(
f"Unable to load valid model configuration from {url} or {local_path}: "
f"CDN error: {cdn_exc}; local error: {local_exc}"
) from local_exc
_MODEL_CONFIG_CACHE[cache_key] = config
return config
def mask_key(key: str) -> str:
"""Return a masked version of an API key for safe logging.
Shows the first 4 and last 4 characters with ``...`` in between.
Keys shorter than 9 characters are fully masked.
"""
if len(key) <= 8:
return "***"
return f"{key[:4]}...{key[-4:]}"
# Match both ordinary URLs and JSON-escaped slashes in SDK/API messages.
_DIAGNOSTIC_URL = re.compile(r"https?:(?:\\*/){2}[^\s<>\"']+", re.IGNORECASE)
# Prefixes of the same URL grammar, including partially written JSON escapes.
_DIAGNOSTIC_URL_PREFIX = re.compile(
r"(?:h|ht|htt|http|https|https?:(?:\\*|\\*/\\*|(?:\\*/){2}))$",
re.IGNORECASE,
)
def sanitize_diagnostic(text: str) -> str:
"""Remove URL queries/fragments from diagnostics, never from API data.
Drop the entire query instead of maintaining a list of credential names:
OSS V1/V4 signatures and temporary security tokens must all stay private.
Call this before truncating an external error message.
"""
def redact(match: re.Match) -> str:
url = match.group()
parts = re.split(r"([?#])", url, maxsplit=1)
if len(parts) == 1:
return url
if re.fullmatch(r"\[REDACTED\][.,;:!)\]}]*", parts[2]):
return url
# Preserve closing prose/Markdown delimiters outside the URL.
suffix = url[len(url.rstrip(".,;:!)]}")):]
return f"{parts[0]}{parts[1]}[REDACTED]{suffix}"
return _DIAGNOSTIC_URL.sub(redact, text)
class DiagnosticStream:
"""Redact streaming diagnostics even when a URL spans multiple writes.
Hold an unfinished URL or scheme prefix; ordinary text stays live.
``flush()`` finishes the message; callers must not flush between chunks.
Structured results should use the original data, not this stream.
"""
def __init__(self, stream: Any = None) -> None:
self._stream = sys.stderr if stream is None else stream
self._pending = ""
def write(self, text: str) -> None:
self._pending += text
end = len(self._pending)
for match in _DIAGNOSTIC_URL.finditer(self._pending):
if match.end() == len(self._pending):
end = match.start()
break
else:
# Even "ht" at the end of a chunk may start the next signed URL.
prefix = _DIAGNOSTIC_URL_PREFIX.search(self._pending)
if prefix:
end = prefix.start()
if end:
self._stream.write(sanitize_diagnostic(self._pending[:end]))
self._stream.flush()
self._pending = self._pending[end:]
def flush(self) -> None:
self._stream.write(sanitize_diagnostic(self._pending))
self._pending = ""
self._stream.flush()
# ---------------------------------------------------------------------------
# Section 2: AIProvider base class
# ---------------------------------------------------------------------------
class AIProvider:
"""Base interface for an AI service provider.
Subclasses must implement all methods. The built-in ``DashScopeProvider``
is registered by default; additional providers can be added via
``register_provider()`` and selected at runtime with the ``QWEN_PROVIDER``
environment variable.
"""
name: str = ""
# --- Authentication ---
def get_api_key_env_name(self) -> str:
"""Return the environment variable name for this provider's API key."""
raise NotImplementedError
def get_console_url(self) -> str:
"""Return the URL where users can obtain an API key."""
raise NotImplementedError
def validate_api_key(
self,
key: str,
*,
allow_coding_plan: bool = False,
domain: str = "",
) -> str:
"""Validate *key* and return it, or ``sys.exit`` with guidance.
Provider-specific checks (e.g. prefix validation) go here.
"""
raise NotImplementedError
# --- Endpoints ---
def compat_base_url(self) -> str:
"""Return the OpenAI-compatible API base URL."""
raise NotImplementedError
def native_base_url(self) -> str:
"""Return the provider's native API base URL."""
raise NotImplementedError
# --- HTTP headers ---
def make_headers(
self,
api_key: str,
payload: Any = None,
) -> dict[str, str]:
"""Build HTTP request headers.
*payload* is provided so that providers can inject conditional
headers (e.g. DashScope's ``X-DashScope-OssResourceResolve``).
"""
raise NotImplementedError
# --- Managed file URLs ---
def has_managed_url(self, obj: Any) -> bool:
"""Check whether *obj* contains any provider-managed file URLs."""
raise NotImplementedError
def managed_url_schemes(self) -> tuple[str, ...]:
"""Return URL scheme prefixes for provider-managed storage.
Used by ``resolve_file`` to detect URLs that should pass through
without modification (e.g. ``("oss://",)`` for DashScope).
"""
raise NotImplementedError
# --- File upload ---
def upload_file(self, api_key: str, model: str, fp: Path) -> str:
"""Upload a local file to provider temp storage.
Returns a provider-managed URL string.
"""
raise NotImplementedError
# --- Async task polling ---
def task_poll_url(self, task_id: str) -> str:
"""Build the URL to poll an async task's status."""
raise NotImplementedError
def extract_task_status(self, result: dict[str, Any]) -> str:
"""Extract the task status string from a poll response."""
raise NotImplementedError
def terminal_statuses(self) -> frozenset[str]:
"""Return the set of terminal task statuses."""
raise NotImplementedError
# ---------------------------------------------------------------------------
# Section 3: Token Plan configuration
# ---------------------------------------------------------------------------
_TOKEN_PLAN_MODEL_CONFIG = "qwencloud-token-plan-config.json"
_TOKEN_PLAN_COMPAT_BASE = "https://token-plan.ap-southeast-1.maas.aliyuncs.com/compatible-mode/v1"
_TOKEN_PLAN_NATIVE_BASE = "https://token-plan.ap-southeast-1.maas.aliyuncs.com/api/v1"
def _validate_token_plan_config(config: dict[str, Any]) -> bool:
personal = config.get("personal_models")
team = config.get("team_models")
if not isinstance(personal, list) or not all(isinstance(item, str) for item in personal):
return False
if not isinstance(team, list) or not all(isinstance(item, str) for item in team):
return False
return (
bool(personal)
and bool(team)
and len(personal) == len(set(personal))
and len(team) == len(set(team))
and set(personal).issubset(team)
)
def _load_token_plan_catalog() -> tuple[frozenset[str], frozenset[str]]:
config = load_cdn_model_config(
_TOKEN_PLAN_MODEL_CONFIG,
required_keys=("personal_models", "team_models"),
validator=_validate_token_plan_config,
)
personal = frozenset(config["personal_models"])
team = frozenset(config["team_models"])
return personal, team
def get_api_key_from_env(fallback_env: str = "DASHSCOPE_API_KEY") -> str:
"""Return the configured API key using the shared credential precedence.
``QWENCLOUD_API_KEY`` is the preferred public name. The legacy
``QWEN_API_KEY`` alias and the provider's ``fallback_env`` remain
supported for existing installations.
"""
return (
os.getenv("QWENCLOUD_API_KEY", "").strip()
or os.getenv("QWEN_API_KEY", "").strip()
or os.getenv(fallback_env, "").strip()
)
def is_token_plan_key(api_key: str | None = None) -> bool:
"""Return True if the given (or env-sourced) API key is a Token Plan key.
When *api_key* is ``None``, falls back to the shared environment lookup:
``QWENCLOUD_API_KEY``, then ``QWEN_API_KEY``, then
``DASHSCOPE_API_KEY``.
"""
if api_key is None:
api_key = get_api_key_from_env()
return api_key.startswith("sk-sp-")
def detect_api_key_type(script_file: str | Path | None = None) -> str:
"""Return the API key type: ``'token-plan'``, ``'payg'``, or ``'not-set'``.
Parameters
----------
script_file : str or Path, optional
``__file__`` of the calling script, used to locate ``.env`` files.
"""
load_env(script_file)
key = get_api_key_from_env()
if not key:
return "not-set"
return "token-plan" if key.startswith("sk-sp-") else "payg"
def is_token_plan_model(model: str) -> bool:
"""Check if a model is supported by Token Plan (any edition)."""
_, team_models = _load_token_plan_catalog()
return model in team_models
def is_token_plan_personal_model(model: str) -> bool:
"""Check if a model is supported by Token Plan Personal Edition."""
personal_models, _ = _load_token_plan_catalog()
return model in personal_models
def is_token_plan_model_supported(model: str, domain: str = "") -> bool:
"""Return whether Token Plan supports *model* in any edition.
``domain`` is retained for caller compatibility but model capabilities are
intentionally not duplicated in the Token Plan configuration.
"""
return is_token_plan_model(model)
def check_token_plan_model_support(model: str, domain: str = "") -> None:
"""Warn with actionable guidance when a Token Plan key is used with an unsupported model.
Prints a friendly suggestion to stderr (does not block execution — the API
may still accept it for edge cases we don't track). Call after
``require_api_key()`` so that ``.env`` files have been loaded.
Parameters
----------
model : str
Model ID being requested. Empty/None stays silent (nothing to check).
domain : str
Optional caller context used only in warning text. Model capabilities
are not stored in the Token Plan configuration. This function also does
not detect Personal vs Team entitlement.
"""
if not is_token_plan_key():
return # PAYG key — no restriction
if not model:
return
if is_token_plan_model_supported(model, domain):
return
capability_note = f" for {domain}" if domain else ""
lines = [f"Warning: model '{model}' is not available on Token Plan{capability_note}."]
_, team_models = _load_token_plan_catalog()
lines.append(
f"All Token Plan-supported models: "
f"{', '.join(sorted(team_models))}"
)
lines.append(
f"Alternative: switch to a supported model above, or use a PAYG key "
f"(sk-...) for '{model}'."
)
lines.extend([
"Personal docs: "
"https://docs.qwencloud.com/token-plan/personal/token-plan-personal-overview",
"Team docs: "
"https://docs.qwencloud.com/token-plan/team/token-plan-team-overview",
])
print("\n".join(lines), file=sys.stderr)
# ---------------------------------------------------------------------------
# Section 4: DashScopeProvider implementation
# ---------------------------------------------------------------------------
class DashScopeProvider(AIProvider):
"""Built-in provider for Alibaba Cloud Model Studio (DashScope)."""
name = "dashscope"
_CONSOLE_URL = "https://home.qwencloud.com/api-keys"
_COMPAT_BASE: dict[str, str] = {
"ap-southeast-1": "https://dashscope-intl.aliyuncs.com/compatible-mode/v1",
}
_NATIVE_BASE: dict[str, str] = {
"ap-southeast-1": "https://dashscope-intl.aliyuncs.com/api/v1",
}
# --- Authentication ---
def get_api_key_env_name(self) -> str:
return "DASHSCOPE_API_KEY"
def get_console_url(self) -> str:
return self._CONSOLE_URL
def validate_api_key(
self,
key: str,
*,
allow_coding_plan: bool = False,
domain: str = "",
) -> str:
self._current_key = key # Remember key for Token Plan routing
return key
# --- Endpoints ---
def compat_base_url(self) -> str:
url = os.getenv("QWEN_BASE_URL")
region = os.getenv("QWEN_REGION", "ap-southeast-1").lower()
if region == "none":
# When QWEN_REGION=none, skip URL assembly and return the user-configured base URL as-is.
if not url:
raise RuntimeError(
"QWEN_BASE_URL must be set when QWEN_REGION=none."
)
return url.rstrip("/")
if url:
return url.rstrip("/")
# Token Plan routing: sk-sp- key → TP endpoint
if is_token_plan_key(self._current_key if hasattr(self, "_current_key") else None):
return _TOKEN_PLAN_COMPAT_BASE
return self._COMPAT_BASE.get(region, self._COMPAT_BASE["ap-southeast-1"])
def native_base_url(self) -> str:
custom = os.getenv("QWEN_BASE_URL")
region = os.getenv("QWEN_REGION", "ap-southeast-1").lower()
if region == "none":
# When QWEN_REGION=none, skip URL assembly and return the user-configured base URL as-is.
if not custom:
raise RuntimeError(
"QWEN_BASE_URL must be set when QWEN_REGION=none."
)
return custom.rstrip("/")
if custom:
parsed = urllib.parse.urlparse(custom.rstrip("/"))
return f"{parsed.scheme}://{parsed.netloc}/api/v1"
# Token Plan routing
if is_token_plan_key(self._current_key if hasattr(self, "_current_key") else None):
return _TOKEN_PLAN_NATIVE_BASE
return self._NATIVE_BASE.get(region, self._NATIVE_BASE["ap-southeast-1"])
# --- HTTP headers ---
def make_headers(
self,
api_key: str,
payload: Any = None,
) -> dict[str, str]:
hdrs: dict[str, str] = {
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
"User-Agent": "qwencloud-skills", # Required for Token Plan
}
if payload and self.has_managed_url(payload):
hdrs["X-DashScope-OssResourceResolve"] = "enable"
source_cfg = build_source_config()
if source_cfg:
hdrs["X-DashScope-Source-Config"] = source_cfg
return hdrs
# --- Managed file URLs ---
def has_managed_url(self, obj: Any) -> bool:
if isinstance(obj, str):
return obj.startswith("oss://")
if isinstance(obj, dict):
return any(self.has_managed_url(v) for v in obj.values())
if isinstance(obj, (list, tuple)):
return any(self.has_managed_url(v) for v in obj)
return False
def managed_url_schemes(self) -> tuple[str, ...]:
return ("oss://",)
# --- File upload (OSS) ---
def upload_file(self, api_key: str, model: str, fp: Path) -> str:
"""Upload a local file and return a URL for API consumption.
When ``QWEN_TMP_OSS_BUCKET`` is set, files go to the user's own
OSS bucket and a presigned ``https://`` URL is returned.
Otherwise, files go to DashScope temp storage (48 h TTL) and an
``oss://`` URL is returned.
"""
if os.getenv("QWEN_TMP_OSS_BUCKET"):
# Custom OSS upload uses the user's own bucket/credentials and is
# available regardless of key type.
return self._upload_to_user_oss(fp)
# Token Plan keys cannot use the upload API
if is_token_plan_key(api_key):
raise RuntimeError(
"Local file upload is not available on Token Plan.\n"
"Alternative: host the file at a publicly accessible https:// URL "
"and pass that URL directly.\n"
"Docs: https://docs.qwencloud.com/token-plan/overview"
)
policy = self._get_upload_policy(api_key, model)
key = f"{policy['upload_dir']}/{fp.name}"
mime = mimetypes.guess_type(fp.name)[0] or "application/octet-stream"
fields = {
"OSSAccessKeyId": policy["oss_access_key_id"],
"Signature": policy["signature"],
"policy": policy["policy"],
"x-oss-object-acl": policy["x_oss_object_acl"],
"x-oss-forbid-overwrite": policy["x_oss_forbid_overwrite"],
"key": key,
"success_action_status": "200",
}
try:
with fp.open("rb") as file_obj:
body, ct, content_length = self._build_multipart(
fields, fp.name, file_obj, mime,
)
try:
req = urllib.request.Request(
policy["upload_host"],
data=body,
method="POST",
headers={
"Content-Type": ct,
"Content-Length": str(content_length),
},
)
with urllib.request.urlopen(req, timeout=120) as resp:
if resp.status != 200:
raise RuntimeError(f"OSS upload HTTP {resp.status}")
finally:
body.close()
except urllib.error.HTTPError as exc:
detail = ""
try:
detail = exc.read().decode("utf-8", errors="replace")
except Exception:
pass
raise RuntimeError(
f"OSS upload failed (HTTP {exc.code}): {detail[:300]}"
) from exc
return f"oss://{key}"
def _upload_to_user_oss(self, fp: Path) -> str:
"""Upload a file to the user's own OSS bucket via alibabacloud-oss-v2.
Returns a presigned ``https://`` URL. Requires
``QWEN_TMP_OSS_BUCKET`` and ``QWEN_TMP_OSS_REGION``.
Credentials are resolved from ``QWEN_TMP_OSS_AK_ID`` /
``QWEN_TMP_OSS_AK_SECRET`` first, falling back to
``OSS_ACCESS_KEY_ID`` / ``OSS_ACCESS_KEY_SECRET``
environment variables via the SDK's built-in provider.
"""
try:
import alibabacloud_oss_v2 as oss # type: ignore[import-untyped]
except ImportError:
print(
"Error: alibabacloud-oss-v2 is required for custom OSS upload.\n"
"Install: pip3 install alibabacloud-oss-v2\n"
"Docs: https://www.alibabacloud.com/help/en/oss/developer-reference/"
"simple-upload-using-oss-sdk-for-python-v2",
file=sys.stderr,
)
sys.exit(1)
from datetime import timedelta
bucket = os.environ["QWEN_TMP_OSS_BUCKET"]
region = os.getenv("QWEN_TMP_OSS_REGION", "")
if not region:
print(
"Error: QWEN_TMP_OSS_REGION is required when "
"QWEN_TMP_OSS_BUCKET is set.",
file=sys.stderr,
)
sys.exit(1)
ak_id = os.getenv("QWEN_TMP_OSS_AK_ID", "")
ak_secret = os.getenv("QWEN_TMP_OSS_AK_SECRET", "")
if ak_id and ak_secret:
credentials_provider = oss.credentials.StaticCredentialsProvider(
ak_id, ak_secret,
)
else:
ak_id = os.getenv("OSS_ACCESS_KEY_ID", "")
ak_secret = os.getenv("OSS_ACCESS_KEY_SECRET", "")
credentials_provider = (
oss.credentials.EnvironmentVariableCredentialsProvider()
)
cfg = oss.config.load_default()
cfg.credentials_provider = credentials_provider
cfg.region = region
endpoint = os.getenv("QWEN_TMP_OSS_ENDPOINT")
if endpoint:
cfg.endpoint = endpoint
client = oss.Client(cfg)
prefix = os.getenv("QWEN_TMP_OSS_PREFIX", "qwencloud-skill-uploads").strip("/")
from datetime import datetime as _dt
date_dir = _dt.now().strftime("%Y%m%d")
key = f"{prefix}/{date_dir}/{uuid.uuid4().hex[:8]}_{fp.name}"
mime = mimetypes.guess_type(fp.name)[0] or "application/octet-stream"
try:
uploader = oss.Uploader(
client,
part_size=8 * 1024 * 1024,
parallel_num=1,
)
uploader.upload_file(
oss.PutObjectRequest(
bucket=bucket,
key=key,
content_type=mime,
),
str(fp),
)
expires_s = int(os.getenv("QWEN_TMP_OSS_URL_EXPIRES", "86400"))
presign_result = client.presign(
oss.GetObjectRequest(bucket=bucket, key=key),
expires=timedelta(seconds=expires_s),
)
return presign_result.url
except Exception as exc:
msg = sanitize_diagnostic(str(exc))
if ak_id:
msg = msg.replace(ak_id, mask_key(ak_id))
if ak_secret:
msg = msg.replace(ak_secret, mask_key(ak_secret))
# The SDK exception may contain an unredacted signed URL too.
raise RuntimeError(f"Custom OSS upload failed: {msg}") from None
def _get_upload_policy(self, api_key: str, model: str) -> dict[str, Any]:
url = (
f"{self.native_base_url()}/uploads?"
f"{urllib.parse.urlencode({'action': 'getPolicy', 'model': model})}"
)
req = urllib.request.Request(
url,
method="GET",
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
},
)
with urllib.request.urlopen(req, timeout=30) as resp:
body = json.loads(resp.read().decode("utf-8"))
data = body.get("data")
if not data:
raise RuntimeError(
f"Upload policy error: {json.dumps(body, ensure_ascii=False)[:300]}"
)
return data
@staticmethod
def _build_multipart(
fields: dict[str, str],
fname: str,
file_obj: BinaryIO,
fmime: str,
) -> tuple[Generator[bytes, None, None], str, int]:
"""Build a streaming multipart body without buffering the file."""
boundary = uuid.uuid4().hex
parts: list[bytes] = []
for k, v in fields.items():
parts.append(
f'--{boundary}\r\nContent-Disposition: form-data; '
f'name="{k}"\r\n\r\n{v}\r\n'.encode()
)
parts.append(
f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="{fname}"\r\n'
f"Content-Type: {fmime}\r\n\r\n".encode()
)
prefix = b"".join(parts)
suffix = f"\r\n--{boundary}--\r\n".encode()
file_size = os.fstat(file_obj.fileno()).st_size
def stream() -> Generator[bytes, None, None]:
yield prefix
remaining = file_size
while remaining:
chunk = file_obj.read(min(1024 * 1024, remaining))
if not chunk:
raise RuntimeError("File changed while it was being uploaded")
remaining -= len(chunk)
yield chunk
yield suffix
content_length = len(prefix) + file_size + len(suffix)
return (
stream(),
f"multipart/form-data; boundary={boundary}",
content_length,
)
# --- Async task polling ---
def task_poll_url(self, task_id: str) -> str:
return f"{self.native_base_url()}/tasks/{task_id}"
def extract_task_status(self, result: dict[str, Any]) -> str:
return result.get("output", {}).get("task_status", "")
def terminal_statuses(self) -> frozenset[str]:
return frozenset({"SUCCEEDED", "FAILED", "CANCELED"})
# ---------------------------------------------------------------------------
# Section 4: Provider registry & resolution
# ---------------------------------------------------------------------------
_PROVIDERS: dict[str, type[AIProvider]] = {}
_cached_provider: AIProvider | None = None
def register_provider(name: str, cls: type[AIProvider]) -> None:
"""Register an AI provider class under *name*.
Example::
class MyProvider(AIProvider):
name = "my_provider"
...
register_provider("my_provider", MyProvider)
"""
_PROVIDERS[name.lower()] = cls
def get_provider() -> AIProvider:
"""Return the active ``AIProvider`` instance.
The provider is selected by the ``QWEN_PROVIDER`` environment variable
(default ``"dashscope"``). The instance is created lazily and cached
for the lifetime of the process.
"""
global _cached_provider
if _cached_provider is not None:
return _cached_provider
name = os.getenv("QWEN_PROVIDER", "dashscope").lower()
cls = _PROVIDERS.get(name)
if cls is None:
available = ", ".join(sorted(_PROVIDERS)) or "(none)"
print(
f"Error: Unknown provider '{name}'. "
f"Available providers: {available}",
file=sys.stderr,
)
sys.exit(1)
_cached_provider = cls()
return _cached_provider
def _reset_provider() -> None:
"""Clear the cached provider instance (for testing only)."""
global _cached_provider
_cached_provider = None
# Register built-in provider
register_provider("dashscope", DashScopeProvider)
# ---------------------------------------------------------------------------
# Section 5: Provider-delegating public facade functions
# ---------------------------------------------------------------------------
def require_api_key(
*,
script_file: str | Path | None = None,
allow_coding_plan: bool = False,
domain: str = "",
) -> str:
"""Load and return the API key for the active provider, or exit with guidance.
Parameters
----------
script_file : str or Path, optional
``__file__`` of the calling script -- used to locate ``.env``.
allow_coding_plan : bool
If ``False`` (default), Coding Plan keys emit a warning (DashScope).
domain : str
Human-readable domain name for error messages (e.g. "Image", "Video").
"""
load_env(script_file)
provider = get_provider()
# Priority: QWENCLOUD_API_KEY > QWEN_API_KEY > provider fallback.
fallback_env = provider.get_api_key_env_name() # DASHSCOPE_API_KEY
key = get_api_key_from_env(fallback_env)
if not key:
console = provider.get_console_url()
print(
f"Error: QWENCLOUD_API_KEY/QWEN_API_KEY/{fallback_env} not found.\n"
f"Option 1: Add to .env file in project root or current directory:\n"
f" echo 'QWENCLOUD_API_KEY=sk-your-key-here' >> .env\n"
f"Option 2: Export as environment variable:\n"
f" export QWENCLOUD_API_KEY='sk-...'\n"
f"Get key: {console}",
file=sys.stderr,
)
sys.exit(1)
return provider.validate_api_key(
key, allow_coding_plan=allow_coding_plan, domain=domain,
)
def compat_base_url() -> str:
"""Return the OpenAI-compatible API base URL for the active provider."""
return get_provider().compat_base_url()
def native_base_url() -> str:
"""Return the native API base URL for the active provider."""
return get_provider().native_base_url()
def chat_url() -> str:
"""Convenience: full OpenAI-compatible chat/completions endpoint."""
return f"{compat_base_url()}/chat/completions"
def has_oss_url(obj: Any) -> bool:
"""Check whether *obj* contains provider-managed file URLs.
Backward-compatible alias for ``get_provider().has_managed_url()``.
The name ``has_oss_url`` is retained for existing consumers; internally
this delegates to the active provider's ``has_managed_url()`` method.
"""
return get_provider().has_managed_url(obj)
def build_source_config(script_file: str | None = None) -> str | None:
"""Extract skill/agent from path. Supports .xxx/skills (dot-prefixed) and skills (OpenClaw)."""
path = script_file
if not path:
for frame_info in inspect.stack():
mod = inspect.getmodule(frame_info.frame)
if frame_info.filename.endswith('.py') and (mod.__name__ if mod else '') != 'qwencloud_lib':
path = frame_info.filename
break
if not path or not path.endswith('.py'):
return None
candidates = [path]
if sys.argv and sys.argv[0].endswith('.py'):
argv_path = sys.argv[0]
candidates.append(argv_path if os.path.isabs(argv_path) else os.path.join(os.getcwd(), argv_path))
# Match: [/.agent]/skills/skill-name/scripts/xxx.py
sep = r'[/\\]'
pattern = sep + r'(?:\.([^/\\]+)' + sep + r')?skills' + sep + r'([^/\\]+)' + sep + r'scripts' + sep + r'[^/\\]+\.py$'
for candidate in candidates:
m = re.search(pattern, candidate)
if m:
agent, skill = m.group(1) or "default", m.group(2)
if len(agent) <= 32 and len(skill) <= 32:
return json.dumps({"channel": "qwencloud-skill", "tags": {"t1": skill, "t2": agent}}, separators=(',', ':'))
return None
# ---------------------------------------------------------------------------
# Section 6: HTTP infrastructure
# ---------------------------------------------------------------------------
_RETRYABLE_CODES = frozenset({429, 500, 502, 503, 504})
def http_request(
method: str,
url: str,
api_key: str,
payload: dict[str, Any] | None = None,
*,
extra_headers: dict[str, str] | None = None,
timeout: int = 120,
retries: int = 2,
backoff: float = 1.5,
) -> dict[str, Any]:
"""Generic HTTP request with retry and exponential backoff.
Handles JSON serialisation, provider header injection, and retryable
HTTP codes.
"""
hdrs = get_provider().make_headers(api_key, payload)
if extra_headers:
hdrs.update(extra_headers)
data = json.dumps(payload).encode("utf-8") if payload else None
last_err = ""
for attempt in range(retries + 1):
req = urllib.request.Request(url, data=data, headers=hdrs, method=method)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
return json.loads(resp.read().decode("utf-8"))
except urllib.error.HTTPError as exc:
body = ""
try:
body = exc.read().decode("utf-8", errors="replace")
except Exception:
pass
if api_key and len(api_key) > 8 and api_key in body:
body = body.replace(api_key, mask_key(api_key))
last_err = f"HTTP {exc.code}: {sanitize_diagnostic(body)[:500]}"
if exc.code not in _RETRYABLE_CODES or attempt >= retries:
if exc.code == 401 and is_token_plan_key(api_key):
print(
"Hint: if this model is not on Token Plan, see supported "
"models at https://docs.qwencloud.com/token-plan",
file=sys.stderr,
)
raise RuntimeError(last_err) from None
time.sleep(backoff * (2 ** attempt))
except urllib.error.URLError as exc:
last_err = sanitize_diagnostic(str(exc.reason))
if attempt >= retries:
raise RuntimeError(f"Network error: {last_err}") from None
time.sleep(backoff * (2 ** attempt))
raise RuntimeError(last_err)
def http_post(
url: str,
api_key: str,
payload: dict[str, Any],
*,
timeout: int = 120,
retries: int = 2,
backoff: float = 1.5,
) -> dict[str, Any]:
"""Convenience: non-streaming POST, returns parsed JSON."""
return http_request(
"POST", url, api_key, payload,
timeout=timeout, retries=retries, backoff=backoff,
)
def _sse_error_message(
chunk: Any,
event_type: str,
api_key: str,
) -> Optional[str]:
"""Return a safe error message for an SSE error payload, if present."""
error_payload = chunk.get("error") if isinstance(chunk, dict) else None
has_top_level_error = isinstance(chunk, dict) and (
not chunk.get("choices")
and chunk.get("code") is not None
and chunk.get("message") is not None
)
if event_type.lower() != "error" and error_payload is None and not has_top_level_error:
return None
payload = error_payload if error_payload is not None else chunk
if isinstance(payload, dict):
code = payload.get("code") or payload.get("type")
message = payload.get("message")
if code and message:
detail = f"{code}: {message}"
elif code or message:
detail = str(code or message)
else:
detail = json.dumps(payload, ensure_ascii=False)
else:
detail = str(payload)
if api_key and len(api_key) > 8 and api_key in detail:
detail = detail.replace(api_key, mask_key(api_key))
return f"SSE error: {sanitize_diagnostic(detail)[:500]}"
def stream_sse(
url: str,
api_key: str,
payload: dict[str, Any],
*,
timeout: int = 180,
) -> Iterator[dict[str, Any]]:
"""Streaming POST. Yields parsed SSE ``data:`` chunks.
Automatically sets ``stream: true`` in the payload and sends the
appropriate ``Accept`` header.
"""
payload["stream"] = True
hdrs = get_provider().make_headers(api_key, payload)
hdrs["Accept"] = "text/event-stream"
data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(url, data=data, headers=hdrs, method="POST")
try:
resp = urllib.request.urlopen(req, timeout=timeout)
except urllib.error.HTTPError as exc:
body = ""
try:
body = exc.read().decode("utf-8", errors="replace")
except Exception:
pass
if api_key and len(api_key) > 8 and api_key in body:
body = body.replace(api_key, mask_key(api_key))
raise RuntimeError(f"HTTP {exc.code}: {sanitize_diagnostic(body)[:500]}") from None
except urllib.error.URLError as exc:
raise RuntimeError(f"Network error: {sanitize_diagnostic(str(exc.reason))}") from None
buf = b""
event_type = ""
try:
while True:
raw = resp.read(4096)
if not raw:
break
buf += raw
while b"\n" in buf:
line_bytes, buf = buf.split(b"\n", 1)
line = line_bytes.decode("utf-8", errors="replace").strip()
if not line:
event_type = ""
continue
if line.startswith("event:"):
event_type = line[6:].strip()
continue
if not line.startswith("data:"):
continue
json_str = line[5:].strip()
if json_str == "[DONE]":
return
try:
chunk = json.loads(json_str)
except json.JSONDecodeError:
if event_type.lower() == "error":
detail = json_str
if api_key and len(api_key) > 8 and api_key in detail:
detail = detail.replace(api_key, mask_key(api_key))
detail = sanitize_diagnostic(detail)[:500]
raise RuntimeError(f"SSE error: {detail}") from None
pass
else:
error_message = _sse_error_message(chunk, event_type, api_key)
if error_message:
raise RuntimeError(error_message)
yield chunk
event_type = ""
finally:
resp.close()
# ---------------------------------------------------------------------------
# Section 7: File upload / download
# ---------------------------------------------------------------------------
_BASE64_FILE_LIMIT = 7 * 1024 * 1024 # 7 MB (base64 adds ~33%; API limit is 10 MB)
def upload_local_file(api_key: str, model: str, fp: Path) -> str:
"""Upload a local file to the active provider's temp storage.
Returns a provider-managed URL (e.g. ``oss://`` for DashScope).
"""
return get_provider().upload_file(api_key, model, fp)
def resolve_file(
value: str,
*,
api_key: str | None = None,
model: str | None = None,
) -> str:
"""Resolve a file reference for API consumption.
URLs (``http``, ``https``, ``data``, and provider-managed schemes)
pass through unchanged. Local files are handled based on context:
- *api_key* + *model* provided: upload to temp storage.
- Otherwise: convert to ``data:`` base64 URI (must be < 7 MB).
"""
provider = get_provider()
pass_through = ("http://", "https://", "data:") + provider.managed_url_schemes()
if value.startswith(pass_through):
return value
if value.startswith("file://"):
parsed = urllib.parse.urlparse(value)
if parsed.netloc and parsed.netloc.lower() != "localhost":
raise ValueError(
f"Cannot resolve non-local file URI authority {parsed.netloc!r}; "
"only an empty authority or 'localhost' is supported."
)
value = urllib.request.url2pathname(parsed.path)
p = Path(value)
if not (p.exists() and p.is_file()):
return value
file_size = p.stat().st_size
if api_key and model:
managed_url = provider.upload_file(api_key, model, p)
tag = "48 h TTL" if managed_url.startswith("oss://") else "custom OSS"
print(sanitize_diagnostic(f"Uploaded {p.name} -> {managed_url} ({tag})"), file=sys.stderr)
return managed_url
if file_size > _BASE64_FILE_LIMIT:
print(
f"Warning: {p.name} is {file_size / 1024 / 1024:.1f} MB -- "
"base64 may exceed the 10 MB API limit. "
"Use --upload-files to auto-upload, or provide an online URL.",
file=sys.stderr,
)
mime = mimetypes.guess_type(p.name)[0] or "application/octet-stream"
b64 = base64.b64encode(p.read_bytes()).decode("ascii")
return f"data:{mime};base64,{b64}"
def download_file(url: str, dest: Path, *, timeout: int = 120) -> Path:
"""Download a file from *url* to *dest*, creating parent dirs as needed."""
dest.parent.mkdir(parents=True, exist_ok=True)
req = urllib.request.Request(url, headers={"User-Agent": "qwencloud-skills"})
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
dest.write_bytes(resp.read())
except urllib.error.URLError as exc:
raise RuntimeError(f"Download failed: {sanitize_diagnostic(str(exc))}") from None
return dest
# ---------------------------------------------------------------------------
# Section 8: Request / response utilities
# ---------------------------------------------------------------------------
def load_request(args: Any) -> dict[str, Any]:
"""Load request dict from ``--request`` or ``--file`` CLI argument."""
if getattr(args, "request", None):
return json.loads(args.request)
if getattr(args, "file", None):
return json.loads(Path(args.file).read_text(encoding="utf-8"))
raise ValueError("Provide --request '{...}' or --file path/to/request.json")
def save_result(result: dict[str, Any], output_path: str | Path) -> None:
"""Write *result* as JSON to *output_path*, creating parent dirs."""
p = Path(output_path)
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
def extract_text(content: Any) -> str:
"""Extract plain text from a message ``content`` field.
Handles ``str``, ``list[{type, text}]``, and ``None``.
"""
if isinstance(content, str):
return content
if isinstance(content, list):
for item in content:
if isinstance(item, dict) and item.get("type") == "text":
text = item.get("text")
if isinstance(text, str):
return text
return json.dumps(content, ensure_ascii=False)
if content is None:
return ""
return json.dumps(content, ensure_ascii=False)
def try_parse_json(text: str) -> Any | None:
"""Attempt to parse *text* as JSON; return ``None`` on failure."""
text = text.strip()
if not text:
return None
try:
return json.loads(text)
except json.JSONDecodeError:
return None
# ---------------------------------------------------------------------------
# Section 9: Async task polling
# ---------------------------------------------------------------------------
def poll_task(
task_id: str,
api_key: str,
*,
timeout_s: int = 600,
interval: int = 10,
verbose: bool = True,
) -> dict[str, Any]:
"""Poll an async task until it reaches a terminal state.
Returns the final result dict. Raises ``TimeoutError`` if
*timeout_s* is exceeded.
"""
provider = get_provider()
url = provider.task_poll_url(task_id)
terminals = provider.terminal_statuses()
start = time.time()
poll_count = 0
while True:
result = http_request("GET", url, api_key)
elapsed = int(time.time() - start)
poll_count += 1
status = provider.extract_task_status(result)
if verbose:
print(f" [{elapsed}s] task={task_id} status={status}", file=sys.stderr)
if status in terminals:
return result
if time.time() - start > timeout_s:
raise TimeoutError(
f"Task {task_id} timed out after {timeout_s}s "
f"({poll_count} polls, last status: {status})"
)
time.sleep(interval)
# ---------------------------------------------------------------------------
# Section 10: Update-check signal
# ---------------------------------------------------------------------------
try:
from gossamer import run as _run_update_signal_impl
except ImportError:
_run_update_signal_impl = None # type: ignore[assignment]
def run_update_signal(caller: str | Path | None = None) -> None:
"""Emit update-check signals to stderr (non-blocking, failure-safe).
Parameters
----------
caller : str or Path, optional
``__file__`` of the calling script.
"""
if _run_update_signal_impl:
try:
_run_update_signal_impl(caller=caller or __file__)
except Exception:
pass
scripts/tts_cosyvoice.py
#!/usr/bin/env python3
"""Synthesize speech from text via CosyVoice models (WebSocket API).
CosyVoice models require the DashScope SDK (WebSocket-based, not HTTP REST).
Run with --help for usage.
Dependencies:
pip install dashscope>=1.25.17
Or with venv:
python3 -m venv .venv && source .venv/bin/activate && pip install dashscope>=1.25.17
"""
from __future__ import annotations
import sys
if sys.version_info < (3, 9):
print(f"Error: Python 3.9+ required (found {sys.version}).", file=sys.stderr)
sys.exit(1)
# Check dashscope dependency before other imports
try:
import dashscope
from dashscope.audio.tts_v2 import SpeechSynthesizer
from dashscope.audio.tts_v2.speech_synthesizer import AudioFormat
except ImportError:
print(
"Error: dashscope SDK not installed.\n\n"
"Install with:\n"
" pip install dashscope>=1.25.17\n\n"
"Or use venv:\n"
" python3 -m venv .venv\n"
" source .venv/bin/activate # Windows: .venv\\Scripts\\activate\n"
" pip install dashscope>=1.25.17",
file=sys.stderr,
)
sys.exit(1)
import argparse
import json
import urllib.parse
from pathlib import Path
from typing import Any
sys.path.insert(0, str(Path(__file__).resolve().parent))
from qwencloud_lib import ( # noqa: E402
build_source_config,
load_cdn_model_config,
native_base_url,
require_api_key,
run_update_signal,
)
# is_token_plan_key is being added to qwencloud_lib; fall back to a local
# check until the shared helper is exported in every installation.
try:
from qwencloud_lib import is_token_plan_key # noqa: E402,F401
except ImportError:
def is_token_plan_key(api_key: str) -> bool: # type: ignore[misc]
return api_key.startswith("sk-sp-")
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
SKILL_USER_AGENT = "qwencloud-skills" # Required header for Token Plan routing
_MODEL_CONFIG_DIR = Path(__file__).resolve().parent.parent / "cdn" / "config"
# The WebSocket endpoint is derived at runtime from the shared native base
# URL (see qwencloud_lib), so a custom QWEN_BASE_URL override and Token Plan
# routing apply to the WebSocket path automatically. With no override,
# native_base_url() defaults to https://dashscope-intl.aliyuncs.com/api/v1,
# resolving to wss://dashscope-intl.aliyuncs.com/api-ws/v1/inference.
_WEBSOCKET_PATH = "/api-ws/v1/inference"
def _validate_model_config(config: dict[str, Any]) -> bool:
cosyvoice = config.get("cosyvoice")
if not isinstance(cosyvoice, dict):
return False
model_voices = cosyvoice.get("model_voices")
instruction_voices = cosyvoice.get("instruction_voices")
if not (
isinstance(cosyvoice.get("default_model"), str)
and bool(cosyvoice["default_model"])
and isinstance(cosyvoice.get("default_voice"), str)
and bool(cosyvoice["default_voice"])
and isinstance(model_voices, dict)
and bool(model_voices)
and all(
isinstance(model, str)
and model
and isinstance(voices, list)
and bool(voices)
and all(isinstance(voice, str) and voice for voice in voices)
for model, voices in model_voices.items()
)
and isinstance(instruction_voices, dict)
and all(
isinstance(model, str)
and model in model_voices
and isinstance(voices, list)
and all(isinstance(voice, str) and voice for voice in voices)
and set(voices).issubset(model_voices[model])
for model, voices in instruction_voices.items()
)
):
return False
default_model = cosyvoice["default_model"]
return (
default_model in model_voices
and cosyvoice["default_voice"] in model_voices[default_model]
)
def _model_config() -> dict[str, Any]:
return load_cdn_model_config(
"qwencloud-audio-tts-config.json",
local_dir=_MODEL_CONFIG_DIR,
required_keys=("cosyvoice",),
validator=_validate_model_config,
)["cosyvoice"]
def _default_model() -> str:
return _model_config()["default_model"]
def _default_voice() -> str:
return _model_config()["default_voice"]
def _voice_map(key: str) -> dict[str, frozenset[str]]:
return {
model: frozenset(voices)
for model, voices in _model_config()[key].items()
}
_SAMPLE_RATES = (8000, 16000, 22050, 24000, 44100, 48000)
# (format, sample_rate) -> SDK AudioFormat enum for the WebSocket synthesizer
_AUDIO_FORMATS = {
("mp3", 8000): AudioFormat.MP3_8000HZ_MONO_128KBPS,
("mp3", 16000): AudioFormat.MP3_16000HZ_MONO_128KBPS,
("mp3", 22050): AudioFormat.MP3_22050HZ_MONO_256KBPS,
("mp3", 24000): AudioFormat.MP3_24000HZ_MONO_256KBPS,
("mp3", 44100): AudioFormat.MP3_44100HZ_MONO_256KBPS,
("mp3", 48000): AudioFormat.MP3_48000HZ_MONO_256KBPS,
("wav", 8000): AudioFormat.WAV_8000HZ_MONO_16BIT,
("wav", 16000): AudioFormat.WAV_16000HZ_MONO_16BIT,
("wav", 22050): AudioFormat.WAV_22050HZ_MONO_16BIT,
("wav", 24000): AudioFormat.WAV_24000HZ_MONO_16BIT,
("wav", 44100): AudioFormat.WAV_44100HZ_MONO_16BIT,
("wav", 48000): AudioFormat.WAV_48000HZ_MONO_16BIT,
("pcm", 8000): AudioFormat.PCM_8000HZ_MONO_16BIT,
("pcm", 16000): AudioFormat.PCM_16000HZ_MONO_16BIT,
("pcm", 22050): AudioFormat.PCM_22050HZ_MONO_16BIT,
("pcm", 24000): AudioFormat.PCM_24000HZ_MONO_16BIT,
("pcm", 44100): AudioFormat.PCM_44100HZ_MONO_16BIT,
("pcm", 48000): AudioFormat.PCM_48000HZ_MONO_16BIT,
}
def _ws_url() -> str:
"""Derive the WebSocket inference endpoint from the shared native base URL.
The ws/wss scheme mirrors the base URL's http/https scheme and the host
is preserved, so QWEN_BASE_URL overrides and Token Plan routing (both
handled inside ``native_base_url()``) apply automatically.
"""
parsed = urllib.parse.urlparse(native_base_url())
scheme = "wss" if parsed.scheme == "https" else "ws"
return f"{scheme}://{parsed.netloc}{_WEBSOCKET_PATH}"
def _resolve_audio_format(fmt: str, sample_rate: int) -> AudioFormat:
try:
return _AUDIO_FORMATS[(fmt, sample_rate)]
except KeyError:
print(
f"Error: unsupported format/sample-rate combination: "
f"{fmt}@{sample_rate}Hz. "
f"Sample rates: {', '.join(str(r) for r in _SAMPLE_RATES)}.",
file=sys.stderr,
)
sys.exit(1)
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main() -> None:
run_update_signal(caller=__file__)
parser = argparse.ArgumentParser(
description="CosyVoice TTS via DashScope SDK (WebSocket)",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=f"""\
model examples (current compatibility/defaults load from CDN configuration):
cosyvoice-v3-flash High quality, fast, supports system voices
cosyvoice-v3-plus Highest quality, supports system voices
voice compatibility:
cosyvoice-v3-plus longanyang, longanhuan
cosyvoice-v3-flash See CDN model configuration/catalog for the full list
instruction compatibility:
The selected model and voice must both be instruction-compatible. The script
validates the exact pair against CDN configuration before making a request.
note:
CosyVoice models are NOT available on Token Plan. With a Token Plan key
(sk-sp-...), use tts.py with qwen-audio-3.0-tts-plus instead.
For Qwen TTS models (qwen3-tts-*, qwen-audio-3.0-tts-plus), use tts.py.
examples:
# Basic synthesis
python {Path(__file__).name} --text "Hello, world!"
# Chinese with specific voice
python {Path(__file__).name} --text "你好世界" --voice longanhuan
# High quality model
python {Path(__file__).name} --text "Hello" --model cosyvoice-v3-plus
# Instruction-guided style control (cosyvoice-v3-flash)
python {Path(__file__).name} --text "欢迎光临" --instruction "用热情洋溢的声音" --language-hints zh
# WAV output at 48 kHz
python {Path(__file__).name} --text "Hello" --format wav --sample-rate 48000
# Save to specific file
python {Path(__file__).name} --text "Hello" --output hello.mp3
""",
)
parser.add_argument("--text", "-t", required=True, help="Text to synthesize")
parser.add_argument("--model", "-m", default=None,
help="Model (default loaded from CDN model configuration)")
parser.add_argument("--voice", "-v", default=None,
help="Voice (default loaded from CDN model configuration)")
parser.add_argument("--output", "-o", type=Path, default=Path("output/qwencloud-audio-tts/cosyvoice.mp3"), help="Output file (default: output/qwencloud-audio-tts/cosyvoice.mp3)")
parser.add_argument("--format", "-f", default="mp3", choices=["mp3", "wav", "pcm"], help="Audio format (default: mp3)")
parser.add_argument("--sample-rate", type=int, default=24000, choices=_SAMPLE_RATES,
help="Sample rate in Hz (default: 24000)")
parser.add_argument("--instruction", type=str, default=None,
help="Free-style speech instruction; supported voices are loaded from CDN configuration")
parser.add_argument("--language-hints", type=str, default=None,
help="Target language hint (e.g. zh, en)")
args = parser.parse_args()
api_key = require_api_key(script_file=__file__, domain="CosyVoice TTS")
if is_token_plan_key(api_key):
print(
"Error: cosyvoice models are not available on Token Plan.\n"
"Use tts.py with qwen-audio-3.0-tts-plus instead (Token Plan supported).",
file=sys.stderr,
)
sys.exit(1)
# Model priority: CLI > default
try:
model = args.model or _default_model()
voice = args.voice or _default_voice()
model_voices = _voice_map("model_voices")
instruction_voices = _voice_map("instruction_voices")
except RuntimeError as exc:
print(f"Error: {exc}", file=sys.stderr)
sys.exit(1)
if model not in model_voices:
print(
f"Error: unsupported CosyVoice model '{model}'. Supported models: "
f"{', '.join(sorted(model_voices))}",
file=sys.stderr,
)
sys.exit(1)
if voice not in model_voices[model]:
print(
f"Error: voice '{voice}' is not supported by model '{model}'. "
f"Supported voices: {', '.join(sorted(model_voices[model]))}",
file=sys.stderr,
)
sys.exit(1)
if args.instruction and voice not in instruction_voices.get(model, frozenset()):
supported = ", ".join(sorted(instruction_voices.get(model, frozenset()))) or "none"
print(
f"Error: --instruction is not supported for voice '{voice}' with model "
f"'{model}'. Instruction-compatible voices: {supported}",
file=sys.stderr,
)
sys.exit(1)
# Setup: derive the WebSocket endpoint AFTER require_api_key so that
# Token Plan / custom QWEN_BASE_URL routing inside native_base_url() is
# already in effect.
dashscope.api_key = api_key
websocket_url = _ws_url()
dashscope.base_websocket_api_url = websocket_url
# Synthesize
print(f"Synthesizing (WebSocket): model={model}, voice={voice}", file=sys.stderr)
try:
headers = {"User-Agent": SKILL_USER_AGENT}
source_config = build_source_config(__file__)
if source_config:
headers["X-DashScope-Source-Config"] = source_config
synthesizer = SpeechSynthesizer(
model=model,
voice=voice,
format=_resolve_audio_format(args.format, args.sample_rate),
instruction=args.instruction or None,
language_hints=[args.language_hints] if args.language_hints else None,
headers=headers,
url=websocket_url,
)
audio_data = synthesizer.call(args.text)
except Exception as e:
print(f"Error: {e}", file=sys.stderr)
sys.exit(1)
if not audio_data:
# Structured WebSocket error extraction: pull error_code /
# error_message / task_id from the last task-failed response header.
try:
response = synthesizer.get_response()
except Exception:
response = None
header = response.get("header", {}) if isinstance(response, dict) else {}
if header.get("event") == "task-failed":
error_code = header.get("error_code", "UnknownError")
error_message = header.get("error_message", "Unknown error")
task_id = header.get("task_id", "")
suffix = f" (task_id={task_id})" if task_id else ""
print(
f"Error: WebSocket task failed: {error_code}: "
f"{error_message}{suffix}",
file=sys.stderr,
)
else:
print("Error: No audio data returned.", file=sys.stderr)
sys.exit(1)
# Save
output = args.output
if output.suffix.lower() not in {".mp3", ".wav", ".pcm"}:
output = output.with_suffix(f".{args.format}")
output.parent.mkdir(parents=True, exist_ok=True)
output.write_bytes(audio_data)
print(f"Audio saved to {output}", file=sys.stderr)
print(json.dumps({"audio_file": str(output), "size_bytes": len(audio_data)}))
if __name__ == "__main__":
main()
scripts/tts.py
#!/usr/bin/env python3
"""Synthesize speech from text via Qwen TTS models on DashScope API.
Supports multiple voices, instruction-controlled style, and automatic audio
download. Self-contained, stdlib only.
Protocol routing:
- qwen3-tts-flash, qwen3-tts-instruct-flash → HTTP REST API
- qwen-audio-3.0-tts-plus/flash → raw WebSocket duplex streaming
- cosyvoice-v3-* → scripts/tts_cosyvoice.py (DashScope SDK)
"""
from __future__ import annotations
import sys
if sys.version_info < (3, 9):
print(f"Error: Python 3.9+ required (found {sys.version}). "
"Install: https://www.python.org/downloads/", file=sys.stderr)
sys.exit(1)
import argparse
import hashlib
import json
import os
import socket
import ssl
import struct
import time
import uuid as uuid_mod
from pathlib import Path
from typing import Any
from urllib.parse import urlparse
sys.path.insert(0, str(Path(__file__).resolve().parent))
from qwencloud_lib import ( # noqa: E402
check_token_plan_model_support,
download_file,
http_request,
load_cdn_model_config,
load_request,
native_base_url,
require_api_key,
run_update_signal,
)
# ---------------------------------------------------------------------------
# TTS constants
# ---------------------------------------------------------------------------
TTS_GENERATION_PATH = "/services/aigc/multimodal-generation/generation"
_AUDIO_EXTS = {".wav", ".mp3", ".flac", ".ogg", ".pcm", ".aac", ".opus"}
_MODEL_CONFIG_DIR = Path(__file__).resolve().parent.parent / "cdn" / "config"
def _validate_model_config(config: dict[str, Any]) -> bool:
tts = config.get("tts")
if not isinstance(tts, dict):
return False
cosyvoice = config.get("cosyvoice")
cosyvoice_models = (
set(cosyvoice.get("model_voices", {}))
if isinstance(cosyvoice, dict) and isinstance(cosyvoice.get("model_voices"), dict)
else set()
)
voices = tts.get("ws_default_voices")
if not (
isinstance(tts.get("default_model"), str)
and bool(tts["default_model"])
and isinstance(tts.get("default_voice"), str)
and bool(tts["default_voice"])
and isinstance(tts.get("ws_models"), list)
and all(isinstance(item, str) and item for item in tts["ws_models"])
and isinstance(voices, dict)
and all(
isinstance(model, str) and isinstance(voice, str) and voice
for model, voice in voices.items()
)
):
return False
ws_models = set(tts["ws_models"])
return (
tts["default_model"] in ws_models
and ws_models == set(voices)
and ws_models.isdisjoint(cosyvoice_models)
)
def _model_config() -> dict[str, Any]:
return load_cdn_model_config(
"qwencloud-audio-tts-config.json",
local_dir=_MODEL_CONFIG_DIR,
required_keys=("tts",),
validator=_validate_model_config,
)["tts"]
def _default_model() -> str:
return _model_config()["default_model"]
def _default_voice() -> str:
return _model_config()["default_voice"]
def _ws_models() -> frozenset[str]:
return frozenset(_model_config()["ws_models"])
def _ws_default_voices() -> dict[str, str]:
return dict(_model_config()["ws_default_voices"])
# WebSocket endpoint paths
_WS_PATH = "/api-ws/v1/inference"
_WS_HOST_PAYG = "dashscope-intl.aliyuncs.com"
_WS_HOST_TOKEN_PLAN = "token-plan.ap-southeast-1.maas.aliyuncs.com"
def _is_audio_file_output(path: Path) -> bool:
"""Return whether *path* denotes an audio file rather than a directory."""
return not path.is_dir() and path.suffix.lower() in _AUDIO_EXTS
# ---------------------------------------------------------------------------
# WebSocket minimal implementation (stdlib: ssl + socket + struct + hashlib)
# ---------------------------------------------------------------------------
def _ws_connect(host: str, path: str, api_key: str) -> socket.socket:
"""Perform WebSocket HTTP Upgrade handshake, return connected socket."""
ctx = ssl.create_default_context()
raw = socket.create_connection((host, 443), timeout=60)
sock = ctx.wrap_socket(raw, server_hostname=host)
ws_key = hashlib.sha1(uuid_mod.uuid4().bytes).digest()
import base64
ws_key_b64 = base64.b64encode(ws_key[:16]).decode()
handshake = (
f"GET {path} HTTP/1.1\r\n"
f"Host: {host}\r\n"
f"Upgrade: websocket\r\n"
f"Connection: Upgrade\r\n"
f"Sec-WebSocket-Key: {ws_key_b64}\r\n"
f"Sec-WebSocket-Version: 13\r\n"
f"Authorization: Bearer {api_key}\r\n"
f"User-Agent: qwencloud-skills\r\n"
f"\r\n"
)
sock.sendall(handshake.encode())
# Read HTTP response
resp_data = b""
while b"\r\n\r\n" not in resp_data:
chunk = sock.recv(4096)
if not chunk:
raise RuntimeError("WebSocket handshake failed: connection closed")
resp_data += chunk
status_line = resp_data.split(b"\r\n", 1)[0].decode()
if "101" not in status_line:
raise RuntimeError(f"WebSocket handshake failed: {status_line}")
return sock
def _ws_send_text(sock: socket.socket, data: str) -> None:
"""Send a masked text frame over WebSocket."""
payload = data.encode("utf-8")
length = len(payload)
# Build frame header: FIN=1, opcode=1 (text), MASK=1
header = bytearray()
header.append(0x81) # FIN + text opcode
if length < 126:
header.append(0x80 | length)
elif length < 65536:
header.append(0x80 | 126)
header.extend(struct.pack("!H", length))
else:
header.append(0x80 | 127)
header.extend(struct.pack("!Q", length))
# Masking key (4 random bytes)
mask = os.urandom(4)
header.extend(mask)
# Mask the payload
masked = bytearray(length)
for i in range(length):
masked[i] = payload[i] ^ mask[i % 4]
sock.sendall(bytes(header) + bytes(masked))
def _ws_recv_frame(sock: socket.socket) -> tuple[int, bytes]:
"""Receive one complete WebSocket message. Returns (opcode, payload)."""
def _recv_exact(n: int) -> bytes:
buf = b""
while len(buf) < n:
chunk = sock.recv(n - len(buf))
if not chunk:
raise RuntimeError("WebSocket connection closed unexpectedly")
buf += chunk
return buf
def _recv_raw_frame() -> tuple[bool, int, bytes]:
hdr = _recv_exact(2)
fin = bool(hdr[0] & 0x80)
opcode = hdr[0] & 0x0F
masked = bool(hdr[1] & 0x80)
length = hdr[1] & 0x7F
if opcode >= 0x08:
if not fin:
raise RuntimeError("WebSocket protocol error: control frame must not be fragmented")
if length > 125:
raise RuntimeError(
"WebSocket protocol error: control frame payload exceeds 125 bytes"
)
if length == 126:
length = struct.unpack("!H", _recv_exact(2))[0]
elif length == 127:
length = struct.unpack("!Q", _recv_exact(8))[0]
mask_key = _recv_exact(4) if masked else None
payload = _recv_exact(length)
if mask_key:
payload = bytes(payload[i] ^ mask_key[i % 4] for i in range(length))
return fin, opcode, payload
message_opcode: int | None = None
message_chunks: list[bytes] = []
while True:
fin, opcode, payload = _recv_raw_frame()
if opcode == 0x09: # Ping control frame may interrupt a fragmented message
_ws_send_pong(sock, payload)
continue
if opcode == 0x0A: # Pong control frame
continue
if opcode == 0x08: # Close control frame
return opcode, payload
if opcode in (0x01, 0x02):
if message_opcode is not None:
raise RuntimeError("WebSocket protocol error: new data frame before continuation")
if fin:
return opcode, payload
message_opcode = opcode
message_chunks.append(payload)
continue
if opcode == 0x00:
if message_opcode is None:
raise RuntimeError("WebSocket protocol error: unexpected continuation frame")
message_chunks.append(payload)
if fin:
return message_opcode, b"".join(message_chunks)
continue
raise RuntimeError(f"WebSocket protocol error: unsupported opcode {opcode}")
def _ws_close(sock: socket.socket) -> None:
"""Send close frame and shut down."""
try:
# Send close frame: FIN=1, opcode=8, masked, empty payload
mask = os.urandom(4)
sock.sendall(b"\x88\x80" + mask)
except OSError:
pass
try:
sock.shutdown(socket.SHUT_RDWR)
except OSError:
pass
sock.close()
def _ws_send_pong(sock: socket.socket, payload: bytes) -> None:
"""Send a masked pong frame with the given payload."""
pong_hdr = bytearray([0x8A]) # FIN + pong opcode
plen = len(payload)
if plen < 126:
pong_hdr.append(0x80 | plen)
else:
pong_hdr.append(0x80 | 126)
pong_hdr.extend(struct.pack("!H", plen))
mask = os.urandom(4)
pong_hdr.extend(mask)
masked_payload = bytes(payload[i] ^ mask[i % 4] for i in range(plen))
sock.sendall(bytes(pong_hdr) + masked_payload)
def _ws_tts(api_key: str, model: str, text: str, voice: str,
output: Path, *, sample_rate: int = 24000,
audio_format: str = "mp3",
style_params: dict[str, Any] | None = None,
use_model_default_voice: bool = False) -> dict[str, Any]:
"""Perform TTS via WebSocket duplex protocol.
*style_params* may carry optional run-task style parameters
(``volume`` / ``rate`` / ``pitch``); only keys explicitly provided
by the user are forwarded.
Returns dict with audio_file path and metadata.
"""
# Determine endpoint based on key type
if api_key.startswith("sk-sp-"):
host = _WS_HOST_TOKEN_PLAN
else:
host = _WS_HOST_PAYG
# Resolve the model default only when no voice was supplied by the caller.
# An explicit voice named "Cherry" must remain Cherry even though that is
# also the generic CLI default.
ws_default_voices = _ws_default_voices()
if use_model_default_voice and model in ws_default_voices:
voice = ws_default_voices[model]
print(f"Using default voice '{voice}' for model {model}", file=sys.stderr)
task_id = str(uuid_mod.uuid4())
print(f"Connecting to wss://{host}{_WS_PATH} ...", file=sys.stderr)
sock = _ws_connect(host, _WS_PATH, api_key)
try:
# 1. Send run-task
ws_params: dict[str, Any] = {
"text_type": "PlainText",
"format": audio_format,
"sample_rate": sample_rate,
"voice": voice,
}
for k in ("volume", "rate", "pitch"):
if style_params and style_params.get(k) is not None:
ws_params[k] = style_params[k]
run_task_msg = json.dumps({
"header": {
"action": "run-task",
"task_id": task_id,
"streaming": "duplex",
},
"payload": {
"task_group": "audio",
"task": "tts",
"function": "SpeechSynthesizer",
"model": model,
"parameters": ws_params,
"input": {},
},
})
_ws_send_text(sock, run_task_msg)
# 2. Wait for task-started event before sending text (protocol requirement)
task_started = False
error_msg = ""
while not task_started:
opcode, frame_payload = _ws_recv_frame(sock)
if opcode == 0x01: # Text frame
try:
event = json.loads(frame_payload.decode("utf-8"))
except (json.JSONDecodeError, UnicodeDecodeError):
continue
header = event.get("header", {})
event_name = header.get("event", "")
if event_name == "task-started":
task_started = True
elif event_name == "task-failed":
error_msg = header.get("error_message", "Unknown error")
raise RuntimeError(f"WebSocket TTS task-start failed: {error_msg}")
elif opcode == 0x08: # Close frame
raise RuntimeError("WebSocket closed before task-started")
# 3. Send continue-task with text
continue_task_msg = json.dumps({
"header": {
"action": "continue-task",
"task_id": task_id,
"streaming": "duplex",
},
"payload": {
"input": {"text": text},
},
})
_ws_send_text(sock, continue_task_msg)
# 4. Send finish-task
finish_task_msg = json.dumps({
"header": {
"action": "finish-task",
"task_id": task_id,
"streaming": "duplex",
},
"payload": {
"input": {},
},
})
_ws_send_text(sock, finish_task_msg)
# 5. Receive frames: binary = audio data, text = event JSON
audio_chunks: list[bytes] = []
task_completed = False
error_msg = ""
while not task_completed:
opcode, frame_payload = _ws_recv_frame(sock)
if opcode == 0x02: # Binary frame → audio data
audio_chunks.append(frame_payload)
elif opcode == 0x01: # Text frame → event JSON
try:
event = json.loads(frame_payload.decode("utf-8"))
except (json.JSONDecodeError, UnicodeDecodeError):
continue
header = event.get("header", {})
event_name = header.get("event", "")
if event_name == "task-finished":
task_completed = True
elif event_name == "task-failed":
error_msg = header.get("error_message") or "Unknown error"
raise RuntimeError(f"WebSocket TTS failed: {error_msg}")
elif opcode == 0x08: # Close frame
raise RuntimeError("WebSocket TTS closed before task-finished")
finally:
_ws_close(sock)
if error_msg:
raise RuntimeError(f"WebSocket TTS failed: {error_msg}")
if not audio_chunks:
raise RuntimeError("WebSocket TTS: no audio data received")
# Concatenate and write audio. A recognized audio suffix means the caller
# supplied an exact file path; otherwise the path is an output directory.
if _is_audio_file_output(output):
audio_file = output
output_dir = output.parent
else:
output_dir = output
ext = f".{audio_format}" if not audio_format.startswith(".") else audio_format
ts = time.strftime("%Y%m%d-%H%M%S")
audio_file = output_dir / f"audio-{ts}{ext}"
output_dir.mkdir(parents=True, exist_ok=True)
audio_file.write_bytes(b"".join(audio_chunks))
total_bytes = sum(len(c) for c in audio_chunks)
print(f"WebSocket TTS complete: {total_bytes} bytes received", file=sys.stderr)
print(f"Audio saved to {audio_file}", file=sys.stderr)
return {
"audio_file": str(audio_file),
"model": model,
"format": audio_format,
"sample_rate": sample_rate,
"bytes": total_bytes,
}
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _guess_audio_ext(url: str, default: str = ".wav") -> str:
"""Extract audio extension from URL path, ignoring query params."""
path = urlparse(url).path
for ext in (".wav", ".mp3", ".flac", ".ogg", ".pcm", ".aac"):
if path.endswith(ext):
return ext
return default
# ---------------------------------------------------------------------------
# HTTP REST TTS (qwen3-tts-flash, qwen3-tts-instruct-flash)
# ---------------------------------------------------------------------------
def _http_tts(api_key: str, model: str, voice: str, text: str,
request: dict[str, Any], args: Any) -> None:
"""Perform TTS via HTTP REST API."""
input_obj: dict[str, Any] = {"text": text, "voice": voice}
if request.get("language_type"):
input_obj["language_type"] = request["language_type"]
payload: dict[str, Any] = {"model": model, "input": input_obj}
if request.get("instructions") and "instruct" in model.lower():
input_obj["instructions"] = request["instructions"]
if request.get("optimize_instructions") is not None:
input_obj["optimize_instructions"] = request["optimize_instructions"]
url = f"{native_base_url()}{TTS_GENERATION_PATH}"
try:
resp = http_request("POST", url, api_key, payload, timeout=60)
except RuntimeError as e:
print(f"API error: {e}", file=sys.stderr)
sys.exit(1)
audio_url = (resp.get("output") or {}).get("audio", {}).get("url")
if not audio_url:
print(f"Error: No audio URL in response: {resp}", file=sys.stderr)
sys.exit(1)
out = args.output
if out.suffix.lower() in _AUDIO_EXTS:
audio_file = out
out_dir = out.parent
else:
out_dir = out
ext = _guess_audio_ext(audio_url)
ts = time.strftime("%Y%m%d-%H%M%S")
audio_file = out_dir / f"audio-{ts}{ext}"
out_dir.mkdir(parents=True, exist_ok=True)
resp_file = out_dir / "response.json"
resp_file.write_text(json.dumps(resp, ensure_ascii=False, indent=2), encoding="utf-8")
print(f"Response saved to {resp_file}", file=sys.stderr)
try:
download_file(audio_url, audio_file)
except Exception as e:
print(f"Warning: Could not download audio: {e}", file=sys.stderr)
print(f"Audio URL (manual download): {audio_url}", file=sys.stderr)
else:
print(f"Audio saved to {audio_file}", file=sys.stderr)
if args.print_response:
print(json.dumps({"audio_url": audio_url, "audio_file": str(audio_file)}, ensure_ascii=False))
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
def main() -> None:
run_update_signal(caller=__file__)
parser = argparse.ArgumentParser(
description="Synthesize speech from text via Qwen TTS",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""\
request JSON fields (--request / --file):
text (required) Text to synthesize into speech
voice Voice ID — overridden by --voice flag; defaults and
WebSocket model mappings load from CDN configuration
model Model ID — overridden by --model flag
(default loaded from CDN configuration)
language_type Force language: "Auto" / "Chinese" / "English" / "Japanese" /
"Korean" etc. (full names, not "zh"/"en" codes).
HTTP models only — WebSocket models ignore this field
(their language follows the selected voice)
instructions Style instructions for instruct models, e.g.
"Speak slowly with a warm, gentle tone"
optimize_instructions true/false — auto-optimize instruction text
sample_rate Audio sample rate in Hz (WebSocket models, default 24000)
format Audio format: mp3/wav/pcm/opus (WebSocket models, default mp3)
volume Volume [0-100], default 50 (WebSocket models only)
rate Speech rate [0.5-2.0], default 1.0; below 1.0 slows
speech, above speeds up (WebSocket models only)
pitch Pitch multiplier [0.5-2.0], default 1.0; above 1.0 =
higher pitch, below = lower (WebSocket models only)
model examples (current routing/defaults load from CDN configuration):
qwen-audio-3.0-tts-plus Next-gen TTS, higher quality,
WebSocket streaming (PAYG + Token Plan)
qwen3-tts-flash Non-streaming HTTP path — fast, multi-voice
qwen3-tts-instruct-flash Instruction-controlled style (non-streaming HTTP)
Note: CosyVoice models (cosyvoice-v3-plus/flash) require WebSocket API.
Use tts_cosyvoice.py instead of this script.
Token Plan (sk-sp-...) keys: only qwen-audio-3.0-tts-plus is supported.
For basic style control on Token Plan, use volume/rate/pitch with
qwen-audio-3.0-tts-plus (instruct models require a PAYG key).
output:
--output can be a directory (audio saved as audio.wav/mp3 inside)
or a specific file path (e.g. output/speech.mp3). Extension is
auto-detected from the API response URL.
environment variables:
QWENCLOUD_API_KEY (preferred) API key — also loaded from .env
QWEN_API_KEY (fallback) Legacy alias
DASHSCOPE_API_KEY (fallback) Legacy provider variable
QWEN_REGION ap-southeast-1 (default)
examples:
# Simple TTS
python scripts/tts.py --request '{"text":"Hello, world!"}'
# Chinese text with specific voice
python scripts/tts.py --request '{"text":"你好世界"}' --voice Chelsie
# Instruction-controlled style
python scripts/tts.py --request '{"text":"Breaking news today...",
"instructions":"Speak in a serious news anchor tone, moderate pace"}' \\
--model qwen3-tts-instruct-flash
# Save to specific file
python scripts/tts.py --request '{"text":"Hello"}' --output output/hello.mp3
# Style control via WebSocket model (default; works with PAYG + Token Plan keys)
python scripts/tts.py --request '{"text":"落霞与孤鹜齐飞",
"rate":0.8,"pitch":1.1,"volume":60}'
# HTTP fallback model (low cost)
python scripts/tts.py --request '{"text":"Hello"}' --model qwen3-tts-flash
""",
)
parser.add_argument("--request", type=str,
help="Inline JSON: must contain 'text' field")
parser.add_argument("--file", type=Path,
help="Path to JSON file containing request body")
parser.add_argument("--output", type=Path, default=Path("output/qwencloud-audio-tts"),
help="Output directory or file path (default: %(default)s)")
parser.add_argument("--print-response", action="store_true",
help="Print audio URL and file path JSON to stdout")
parser.add_argument("--model", type=str, default=None,
help="Model ID (default loaded from CDN model configuration)")
parser.add_argument("--voice", type=str, default=None,
help="Voice ID (default loaded from CDN model configuration)")
args = parser.parse_args()
try:
request = load_request(args)
except ValueError as e:
print(f"Error: {e}", file=sys.stderr)
sys.exit(1)
text = request.get("text")
if not text:
print("Error: text is required.", file=sys.stderr)
sys.exit(1)
requested_model = args.model if args.model is not None else request.get("model")
if isinstance(requested_model, str) and requested_model.startswith("cosyvoice-"):
print(
f"Error: {requested_model} is not supported by tts.py. "
"Use scripts/tts_cosyvoice.py for CosyVoice models.",
file=sys.stderr,
)
sys.exit(1)
api_key = require_api_key(script_file=__file__, domain="TTS")
try:
# CLI flags intentionally override values loaded from request JSON.
voice_explicit = args.voice is not None or bool(request.get("voice"))
voice = args.voice if args.voice is not None else request.get("voice") or _default_voice()
model = requested_model or _default_model()
check_token_plan_model_support(model, domain="TTS")
use_websocket = model in _ws_models()
except RuntimeError as e:
print(f"Error: {e}", file=sys.stderr)
sys.exit(1)
# --- Route by protocol: WebSocket models vs HTTP models ---
if use_websocket:
# Raw WebSocket path for Qwen-Audio TTS models only.
out = args.output
if _is_audio_file_output(out):
out_dir = out.parent
else:
out_dir = out
sample_rate = request.get("sample_rate", 24000)
audio_format = request.get("format", "mp3")
# Optional style control (WebSocket models only) — forward only if
# explicitly provided by the user. This gives Token Plan users basic
# style control as an alternative to instruct models' instructions.
style_params = {
k: request[k] for k in ("volume", "rate", "pitch")
if request.get(k) is not None
}
try:
result = _ws_tts(
api_key, model, text, voice, out,
sample_rate=sample_rate, audio_format=audio_format,
style_params=style_params,
use_model_default_voice=not voice_explicit,
)
except RuntimeError as e:
print(f"Error: {e}", file=sys.stderr)
sys.exit(1)
# Save response JSON
resp_file = out_dir / "response.json"
resp_file.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
print(f"Response saved to {resp_file}", file=sys.stderr)
if args.print_response:
print(json.dumps(result, ensure_ascii=False))
else:
# HTTP REST path for qwen3-tts-flash, qwen3-tts-instruct-flash
_http_tts(api_key, model, voice, text, request, args)
if __name__ == "__main__":
main()
SKILL.md
---
name: qwencloud-audio-tts
description: "Synthesize speech from text with Qwen TTS models. TRIGGER when: user wants to convert text to speech, create voiceovers, generate audio narration, read text aloud, build TTS applications, mentions speech synthesis/voice generation/audio output from text, or explicitly invokes this skill by name (e.g. use qwencloud-audio-tts). DO NOT TRIGGER when: user wants speech recognition/ASR, text generation without audio, non-Qwen audio tasks."
compatibility: "Requires Python 3.9+; curl is PAYG-only. Cursor: auto-loaded. Claude Code: read this skill's SKILL.md before first use."
---
# Qwen Audio TTS (Text-to-Speech)
Synthesize natural speech from text using Qwen TTS models.
This skill is part of **qwencloud/qwencloud-ai**.
## Skill directory
Use this skill's internal files to execute and learn. Load reference files on demand when the default path fails or you need details.
| Location | Purpose |
|----------|---------|
| `scripts/tts.py` | Qwen TTS / Qwen Audio TTS (HTTP and WebSocket); current compatibility and defaults are in the model catalog below |
| `scripts/tts_cosyvoice.py` | CosyVoice (WebSocket API) — requires `dashscope` SDK; current compatibility and defaults are in the model catalog below |
| `references/cosyvoice-guide.md` | CosyVoice setup, voices, examples, errors |
| `references/execution-guide.md` | Fallback: curl (standard, instruct, streaming), code generation |
| `references/prompt-guide.md` | Text formatting for speech, instructions templates, voice selection |
| `references/api-guide.md` | API supplement |
| `references/sources.md` | Official documentation URLs |
## Security
**NEVER output any API key or credential in plaintext.** Always use variable references (`$QWENCLOUD_API_KEY` in shell, `os.environ["QWENCLOUD_API_KEY"]` in Python). The scripts accept `QWENCLOUD_API_KEY`, then `QWEN_API_KEY`, then `DASHSCOPE_API_KEY`. Any check or detection of credentials must be **non-plaintext**: report only status (e.g. "set" / "not set", "valid" / "invalid"), never the value. Never display contents of `.env` or config files that may contain secrets.
**When the API key is not configured, NEVER ask the user to provide it directly.** Instead, help create a `.env` file with a placeholder (`QWENCLOUD_API_KEY=sk-your-key-here`) and instruct the user to replace it with their actual key from the [QwenCloud Console](https://home.qwencloud.com/api-keys). Only write the actual key value if the user explicitly requests it.
## Key Compatibility
Scripts support both **standard QwenCloud API keys** (`sk-...`) and **Token Plan keys** (`sk-sp-...`). Token Plan keys are automatically routed to the Token Plan endpoint for supported TTS models — see [Token Plan Support](#token-plan-support) below.
**Token Plan: do not use curl; always use the bundled Python scripts.**
Coding Plan keys (also `sk-sp-` prefix but purchased via Coding Plan subscription) cannot be used — TTS models are not available on Coding Plan. The script detects key type at startup and routes accordingly. If qwencloud-ops-auth is installed, see its `references/codingplan.md` for full details.
Detect the API key type without exposing the key:
```bash
python3 -c "
import sys; sys.path.insert(0, 'scripts')
from qwencloud_lib import detect_api_key_type
print(detect_api_key_type('scripts/qwencloud_lib.py'))
"
```
| Output | Meaning |
|--------|---------|
| `token-plan` | Token Plan key detected (`sk-sp-` prefix) |
| `payg` | Standard PAYG key detected |
| `not-set` | No API key found in environment |
For Token Plan, fetch and read the current [Token Plan model catalog](https://alioth-intl.alicdn.com/skills-info/models/references/qwencloud-token-plan-models.md), then use an exact listed model with `scripts/tts.py`. If CDN access fails, use the [local fallback](cdn/references/qwencloud-token-plan-models.md).
## Model Selection
> **🚫 CRITICAL — Never override the user's model choice.** If the user (or the request JSON) explicitly specifies a `model`, you MUST use exactly that model. Do NOT:
> - Replace it with a "better suited" model (e.g., switching to `qwen3-tts-instruct-flash` for stylized/poetic text)
> - Add `instructions` the user did not ask for
> - Modify any parameter the user explicitly provided
>
> Model selection guidance below applies **only when the user has NOT specified a model**.
Before selecting, recommending, or defaulting a model, fetch and read the current [QwenCloud audio TTS model catalog](https://alioth-intl.alicdn.com/skills-info/models/references/qwencloud-audio-tts-models.md). It contains the model list, basic model information, script and voice compatibility, recommendations, and defaults. If CDN access fails, use the [local fallback](cdn/references/qwencloud-audio-tts-models.md).
Consult the **qwencloud-model-selector** skill when model choice depends on capability, scenario, or pricing. CosyVoice requires the `dashscope` SDK and uses different voices; see [cosyvoice-guide.md](references/cosyvoice-guide.md).
> **⚠️ Important**: The model catalog is a **point-in-time snapshot** and may be outdated. Model availability
> changes frequently. **Always check the [official model list](https://www.qwencloud.com/models)
> for the authoritative, up-to-date catalog before making model decisions.**
> **Model details**: For more information about a specific model, direct the user to `https://www.qwencloud.com/models/<model-name>`. Replace `<model-name>` with the exact model ID; never modify or guess it.
> **Dynamic model queries**: If the **qwencloud-model-selector** skill or **QwenCloud CLI** (`qwencloud models info <model>`) is available, use it for real-time model data. CLI requires authentication — see the **qwencloud-usage** skill for login flow.
## Available Voices
Use the QwenCloud audio TTS model catalog linked above for current model-to-voice compatibility and defaults. For complete voice inventories, use the official voice-list links in [sources.md](references/sources.md).
> **Voice defaulting**: When no voice is supplied for a Qwen-Audio WebSocket model, `tts.py` selects that model's configured default voice and prints a notice to stderr. Any explicit voice—including explicit `Cherry`—is passed through unchanged. Fetch the model catalog before choosing a voice.
## Execution
> **⚠️ Multiple artifacts**: When generating multiple files in a single session, you MUST append a numeric suffix to each filename (e.g. `out_1.wav`, `out_2.wav`) to prevent overwrites.
### Qwen TTS (HTTP API) — `tts.py`
#### Prerequisites
- **API Key**: Check `QWENCLOUD_API_KEY`, `QWEN_API_KEY`, then `DASHSCOPE_API_KEY` using a **non-plaintext** check only (e.g. in shell: `[ -n "$QWENCLOUD_API_KEY" ]`; report only "set" or "not set", never the key value). If not set: run the **qwencloud-ops-auth** skill if available; otherwise guide the user to obtain a key from [QwenCloud Console](https://home.qwencloud.com/api-keys) and set it via `.env` file (`echo 'QWENCLOUD_API_KEY=sk-your-key-here' >> .env` in project root or current directory) or environment variable. The script searches for `.env` in the current working directory and the project root. Skills may be installed independently — do not assume qwencloud-ops-auth is present.
**Note**: The script auto-loads `.env` from the current directory and the project root (in addition to any exported environment variable). A shell check showing `$QWENCLOUD_API_KEY` as "not set" does NOT mean the script will fail — it may still find the key in `.env`. Treat the shell check as informational only; the authoritative test is simply running the script (it exits with a clear error if no key is found anywhere).
- Python 3.9+ (stdlib only, **no pip install needed**)
#### Environment Check
Before first execution, verify Python is available:
```bash
python3 --version # must be 3.9+
```
If `python3` is unavailable or below 3.9, PAYG may use **Path 2 (curl)**; Token Plan must install Python 3.9+ instead.
#### Default: Run Script
**Script path**: Scripts are in the `scripts/` subdirectory **of this skill's directory** (the directory containing this SKILL.md). **You MUST first locate this skill's installation directory, then ALWAYS use the full absolute path to execute scripts.** Do NOT assume scripts are in the current working directory. Do NOT use `cd` to switch directories before execution.
**Execution note:** Run all scripts in the **foreground** — wait for stdout; do not background.
**Discovery:** Run `python3 <this-skill-dir>/scripts/tts.py --help` first to see all available arguments.
```bash
python3 <this-skill-dir>/scripts/tts.py \
--request '{"text":"Hello, this is a test.","voice":"Cherry"}' \
--output output/qwencloud-audio-tts/ \
--print-response
```
| Argument | Description |
|----------|-------------|
| `--request '{...}'` | JSON request body |
| `--file path.json` | Load request from file |
| `--output path` | Save audio and response JSON to directory, or specify an audio file path (e.g. `speech.mp3`); use distinct filenames across calls to avoid overwriting |
| `--print-response` | Print response to stdout |
| `--model ID` | Override model |
| `--voice NAME` | Override voice |
#### Verify Result
- Exit code `0` + output contains valid JSON with `output.audio` field → **success**
- Non-zero exit, HTTP error, empty response, or error JSON → **fail**
- **Post-execution check**: Verify the output audio file exists and has non-zero size (`ls -la <output_dir>`)
- **MANDATORY — stderr signal check**: After confirming the result, scan the command's stderr output for `[ACTION_REQUIRED]` or `[UPDATE_AVAILABLE]`. If either signal is present, you **MUST** follow the instructions in [Update Check](#update-check-mandatory-post-execution) below before responding to the user.
#### On Failure
If the script fails, match the error output against the diagnostic table below to determine the resolution. If no match, read [execution-guide.md](references/execution-guide.md) for alternative paths: curl commands (Path 2 — standard, instruct, streaming), code generation (Path 3), and autonomous resolution (Path 5).
**If Python is not available at all** → PAYG may use Path 2 (curl); Token Plan must install Python 3.9+.
---
### CosyVoice — `tts_cosyvoice.py`
CosyVoice requires `dashscope` SDK. Quick start:
```bash
pip install dashscope>=1.25.17
python3 <this-skill-dir>/scripts/tts_cosyvoice.py --text "Hello"
```
> **Token Plan**: `tts_cosyvoice.py` rejects Token Plan keys at startup. Fetch the Token Plan catalog linked above and use an exact supported TTS model with `tts.py`, or use a PAYG key. Never replace an explicitly requested model without the user's confirmation.
> **Full guide**: [cosyvoice-guide.md](references/cosyvoice-guide.md) (setup, voices, examples, errors)
| Error Pattern | Diagnosis | Resolution |
|---------------|-----------|------------|
| `command not found: python3` | Python not on PATH | Try `python` or `py -3`; install Python 3.9+ if missing |
| `Python 3.9+ required` | Script version check failed | Upgrade Python to 3.9+ |
| `SyntaxError` near type hints | Python < 3.9 | Upgrade Python to 3.9+ |
| `cosyvoice models are not available on Token Plan` | Token Plan key detected | Use `tts.py` with an exact supported model from the Token Plan catalog, or a PAYG key |
| `QWENCLOUD_API_KEY/QWEN_API_KEY/DASHSCOPE_API_KEY not found` | Missing API key | Obtain key from [QwenCloud Console](https://home.qwencloud.com/api-keys); add to `.env`: `echo 'QWENCLOUD_API_KEY=sk-...' >> .env`; or run **qwencloud-ops-auth** if available |
| `HTTP 401` | Invalid or mismatched key | Run **qwencloud-ops-auth** (non-plaintext check only); verify key is valid |
| `SSL: CERTIFICATE_VERIFY_FAILED` | SSL cert issue (proxy/corporate) | macOS: run `Install Certificates.command`; else set `SSL_CERT_FILE` env var |
| `URLError` / `ConnectionError` | Network unreachable | Check internet; set `HTTPS_PROXY` if behind proxy |
| `HTTP 429` | Rate limited | Wait and retry with backoff |
| `HTTP 5xx` | Server error | Retry with backoff |
| `PermissionError` | Can't write output | Use `--output` to specify writable directory |
## Quick Reference
### Request Fields
| Field | Type | Description |
|-------|------|-------------|
| `text` | string | **Required** — text to synthesize; check the model catalog above for model-specific limits |
| `voice` | string | Voice ID; check the model catalog above for current defaults and compatibility |
| `model` | string | Model ID; check the model catalog above for the current default |
| `language_type` | string | `Auto`, `Chinese`, `English`, `Japanese`, `Korean`, `French`, `German`, etc. — full names, not codes (`zh`, `en`) |
| `instructions` | string | Tone/style instructions; check the model catalog above for current compatibility and limits |
| `volume` | int | Volume `[0-100]`, default 50 — supported WebSocket models only; check the model catalog |
| `rate` | float | Speech rate `[0.5-2.0]`, default 1.0; below 1.0 slows speech — **WebSocket models only** |
| `pitch` | float | Pitch multiplier `[0.5-2.0]`, default 1.0; above 1.0 raises pitch — **WebSocket models only** |
| `sample_rate` | int | Sample rate in Hz, default 24000 — **WebSocket models only** |
| `format` | string | Audio format: `mp3`/`wav`/`pcm`/`opus`, default `mp3` — **WebSocket models only** |
| `stream` | bool | Direct Qwen3-TTS HTTP API field for SSE streaming. Bundled `tts.py` does not implement this field; use the direct SSE path in [execution-guide.md](references/execution-guide.md) |
### Response Fields
| Field | Description |
|-------|-------------|
| `audio_url` | URL of generated audio (valid 24h) |
| `audio_format` | Format (e.g. wav) |
| `sample_rate` | Sample rate (e.g. 24000) |
| `usage` | Character usage |
## Important Notes
- **text**: Check the model catalog above for the selected model's current request limit.
- **instructions**: Support is model-specific; check the model catalog above before using it.
- **language_type**: `Auto` for mixed language; specify for better pronunciation. Values are full names (`Chinese`, `English`, `Japanese`, ...), not codes (`zh`, `en`). It applies only to compatible HTTP models; the script ignores it for WebSocket models whose language follows the selected voice.
- **audio_url**: Valid for 24 hours — download promptly.
- **Real-time/streaming TTS**: Bundled `tts.py` uses non-streaming HTTP for Qwen3-TTS and raw WebSocket only for Qwen-Audio TTS; CosyVoice is handled by `tts_cosyvoice.py`. The Qwen3-TTS API itself supports optional SSE; use the direct example in [execution-guide.md](references/execution-guide.md) when needed.
## Cross-Skill Chaining
When passing generated audio to another skill (e.g., video-gen audio overlay):
- **Pass `audio_url` directly** — scripts detect URL prefixes and pass through without re-upload
- Use `audio_file` only for local playback or non-API operations
## Error Handling
| Error | Cause | Action |
|-------|-------|--------|
| `401 Unauthorized` | Invalid or missing API key | Run **qwencloud-ops-auth** if available; else prompt user to set key (non-plaintext check only) |
| `400` | Invalid parameters (missing text/voice) | Validate request body |
| `429` / `5xx` | Rate limit or server error | Retry with backoff |
> **Usage & billing**: Use the **qwencloud-usage** skill to check usage, free tier quota, and billing directly. Alternatively, the user can visit the QwenCloud console:
> [Usage Analytics](https://home.qwencloud.com/analytics) |
> [Pay-as-you-go Billing](https://home.qwencloud.com/billing/pay-as-you-go) |
> [Coding Plan Billing](https://home.qwencloud.com/billing/coding-plan)
>
> For the current pricing model list and billing units, fetch the [CDN model-pricing reference](https://alioth-intl.alicdn.com/skills-info/models/references/qwencloud-model-pricing.md). If CDN access fails, use the [local fallback](cdn/references/qwencloud-model-pricing.md).
>
> **NEVER fabricate, guess, or construct usage/billing/console URLs.** Only provide the exact links listed in this skill. If a URL is not listed here, do not invent one.
## Output Location
Prefer the **current working directory**. Default subdirectory: `./output/qwencloud-audio-tts/`.
**Write prohibition**: NEVER write output files into this skill's installation directory or any `skills/` hierarchy. All generated content must go to `output/` under the current working directory or a user-specified path.
## Token Plan Support
Token Plan supports only a subset of TTS models. Fetch and read the current [Token Plan model catalog](https://alioth-intl.alicdn.com/skills-info/models/references/qwencloud-token-plan-models.md) before selecting or validating a model. If CDN access fails, use the [local fallback](cdn/references/qwencloud-token-plan-models.md).
When a Token Plan key is combined with an unsupported model, `tts.py` prints a warning to stderr listing the currently supported models and suggesting alternatives; it does not block the call. If you see this warning, tell the user which models the current catalog supports and offer either a catalog-listed alternative or a PAYG key for the original model. Never silently replace an explicitly requested model.
For CosyVoice specifically, `tts_cosyvoice.py` **hard-blocks** Token Plan keys at startup. Use `tts.py` with a model listed in the current Token Plan catalog, or use a PAYG key.
### Style control on Token Plan
Use only style-control fields supported by the selected Token Plan model in the audio model catalog. The following is a protocol example for a currently supported model; do not substitute it for a user-specified model:
```json
{"text": "落霞与孤鹜齐飞，秋水共长天一色。", "model": "qwen-audio-3.0-tts-plus", "rate": 0.8, "pitch": 1.1, "volume": 60}
```
- `rate` below 1.0 slows speech, above 1.0 speeds it up
- `pitch` above 1.0 raises pitch, below 1.0 lowers it
- `volume` scales linearly (0 = silent, 50 = default, 100 = max)
### Protocol note
The [QwenCloud model page](https://www.qwencloud.com/models/qwen-audio-3.0-tts-plus) documents **WSS (WebSocket)** for `qwen-audio-3.0-tts-plus`. The script includes a WebSocket path branch that:
1. Connects to `wss://token-plan.ap-southeast-1.maas.aliyuncs.com/api-ws/v1/inference` (or standard endpoint for PAYG)
2. Sets `User-Agent: qwencloud-skills` in the WebSocket handshake
3. Sends run-task → continue-task → finish-task messages
4. Receives binary audio frames and concatenates them into the output file
### Required header
`User-Agent: qwencloud-skills` is automatically included in both HTTP and WebSocket requests.
## Update Check (MANDATORY Post-Execution)
After script execution, check stderr for action signals. The script automatically detects update status and emits structured signals. You MUST act on them. This section also applies when the user explicitly asks to check for updates.
### Signal: `[ACTION_REQUIRED] qwencloud-update-check skill is not installed`
When stderr contains this signal, you MUST present the following options to the user and **wait for their response before proceeding**. Do NOT silently skip this.
> The update-check skill is not installed. It keeps you informed about new qwencloud/qwencloud-ai features and bug fixes.
>
> 1) **Install now** — I'll run the install command for you
> 2) **Skip for now** — do nothing this time
> 3) **Never remind me** — permanently suppress this prompt
After the user responds:
- User chooses **1** → run: `npx skills add QwenCloud/qwencloud-ai --skill qwencloud-update-check -y`, then run: `python3 <this-skill-scripts-dir>/gossamer.py --dismiss`
- User chooses **2** → run: `python3 <this-skill-scripts-dir>/gossamer.py --dismiss`, then continue normally
- User chooses **3** → run: `python3 <this-skill-scripts-dir>/gossamer.py --never-install`, then continue normally
### Signal: `[UPDATE_AVAILABLE]`
When stderr contains this signal, you MUST append a brief update notice to your response, including the version info and the update command shown in the stderr output.
### No signal in stderr
If stderr contains neither `[ACTION_REQUIRED]` nor `[UPDATE_AVAILABLE]`, no action is needed — the skill is installed and up to date (or cached within 24h).
### Explicit user request
When the user explicitly asks to check for updates (e.g. "check for updates", "check version"):
1. Look for `qwencloud-update-check/SKILL.md` in sibling skill directories.
2. If found — run: `python3 <qwencloud-update-check-dir>/scripts/check_update.py --print-response` and report the result.
3. If not found — present the install options above.
## References
- [execution-guide.md](references/execution-guide.md) — Fallback paths (curl standard/instruct/streaming, code generation, autonomous)
- [api-guide.md](references/api-guide.md) — API supplementary guide
- [sources.md](references/sources.md) — Official documentation URLs