除控制台操作外，Computer Use Agent（CUA）还为您提供了 OpenAPI，您可以通过 OpenAPI 将任务下发、状态查询等能力嵌入自有业务系统。

获取 AccessKey、SecretKey。

将 AccessKey、SecretKey 配置为环境变量 VOLC_ACCESSKEY 、VOLC_SECRETKEY。

export VOLC_ACCESSKEY="<Your-Access-Key-ID>"
export VOLC_SECRETKEY="<Your-Secret-Access-Key>"

在您的本地环境中执行如下命令，安装火山引擎 Python SDK。

pip install volcengine-python-sdk

参考如下示例代码进行接入。

import os
import time
import uuid
import volcenginesdkcore
from volcenginesdkcore.rest import ApiException

if __name__ == '__main__':
# 从环境变量获取 AK/SK
ak = os.environ.get("VOLC_ACCESSKEY")
sk = os.environ.get("VOLC_SECRETKEY")

# 配置 SDK
configuration = volcenginesdkcore.Configuration()
configuration.ak = ak
configuration.sk = sk
configuration.region = "cn-beijing"

# 创建 API 客户端
api_instance = volcenginesdkcore.UniversalApi(
volcenginesdkcore.ApiClient(configuration)
)

# -----------------------------
# 请替换为实际参数
# -----------------------------
ecs_id = "" # CUA 节点对应的 ECS 实例 ID
tos_bucket = "" # TOS 桶名称
tos_endpoint = "" # TOS endpoint，例如：https://tos-cn-beijing.volces.com
tos_region = "" # TOS region，例如：cn-beijing

user_prompt = "打开浏览器并搜索火山引擎"

try:
# =========================================================
# 1. 创建 noVNC 拉流会话
# 接口：CreateCuaNodeNoVNCSession
# 服务：cua 版本：2024-01-01
# =========================================================
novnc_body = {}
if ecs_id:
novnc_body["Ecsid"] = ecs_id

if not novnc_body:
raise ValueError("Id 和 Ecsid 需要二选一传入，不能都为空")

novnc_resp = api_instance.do_call(
volcenginesdkcore.UniversalInfo(
method="POST",
action="CreateCuaNodeNoVNCSession",
service="cua",
version="2024-01-01",
content_type="application/json"
),
volcenginesdkcore.Flatten(novnc_body).flat()
)

print("\nCreateCuaNodeNoVNCSession response:")
print(novnc_resp)

novnc_result = novnc_resp.get("Result", novnc_resp)
print("\nnoVNC 会话信息：")
print("ViewerUrl :", novnc_result.get("ViewerUrl"))
print("WebsocketUrl:", novnc_result.get("WebsocketUrl"))
print("ExpiresAt :", novnc_result.get("ExpiresAt"))

# =========================================================
# 2. 运行 Agent 任务
# 接口：RunAgentTaskOneStep
# 服务：ipaas 版本：2023-08-01
# =========================================================
run_body = volcenginesdkcore.Flatten({
"RunName": "computer-use 测试 demo",
"ThreadId": str(uuid.uuid4()),
"AgentType": "cua",
"Ecsid": ecs_id,
"UserPrompt": user_prompt,
"TosBucket": tos_bucket,
"TosEndpoint": tos_endpoint,
"TosRegion": tos_region
}).flat()

run_resp = api_instance.do_call(
volcenginesdkcore.UniversalInfo(
method="POST",
action="RunAgentTaskOneStep",
service="ipaas",
version="2023-08-01",
content_type="application/json"
),
run_body
)

print("\nRunAgentTaskOneStep response:")
print(run_resp)

run_info = run_resp.get("Result", run_resp)
run_id = run_info["RunId"]
print("\nRunId:", run_id)

# =========================================================
# 3. 轮询 Agent 当前步骤
# 接口：ListAgentRunCurrentStep
# 服务：ipaas 版本：2023-08-01
# 状态 3/5/6/7 表示任务结束
# =========================================================
time.sleep(5) # 等待任务开始执行

while True:
step_resp = api_instance.do_call(
volcenginesdkcore.UniversalInfo(
method="GET",
action="ListAgentRunCurrentStep",
service="ipaas",
version="2023-08-01",
content_type="application/json"
),
volcenginesdkcore.Flatten({"RunId": run_id}).flat()
)

print("\nListAgentRunCurrentStep response:")
print(step_resp)

step_info = step_resp.get("Result", step_resp)
if step_info.get("Status") in [3, 5, 6, 7]:
print("\n任务已结束，最终状态:", step_info.get("Status"))
break

time.sleep(2) # 继续轮询

# =========================================================
# 4. 获取 Agent 任务结果
# 接口：GetAgentResult
# 服务：ipaas 版本：2023-08-01
# =========================================================
result_resp = api_instance.do_call(
volcenginesdkcore.UniversalInfo(
method="GET",
action="GetAgentResult",
service="ipaas",
version="2023-08-01",
content_type="application/json"
),
volcenginesdkcore.Flatten({"RunId": run_id}).flat()
)

print("\nGetAgentResult response:")
print(result_resp)

except ApiException as e:
print(f"API Exception: {e}")
except Exception as e:
print(f"General Exception: {e}")

调用流程说明

调用 CreateCuaNodeNoVNCSession 接口，创建 noVNC 可视化会话。

调用 RunAgentTaskOneStep 接口，传入 UserPrompt（自然语言指令）、目标节点 EcsId 等参数下发任务，返回 RunId。

调用 ListAgentRunCurrentStep 接口，通过 RunId 轮询任务当前执行步骤，配合回调地址获取最终结果。

调用 GetAgentResult 接口，通过 RunId 查看任务执行结果。

参数说明

下表为您汇总了示例代码中需要您自行填写的参数信息。

参数	说明
ecs_id	请填写作为 CUA 设备的云服务器实例 ID，您可以登录火山引擎云服务器控制台获取。说明若您还未创建作为 CUA 设备的云服务器，请根据准备资源指引创建实例。
tos_bucket	请填写用于结果回流的 TOS 桶名称，您可以登录对象存储控制台获取。
tos_endpoint	请填写用于结果回流的 TOS 桶 Endpoint，您可以登录对象存储控制台获取。
tos_region	请填写用于结果回流的 TOS 桶所属地域的 Region ID，您可以登录对象存储控制台查看其所属地域，并参考地域和访问域名（Endpoint）获取该地域的 Region ID。

云手机