除控制台操作外，Computer Use Agent（CUA）还为您提供了 OpenAPI，您可以通过 OpenAPI 将任务下发、状态查询等能力嵌入自有业务系统。 1. [获取 AccessKey、SecretKey](https://www.volcengine.com/docs/6291/65568)。 2. 将 AccessKey、SecretKey 配置为环境变量 `VOLC_ACCESSKEY` 、`VOLC_SECRETKEY`。 ```Bash export VOLC_ACCESSKEY="" export VOLC_SECRETKEY="" ``` 3. 在您的本地环境中执行如下命令，安装火山引擎 Python SDK。 ```Python pip install volcengine-python-sdk ``` 4. 参考如下示例代码进行接入。 ```Python 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}") ``` * **调用流程说明** 1. 调用 [CreateCuaNodeNoVNCSession](https://docs.volcengine.com/docs/Cloudphone/create-novnc-pull-stream-session-createcuanodenovncsession) 接口，创建 noVNC 可视化会话。 2. 调用 [RunAgentTaskOneStep](https://docs.volcengine.com/docs/Cloudphone/runagenttaskonestep-1) 接口，传入 UserPrompt（自然语言指令）、目标节点 EcsId 等参数下发任务，返回 RunId。 3. 调用 [ListAgentRunCurrentStep](https://docs.volcengine.com/docs/Cloudphone/query-task-current-steps-listagentruncurrentstep) 接口，通过 RunId 轮询任务当前执行步骤，配合回调地址获取最终结果。 4. 调用 [GetAgentResult](https://docs.volcengine.com/docs/Cloudphone/get-the-task-running-result-getagentresult) 接口，通过 RunId 查看任务执行结果。 * **参数说明** 下表为您汇总了示例代码中需要您自行填写的参数信息。 |参数 |说明 | |---|---| |ecs_id |请填写作为 CUA 设备的云服务器实例 ID，您可以登录 [火山引擎云服务器控制台](https://console.volcengine.com/ecs/region:ecs+cn-beijing/instance?projectName=default) 获取。

![图片](https://portal.volccdn.com/obj/volcfe/cloud-universal-doc/upload_4229b00a658ca3875dea2e2d9be04935.png)

说明

若您还未创建作为 CUA 设备的云服务器，请根据准备资源指引创建实例。

| |tos_bucket |请填写用于结果回流的 TOS 桶名称，您可以登录 [对象存储控制台](https://console.volcengine.com/tos/bucket?projectName=default) 获取。 | |tos_endpoint |请填写用于结果回流的 TOS 桶 Endpoint，您可以登录 [对象存储控制台](https://console.volcengine.com/tos/bucket?projectName=default) 获取。 | |tos_region |请填写用于结果回流的 TOS 桶所属地域的 Region ID，您可以登录 [对象存储控制台](https://console.volcengine.com/tos/bucket?projectName=default) 查看其所属地域，并参考 [地域和访问域名（Endpoint）](https://www.volcengine.com/docs/6349/107356) 获取该地域的 Region ID。 |

云手机