Responses API Calls - 阿里云百炼 Agent Studio

Applicability

Region restriction: Available only for the China Mainland version (Beijing region)
Application types: Supports agent applications and workflow applications
API Key: Requires a valid DashScope API Key that has been obtained and configured
Application ID: Requires a Bailing application already created with its Application ID retrieved

Prerequisites

Before using the Responses API, ensure the following conditions are met:

API Key: A DashScope API Key has been obtained
Application ID: An agent or workflow application has been created on the Bailing platform, and its Application ID has been retrieved
SDK version: OpenAI Python SDK >= 1.0.0
Python version: Python >= 3.7

Quick Start

import os
from openai import OpenAI

# Configure authentication
api_key = os.getenv("DASHSCOPE_API_KEY")
if not api_key:
    raise ValueError("Please set the DASHSCOPE_API_KEY environment variable")

# Configure application information
app_id = os.getenv("APP_ID")  # Your application ID
if not app_id:
    raise ValueError("Please set the APP_ID environment variable")

# Construct base URL
# Use /agent/ path for agent applications; use /workflow/ path for workflow applications
base_url = f"https://dashscope.aliyuncs.com/api/v2/apps/agent/{app_id}/compatible-mode/v1/"

# Initialize client
client = OpenAI(api_key=api_key, base_url=base_url)
try:
    # Send simple request
    response = client.responses.create(
        input="Hello, please introduce yourself"
    )

    # Retrieve response content
    result = response.output[0].content[0].text
    print(f"Response: {result}")

except Exception as e:
    print(f"Call failed: {e}")

Invocation Modes

Synchronous Invocation

Synchronous invocation is suitable for scenarios requiring immediate responses. The API maintains the connection until the task completes and returns the result. Use cases:

Real-time conversational interactions
Simple query tasks
Operations with predictable response times

Characteristics:

Request blocks until full response is received
Supports streaming output
Suitable for short-duration tasks (recommended < 30 seconds)

from openai import OpenAI
import os

api_key = os.getenv("DASHSCOPE_API_KEY")
app_id = 'APP_ID'  # Replace with your actual application ID
base_url = f'https://dashscope.aliyuncs.com/api/v2/apps/agent/{app_id}/compatible-mode/v1/'

client = OpenAI(api_key=api_key, base_url=base_url)

response = client.responses.create(
    input="Who are you?",
)

# Retrieve text response
print(response.output[0].content[0].text)

Asynchronous Invocation

Asynchronous invocation is suitable for time-consuming tasks. The API immediately returns a task ID, and the final result is retrieved via polling. Use cases:

Generating long documents or reports
Multi-step tool calls
Batch data processing
Complex tasks potentially exceeding timeout limits

Characteristics:

Returns immediately without blocking
Status queried using task ID
Supports cancellation of in-progress tasks
Does not support streaming output

from openai import AsyncOpenAI
import asyncio
import os

api_key = os.getenv("DASHSCOPE_API_KEY")
app_id = 'APP_ID'  # Replace with your actual application ID
base_url = f'https://dashscope.aliyuncs.com/api/v2/apps/agent/{app_id}/compatible-mode/v1/'

client = AsyncOpenAI(api_key=api_key, base_url=base_url)

async def main():
    # Create asynchronous task
    create_response = await client.responses.create(
        input="Please plan a three-day Beijing travel itinerary for me",
        background=True
    )
    task_id = create_response.id
    print(f"Task ID: {task_id}")

    # Poll task status
    while True:
        retrieve_response = await client.responses.retrieve(task_id)
        if retrieve_response.status in ['completed', 'failed', 'cancelled']:
            if retrieve_response.status == 'completed':
                print(retrieve_response.output[0].content[0].text)
            break
        await asyncio.sleep(2)

asyncio.run(main())

Synchronous Invocation Usage

Synchronous invocation is ideal for real-time interactive scenarios where results must be obtained immediately. The API maintains the connection until the task completes.

Workflow Overview

Initialize the client and configure the Application ID
Construct input content (text, image, or file)
Initiate synchronous request
Process response result

Sending Text Messages

from openai import OpenAI
import os

api_key = os.getenv("DASHSCOPE_API_KEY")
app_id = 'APP_ID'  # Replace with your actual application ID
base_url = f'https://dashscope.aliyuncs.com/api/v2/apps/agent/{app_id}/compatible-mode/v1/'

client = OpenAI(api_key=api_key, base_url=base_url)

# Simple string input
response = client.responses.create(
    input="Who are you?",
)

# Retrieve text response
result_text = response.output[0].content[0].text
print(result_text)

Parameter description:

input: Can be a simple string; the SDK automatically converts it to standard format

Sending Multi-turn Conversations

messages = [
    {"role": "user", "content": "Who are you?"},
    {"role": "assistant", "content": "I am an AI assistant."},
    {"role": "user", "content": "What can you do?"}
]

response = client.responses.create(input=messages)
print(response.output[0].content[0].text)

Parameter description:

input: Array of messages containing complete conversation history
Currently requires passing full conversation history with each request; context management via pre_response_id will be supported in future releases

Sending Images

Prerequisites:

Agent applications: Must use Qwen-VL series models, select "Custom Processing" for file handling method, and republish the application
Workflow applications: Must use Qwen-VL series models, set model input parameter variable to imageList in the model node, and republish the application

response = client.responses.create(
    input=[
        {
            "role": "user",
            "content": [
                {"type": "input_text", "text": "What is this?"},
                {
                    "type": "input_image",
                    "image_url": "https://dashscope.oss-cn-beijing.aliyuncs.com/images/dog_and_girl.jpeg"
                }
            ]
        }
    ]
)

print(response.output[0].content[0].text)

Sending Files (Agent Applications Only)

Prerequisites:

Supported only for agent applications
File handling method in the application must be set to "Full-text Reference" or "Chunk Retrieval"

response = client.responses.create(
    input=[
        {
            "role": "user",
            "content": [
                {"type": "input_text", "text": "Summarize the content of this file"},
                {
                    "type": "input_file",
                    "file_url": "https://dashscope.oss-cn-beijing.aliyuncs.com/audios/welcome.mp3"
                }
            ]
        }
    ]
)

print(response.output[0].content[0].text)

Enabling Streaming Output

Prerequisites:

Workflow applications: Must enable the "Streaming Output" toggle in the end node or workflow output node, then republish the application

stream = client.responses.create(
    input="Introduce yourself in no fewer than 100 characters",
    stream=True,
)

# Iterate and process event stream
for chunk in stream:
    if hasattr(chunk, 'delta') and chunk.delta:
        print(chunk.delta, end='', flush=True)

Parameter description:

stream=True: Enables streaming output, returning events in Server-Sent Events (SSE) format
Primary event types: response.output_text.delta (text delta), response.completed (response completion)

Asynchronous Invocation Usage

Asynchronous invocation is suitable for time-consuming tasks (e.g., report generation, multi-step tool calls), avoiding request timeouts through a "submit first, retrieve later" approach.

Workflow Overview

Create asynchronous task (background=True) to obtain task ID
Poll task status
Retrieve result upon task completion
(Optional) Cancel or delete task

Creating Asynchronous Tasks

from openai import AsyncOpenAI
import asyncio
import os

api_key = os.getenv("DASHSCOPE_API_KEY")
app_id = 'APP_ID'  # Replace with your actual application ID
base_url = f'https://dashscope.aliyuncs.com/api/v2/apps/agent/{app_id}/compatible-mode/v1/'

client = AsyncOpenAI(api_key=api_key, base_url=base_url)

async def main():
    create_response = await client.responses.create(
        input="Please plan a three-day Beijing travel itinerary for me, including the Forbidden City and the Great Wall.",
        background=True
    )
    task_id = create_response.id
    print(f"Task ID: {task_id}")
    print(f"Initial status: {create_response.status}")

asyncio.run(main())

Parameter description:

background=True: Enables asynchronous mode; API immediately returns task ID
Streaming output (stream=true) is not currently supported for asynchronous tasks

Querying Task Status

async def poll_task(task_id):
    while True:
        retrieve_response = await client.responses.retrieve(task_id)
        status = retrieve_response.status

        print(f"Current status: {status}")

        # Check if task has reached terminal state
        if status in ['completed', 'failed', 'cancelled']:
            if status == 'completed':
                result_text = retrieve_response.output[0].content[0].text
                print(f"\nTask result:\n{result_text}")
            else:
                print(f"Task status: {status}")
            break

        # Wait 2 seconds before next query
        await asyncio.sleep(2)

Task status description:

queued: Task created and waiting in queue for scheduling
running: Task is currently executing
completed: Task completed successfully; result available in output field
failed: Task execution failed; error details available in output field
cancelled: Task was cancelled by user

Cancelling Tasks

async def cancel_task(task_id):
    cancel_response = await client.responses.cancel(task_id)
    print(f"Cancellation status: {cancel_response.status}")

Limitations:

Only tasks in queued or running state can be cancelled
Tasks already in terminal states (completed, failed, cancelled) cannot be cancelled

Deleting Task Records

async def delete_task(task_id):
    response_wrapper = await client.responses.with_raw_response.delete(task_id)
    response_json = response_wrapper.http_response.json()

    if response_json.get("deleted") is True:
        print("Task record successfully deleted")
    else:
        print("Deletion operation failed")

Limitations:

Only tasks in terminal states (completed, failed, cancelled) can be deleted
This operation is irreversible

API Reference

Request Format

Creating Responses (Synchronous/Asynchronous)

Endpoint: POST /responses Request Parameters:

Parameter	Type	Required	Default	Description
input	string \| array	Yes	-	Input content; supports string or message array
background	boolean	No	false	Whether to use asynchronous mode
stream	boolean	No	false	Whether to enable streaming output (synchronous mode only)
conversation_id	string	No	-	Conversation ID (not supported yet)
pre_response_id	string	No	-	Previous response ID (not supported yet)

Response Format

Synchronous Response

{
    "id": "resp_xxx",
    "status": "completed",
    "output": [
        {
            "content": [
                {
                    "type": "text",
                    "text": "Response content"
                }
            ],
            "role": "assistant"
        }
    ],
    "usage": {
        "prompt_tokens": 10,
        "completion_tokens": 20,
        "total_tokens": 30
    }
}

Asynchronous Response (Task Creation)

{
    "id": "task_xxx",
    "status": "queued"
}

Task Management APIs (Asynchronous Mode)

Querying Task Status

Endpoint: GET /responses/{task_id}

response = await client.responses.retrieve(task_id)

Cancelling Tasks

Endpoint: POST /responses/{task_id}/cancel

cancel_response = await client.responses.cancel(task_id)
print(f"Cancellation status: {cancel_response.status}")

Limitation: Only tasks in queued or running state can be cancelled

Deleting Task Records

Endpoint: DELETE /responses/{task_id}

async def delete_task(task_id):
    response_wrapper = await client.responses.with_raw_response.delete(task_id)
    response_json = response_wrapper.http_response.json()

    if response_json.get("deleted") is True:
        print("Task record successfully deleted")
    else:
        print("Deletion operation failed")

Limitation: Only tasks in terminal states (completed, failed, cancelled) can be deleted

Task Status Reference

Status	Description	Supported Actions
queued	Task created and awaiting execution	Query, Cancel
running	Task currently executing	Query, Cancel
completed	Task completed successfully	Query, Delete
failed	Task execution failed	Query, Delete
cancelled	Task cancelled by user	Query, Delete

​Applicability

​Prerequisites

​Quick Start

​Invocation Modes

​Synchronous Invocation

​Asynchronous Invocation

​Synchronous Invocation Usage

​Workflow Overview

​Sending Text Messages

​Sending Multi-turn Conversations

​Sending Images

​Sending Files (Agent Applications Only)

​Enabling Streaming Output

​Asynchronous Invocation Usage

​Workflow Overview

​Creating Asynchronous Tasks

​Querying Task Status

​Cancelling Tasks

​Deleting Task Records

​API Reference

​Request Format

​Creating Responses (Synchronous/Asynchronous)

​Response Format

​Synchronous Response

​Asynchronous Response (Task Creation)

​Task Management APIs (Asynchronous Mode)

​Querying Task Status

​Cancelling Tasks

​Deleting Task Records

​Task Status Reference

Applicability

Prerequisites

Quick Start

Invocation Modes

Synchronous Invocation

Asynchronous Invocation

Synchronous Invocation Usage

Workflow Overview

Sending Text Messages

Sending Multi-turn Conversations

Sending Images

Sending Files (Agent Applications Only)

Enabling Streaming Output

Asynchronous Invocation Usage

Workflow Overview

Creating Asynchronous Tasks

Querying Task Status

Cancelling Tasks

Deleting Task Records

API Reference

Request Format

Creating Responses (Synchronous/Asynchronous)

Response Format

Synchronous Response

Asynchronous Response (Task Creation)

Task Management APIs (Asynchronous Mode)

Querying Task Status

Cancelling Tasks

Deleting Task Records

Task Status Reference