Skip to main content

Asynchronous invocation API

This topic describes how to make an asynchronous call to Alibaba Cloud Model Studio applications ( Agent and Workflow ) using the OpenAI-compatible Responses API. For long-running tasks, set the background parameter to true in your request. The API then immediately returns a task ID for querying and managing the task. This "submit first, query later" pattern helps avoid request timeouts and long waits.

References
This topic is applicable only to the China (Beijing) region.
Asynchronous tasks do not support streaming output (stream=true).

Prerequisites

Quick start

This section provides complete Python and Java examples to demonstrate how to start an asynchronous task, poll for its status until it is complete, and retrieve the final result. This example covers the core workflow for asynchronous invocation:
  1. Create a task: Call the create method and set background=True to get the task ID.
  2. Poll for status: Periodically call the retrieve method to query the task status.
  3. Process the result: When the task status is completed, failed, or cancelled, exit the loop and process the final result.

Code example

Python
import asyncio
import os
from openai import AsyncOpenAI
import time

# 1. Initialize the client
# If the environment variable is not set, you can replace the following line with your Model Studio API key: api_key="sk-xxx".
# To prevent API key exposure, avoid hardcoding the key in production code.
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():
    start_time = time.time()

    # 2. Start an asynchronous task
    print("Step 1: Creating an asynchronous task...")

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

    # 3. Poll for the task status until the task is complete
    print("\nStep 2: Polling for task status (every 2 seconds)...")
    while True:
        retrieve_response = await client.responses.retrieve(task_id)
        status = retrieve_response.status

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

        # Check if the task has reached a terminal state
        if status in ['completed', 'failed', 'cancelled']:
            print("\nStep 3: The task is complete or has been terminated.")
            final_response = retrieve_response
            break

        # Wait for 2 seconds before the next poll
        await asyncio.sleep(2)

    # 4. Process and print the final result
    if final_response.status == 'completed':
        print("  - Status: Success")
        # Extract and print the text result
        if final_response.output and final_response.output[0].content:
            result_text = final_response.output[0].content[0].text
            print("\n--- Task Output ---")
            print(result_text)
            print("-----------------")
        else:
            print("The task is complete, but no valid output was returned.")

    else: # 'failed' or 'cancelled'
        print(f"  - Status: {final_response.status.upper()}")
        print("\n--- Final Response Details ---")
        # Print the complete JSON for debugging
        print(final_response.model_dump_json(indent=2))
        print("-----------------------")

    end_time = time.time()
    print(f"\nTotal time elapsed: {end_time - start_time:.2f} seconds")

if __name__ == "__main__":
    asyncio.run(main())

Workflow

The following sections describe the API operations for creating, querying, canceling, and deleting asynchronous tasks.

Create a task

Set the background parameter to true to enable asynchronous mode. This creates an asynchronous task and immediately returns a task ID. Endpoint:POST https://dashscope.aliyuncs.com/api/v2/apps/agent/{APP_ID}/compatible-mode/v1/responses
Replace {APP_ID} with your application ID.

Request example

  • Single-turn conversation
  • Parameter passing
Python
from openai import AsyncOpenAI
import asyncio
import os

# If the environment variable is not configured, replace the following line with your Model Studio API key: api_key="sk-xxx".
# However, to avoid security risks, we recommend not hard-coding the API key in production code.
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/'

# Initialize the asynchronous client
client = AsyncOpenAI(
  api_key=api_key,
  base_url=base_url
)

async def main():
  # Asynchronous call
  response = await client.responses.create(
    input="Who are you?",
    background=True
  )

  # Print the complete response JSON object
  print(response.model_dump_json(indent=2))

# Run the asynchronous main function
if __name__ == "__main__":
  asyncio.run(main())

Request fields

Parameter

Type

Required

Description

input

string/array

Yes

The core input for the request, which can be a single string or an array of messages representing a multi-turn conversation history.

background

boolean

Yes

Specifies whether to execute the task asynchronously.

  • false (default): Uses synchronous mode. The API keeps the connection open until the task is complete.

  • true: Enables asynchronous mode. The API immediately returns a task ID, which can be used to check the task status and retrieve the result.

extra_body

object

No

A field for additional parameters.

extra_body.biz_params

object

No

Used to pass parameters defined in the application, such as custom variables, node inputs, or plugin arguments.

user_defined_params

object

No

Custom plugin parameter information.

An application supports up to 10 unique plugins.

user_defined_tokens

object

No

User-level authentication information for a custom plugin.

An application supports up to 10 unique plugins.

user_defined_tokens.user_token

string

No

The user authentication information required by the plugin.

Response example

{
  "id": "bcb9728b-a7f8-480c-ace0-8d61ff776857",
  "created_at": 1761875532,
  "error": null,
  "incomplete_details": null,
  "instructions": null,
  "metadata": null,
  "model": "",
  "object": "response",
  "output": [],
  "parallel_tool_calls": false,
  "temperature": null,
  "tool_choice": "auto",
  "tools": [],
  "top_p": null,
  "background": null,
  "conversation": null,
  "max_output_tokens": null,
  "max_tool_calls": null,
  "previous_response_id": null,
  "prompt": null,
  "prompt_cache_key": null,
  "reasoning": null,
  "safety_identifier": null,
  "service_tier": null,
  "status": "queued",
  "text": null,
  "top_logprobs": null,
  "truncation": null,
  "usage": null,
  "user": null
}

Response fields

Parameter

Type

Description

id

string

A unique identifier for the asynchronous task. Use it to retrieve, cancel, or delete the task.

status

string

The initial status of the task, which is typically queued.

created_at

integer

The task's creation time, represented as a Unix timestamp in seconds.

object

string

The object type. This value is always response.

Retrieve task

Retrieves the status and result of a specified asynchronous task. Endpoint:GET https://dashscope.aliyuncs.com/api/v2/apps/agent/{APP_ID}/compatible-mode/v1/responses/{RESPONSE_ID}
Replace {APP_ID} with your App ID and {RESPONSE_ID} with the Response ID from the task creation response.

Request example

Python
import asyncio
import os
from openai import AsyncOpenAI

# If the environment variable is not set, you can provide your DashScope API Key directly: api_key="sk-xxx".
# For security, avoid hard-coding the key in production environments.
api_key=os.getenv("DASHSCOPE_API_KEY")
app_id='APP_ID' # 1. Replace with your App ID
base_url = f'https://dashscope.aliyuncs.com/api/v2/apps/agent/{app_id}/compatible-mode/v1/'

# Initialize the client
client = AsyncOpenAI(api_key=api_key, base_url=base_url)

async def main():
  # Query task status
  response = await client.responses.retrieve("6e4ba287-0e93-4d67-9955-ef1017ed3384") # 2. Replace with your Response ID
  print(response.model_dump_json(indent=2))

if __name__ == "__main__":
  asyncio.run(main())

Response example

{
  "id": "bcb9728b-a7f8-480c-ace0-8d61ff776857",
  "created_at": 1761875537,
  "error": null,
  "incomplete_details": null,
  "instructions": null,
  "metadata": null,
  "model": "",
  "object": "response",
  "output": [
    {
      "id": "msg_d89e8cb9-f8fa-4092-bb86-928e21839f64",
      "content": [
        {
          "annotations": [],
          "text": "Hello, I am Qwen, a large-scale language model independently developed by the Qwen Lab at Alibaba Group. I can answer questions, create text such as stories, official documents, emails, and scripts, and also perform logical reasoning, programming, and even express opinions and play games. My goal is to be your trustworthy intelligent assistant. Is there anything I can help you with?",
          "type": "output_text",
          "logprobs": null
        }
      ],
      "role": "assistant",
      "status": "completed",
      "type": "message"
    }
  ],
  "parallel_tool_calls": false,
  "temperature": null,
  "tool_choice": "auto",
  "tools": [],
  "top_p": null,
  "background": null,
  "conversation": null,
  "max_output_tokens": null,
  "max_tool_calls": null,
  "previous_response_id": null,
  "prompt": null,
  "prompt_cache_key": null,
  "reasoning": null,
  "safety_identifier": null,
  "service_tier": null,
  "status": "completed",
  "text": null,
  "top_logprobs": null,
  "truncation": null,
  "usage": null,
  "user": null
}

Response fields

Parameter

Type

Description

id

string

The unique identifier for the asynchronous task.

status

string

The current status of the asynchronous task. For details, see Task Lifecycle.

output

array

The output of the task. When status is completed , this field contains the final result.

Cancel a task

Cancels an asynchronous task in the queued or running state. Endpoint:POST https://dashscope.aliyuncs.com/api/v2/apps/agent/{APP_ID}/compatible-mode/v1/responses/{RESPONSE_ID}/cancel
Replace {APP_ID} with your application ID and {RESPONSE_ID} with the task ID returned on task creation.

Request example

import asyncio
import os
from openai import AsyncOpenAI

# If the environment variable is not set, replace the following line with your DashScope API key: api_key="sk-xxx".
# However, we do not recommend hardcoding the API key in a production environment to reduce security risks.
api_key=os.getenv("DASHSCOPE_API_KEY")
app_id='APP_ID' # 1. Replace with your application ID
base_url = f'https://dashscope.aliyuncs.com/api/v2/apps/agent/{app_id}/compatible-mode/v1/'

# Initialize the client
client = AsyncOpenAI(api_key=api_key, base_url=base_url)

async def main():
  # Cancel the task
  response = await client.responses.cancel("6e4ba287-0e93-4d67-9955-ef1017ed3384") # 2. Replace with the task ID
  print(response.model_dump_json(indent=2))

if __name__ == "__main__":
  asyncio.run(main())

Response example

{
    "created_at": 1761877521,
    "id": "568b1459-126d-4719-b58f-b41078bacb48",
    "model": "",
    "object": "response",
    "output": [

    ],
    "parallel_tool_calls": false,
    "status": "cancelled",
    "tool_choice": "auto",
    "tools": [

    ]
}

Response fields

Field

Type

Description

id

string

Unique identifier of the task.

status

string

The status of the task after cancellation. For example:

cancelled: The task was successfully canceled.

failed: The cancellation failed because the task was already in a final state, such as completed or failed.

object

string

The object type, always response.

Delete a response

Deletes a response that is in a final state (completed, failed, or cancelled). This operation is irreversible. Endpoint:DELETE https://dashscope.aliyuncs.com/api/v2/apps/agent/{APP_ID}/compatible-mode/v1/responses/{RESPONSE_ID}
Replace {APP_ID} with your application ID and {RESPONSE_ID} with the response ID returned when the asynchronous task was created.

Request example

Python
import asyncio
import os
from openai import AsyncOpenAI

# If the environment variable is not configured, you can replace the following line with your Model Studio API key: api_key="sk-xxx".
# We do not recommend hard-coding the API key in your code in a production environment to reduce the risk of API key exposure.
api_key = os.getenv("DASHSCOPE_API_KEY")
app_id = 'APP_ID'  # 1. Replace with your 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():
    # 2. Replace with the response ID to delete
    response_id_to_delete = "b17c64f94-7034-403e-aef5-4f48647656d5"
    try:
        response_wrapper = await client.responses.with_raw_response.delete(response_id_to_delete)
        http_response = response_wrapper.http_response
        response_json = http_response.json()

        # Check if the "deleted" field in the response is true.
        if response_json.get("deleted") is True:
            print("The response was successfully deleted.")
            print(response_json)
        else:
            print("The deletion failed.")
            print(response_json)

    except Exception as e:
        print(f"\nAn error occurred while calling the API: {e}")

if __name__ == "__main__":
    asyncio.run(main())

Response example

{
    "deleted": true,
    "id": "c8eddc0e-faf5-4228-a853-9c411194015f",
    "object": "response"
}

Response fields

Field

Type

Description

id

string

The unique identifier of the deleted response.

deleted

boolean

Set to true if the response was deleted successfully.

object

string

The object type. This value is always response.

Task lifecycle

An asynchronous task has the following states:

State

Description

queued

The task is created and awaits scheduling in the queue.

running

The task is running.

completed

The task completed successfully. The output field of the response contains the result.

cancelled

The user cancelled the task before it completed.

failed

The task failed to execute. The output field of the response contains the error message.

Error codes

If an application call fails, see Error codes to resolve the error.
Managed Agent API
Flow Agent API
Sandbox API
RAG API
Long-term Memory
ParseX API
Security API
Framework Integration