DashScope Python SDK 1.x レガシー使用方法リファレンス

# 1.x: globals
import dashscope
dashscope.api_key = os.getenv("DASHSCOPE_API_KEY")
dashscope.base_http_api_url = "https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com/api/v1"

# 2.0: client construction
import dashscope
client = dashscope.Client(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    workspace="{WorkspaceId}",
    region="cn-beijing",
)

# 1.x
from dashscope import Generation
response = Generation.call(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    model="qwen-plus",
    messages=messages,
    result_format="message",
)
if response.status_code == 200:
    print(response.output.choices[0].message.content)
else:
    print(f"HTTP status: {response.status_code}")
    print(f"Error code: {response.code}")
    print(f"Error message: {response.message}")

# 2.0
from dashscope.v2 import DashScopeError
try:
    response = client.chat.completions.create(
        model="qwen-plus",
        messages=messages,
    )
    print(response.choices[0].message.content)
except DashScopeError as e:
    print(f"HTTP status: {e.status_code}")
    print(f"Error code: {e.code}")
    print(f"Error message: {e.message}")

# 1.x
from dashscope import MultiModalConversation
response = MultiModalConversation.call(
    api_key=os.getenv('DASHSCOPE_API_KEY'),
    model='qwen3-vl-plus',
    messages=[{
        "role": "user",
        "content": [
            {"image": "https://example.com/image.png"},
            {"text": "What is in this picture?"},
        ],
    }])
print(response.output.choices[0].message.content[0]['text'])

# 2.0: content parts shape kept, flat access
response = client.multimodal.chat.create(
    model='qwen3-vl-plus',
    messages=[{
        "role": "user",
        "content": [
            {"image": "https://example.com/image.png"},
            {"text": "What is in this picture?"},
        ],
    }])
print(response.output_text)

# 1.x: synchronous call (blocks until done)
from dashscope import ImageSynthesis
rsp = ImageSynthesis.call(api_key=os.getenv("DASHSCOPE_API_KEY"),
                          model=ImageSynthesis.Models.wanx_v1,
                          prompt=prompt,
                          n=1,
                          style='<watercolor>',
                          size='1024*1024')
if rsp.status_code == HTTPStatus.OK:
    for result in rsp.output.results:
        print(result.url)
else:
    print('Failed, status_code: %s, code: %s, message: %s' %
          (rsp.status_code, rsp.code, rsp.message))

# 2.0: generate one-shot (submit and wait); enum classes become strings
job = client.images.generate(
    model="wanx2.1-t2i-turbo",
    prompt=prompt,
    n=1,
    size='1024*1024',
)
for url in job.image_urls:
    print(url)

# 1.x
from dashscope import VideoSynthesis
rsp = VideoSynthesis.call(api_key=api_key,
                          model='wan3.0-video',
                          prompt=prompt,
                          resolution="720P",
                          ratio="16:9",
                          duration=15,
                          watermark=True)
if rsp.status_code == HTTPStatus.OK:
    print("video_url:", rsp.output.video_url)

# 2.0
job = client.videos.generate(
    model='wan3.0-video',
    prompt=prompt,
    resolution="720P",
    ratio="16:9",
    duration=15,
    watermark=True,
)
print("video_url:", job.video_url)

# 1.x: async_call submit + wait
from dashscope.audio.asr import Transcription
task_response = Transcription.async_call(
    model='paraformer-v2',
    file_urls=['{YOUR_AUDIO_URL}'],
    language_hints=['zh', 'en'],
)
transcribe_response = Transcription.wait(task=task_response.output.task_id)
if transcribe_response.status_code == HTTPStatus.OK:
    print(json.dumps(transcribe_response.output, indent=4, ensure_ascii=False))

# 2.0: transcribe one-shot; non-signature parameters go via parameters
job = client.audio.asr.transcribe(
    ['{YOUR_AUDIO_URL}'],
    model='paraformer-v2',
    parameters={'language_hints': ['zh', 'en']},
)
print(json.dumps(job.model_dump(), indent=4, ensure_ascii=False))

# 1.x (callback style, excerpt)
from dashscope.audio.asr import Recognition
class Callback(RecognitionCallback):
    def on_result(self, result): ...
recognition = Recognition(model='paraformer-realtime-v2', callback=Callback())
recognition.start()
recognition.send_audio_frame(chunk)
recognition.stop()

# 2.0 (event iteration, send and receive concurrently)
from concurrent.futures import ThreadPoolExecutor

def receive_events(rec):
    for event in rec.events():
        print(event)

with client.audio.asr.recognize(
        model='paraformer-realtime-v2',
        format='pcm',
        sample_rate=16000,
) as recognition:
    with ThreadPoolExecutor(max_workers=1) as pool:
        receiver = pool.submit(receive_events, recognition)
        for chunk in iter_pcm_chunks():  # real-time PCM frames provided by the application
            recognition.send_audio_frame(chunk)
        recognition.stop()
        receiver.result()

# 1.x (callback style, excerpt)
from dashscope.audio.tts_v2 import SpeechSynthesizer, ResultCallback
class Callback(ResultCallback):
    def on_data(self, data: bytes): audio.write(data)
synthesizer = SpeechSynthesizer(
    model="cosyvoice-v3-flash",
    voice="longanyang",
    callback=Callback(),
)
synthesizer.streaming_call(text)
synthesizer.streaming_complete()

# 2.0: connect session (segmented input + segmented audio)
with client.audio.tts.connect(
        model="cosyvoice-v3-flash",
        voice="longanyang",
        format="mp3",
) as session:
    session.send_text(text)
    session.flush()
    session.stop()  # End the task and wait for remaining audio
    for chunk in session.audio_chunks():
        audio.write(chunk)

# 1.x
resp = dashscope.TextEmbedding.call(
    model="qwen3.7-text-embedding",
    input="The quality of this clothing is excellent.",
)
if resp.status_code == HTTPStatus.OK:
    print(resp)

# 2.0
resp = client.embeddings.create(
    model="qwen3.7-text-embedding",
    input="The quality of this clothing is excellent.",
)
for item in resp.embeddings:
    print(item.index, len(item.embedding))

# 1.x
from dashscope import Application
response = Application.call(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    app_id='APP_ID',
    prompt='Who are you?')
if response.status_code != HTTPStatus.OK:
    print(f'code={response.status_code}, message={response.message}')
else:
    print(response.output.text)

# 2.0
from dashscope.v2 import DashScopeError
try:
    response = client.apps.completions.create(
        app_id='APP_ID',
        prompt='Who are you?')
    print(response.text)
except DashScopeError as e:
    print(f'code={e.code}, message={e.message}')