#!/usr/bin/env python3
"""QI GUI Agent —— 客户接入最小实现（单文件 · 仅标准库）。
接入模型：客户端负责「截图 + 执行」，QI 负责「决策」。
每一轮把当前截图和任务发给 QI，QI 返回一个动作（点击/滑动/输入…），
客户端在设备上执行后，带着新截图请求下一轮，直到 QI 返回 terminate。
┌─────────┐ session_id + query + screenshot ┌─────────┐
│ 客户端 │ ─────────────────────────────────▶ │ QI │
│(截图/执行)│ ◀───────────────────────────────── │ (决策) │
└─────────┘ action: click/swipe/... └─────────┘
运行：
export QI_AK_ID=your_ak_id
export QI_AK_SECRET=your_ak_secret
export GUI_API_KEY=qis_xxx
export GUI_INSTANCE_ID=GUI_xxx
python qi_min_demo.py --device emulator-5554 --task "打开设置，进入 WiFi 页面" --app 设置
单轮测试（无需真机，只发一张图看 QI 返回什么动作）：
python qi_min_demo.py --screenshot screen.png --task "打开设置"
"""
from __future__ import annotations
import argparse
import base64
import json
import hashlib
import hmac
from datetime import datetime, timezone
import os
import re
import subprocess
import time
import http.client
import urllib.parse
import uuid
ENDPOINT = os.environ.get("QI_ENDPOINT", "https://maasqiservice.cn-beijing.aliyuncs.com/gui/v1/chat/completions")
ACS_ACTION = os.environ.get("QI_ACS_ACTION", "GuiChatCompletionStream")
ACS_VERSION = os.environ.get("QI_ACS_VERSION", "2026-08-31")
# open 动作的 App 名 -> 包名映射（示例，客户按需补充, 不在表内则执行报错）。
APP_MAP = {
"设置": "com.android.settings",
"微信": "com.tencent.mm",
"淘宝": "com.taobao.taobao",
}
def validate_config() -> None:
"""只报告缺失变量名，不输出凭据；在截图或设备操作前检查。"""
required = ("QI_AK_ID", "QI_AK_SECRET", "GUI_API_KEY", "GUI_INSTANCE_ID")
missing = [name for name in required if not os.environ.get(name, "").strip()]
if missing:
raise ValueError("缺少必填环境变量：" + ", ".join(missing))
endpoint = urllib.parse.urlsplit(ENDPOINT)
if (endpoint.scheme != "https" or not endpoint.hostname or endpoint.username is not None
or endpoint.password is not None or endpoint.query or endpoint.fragment):
raise ValueError("QI_ENDPOINT 必须是无账号、query 和 fragment 的 HTTPS URL")
def sign_headers(body: bytes, session_id: str, request_id: str, *, stream: bool = False) -> dict:
"""ACS3 签名：body 必须与实际发送字节完全一致。会话 ID 复用，请求 ID 每轮更新。"""
validate_config()
endpoint = urllib.parse.urlsplit(ENDPOINT)
body_hash = hashlib.sha256(body).hexdigest()
headers = {
"host": endpoint.netloc,
"content-type": "application/json; charset=utf-8",
"accept": "text/event-stream" if stream else "application/json",
"x-acs-action": ACS_ACTION,
"x-acs-version": ACS_VERSION,
"x-acs-date": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
"x-acs-signature-nonce": str(uuid.uuid4()),
"x-acs-content-sha256": body_hash,
"X-QI-Agent-Api-Key": os.environ["GUI_API_KEY"],
"X-QI-Instance-Id": os.environ["GUI_INSTANCE_ID"],
"X-QI-Session-Id": session_id,
"X-QI-Request-Id": request_id,
}
token = os.environ.get("QI_STS_TOKEN", "")
if token:
headers["x-acs-security-token"] = token
signed = dict(sorted((k.lower(), v.strip()) for k, v in headers.items()
if k.lower() in ("host", "content-type") or k.lower().startswith("x-acs-")))
names = ";".join(signed)
canonical = "\n".join([
"POST", endpoint.path or "/", "",
"".join(f"{k}:{v}\n" for k, v in signed.items()), names, body_hash,
])
to_sign = "ACS3-HMAC-SHA256\n" + hashlib.sha256(canonical.encode("utf-8")).hexdigest()
signature = hmac.new(os.environ["QI_AK_SECRET"].encode("utf-8"),
to_sign.encode("utf-8"), hashlib.sha256).hexdigest()
headers["Authorization"] = (
f"ACS3-HMAC-SHA256 Credential={os.environ['QI_AK_ID']},"
f"SignedHeaders={names},Signature={signature}"
)
return headers
def resolve_package(name: str) -> str:
"""App 名解析成包名。"""
assert name in APP_MAP, f"未知的 App 名：{name}"
return APP_MAP[name]
# ======================================================================
# 1. QI 客户端：一次请求 = 一轮决策
# ======================================================================
def to_data_url(screenshot: str | bytes) -> str:
"""把截图（PNG/JPEG 字节或本地路径）转成 data URL。"""
data = screenshot if isinstance(screenshot, bytes) else open(screenshot, "rb").read()
mime = "image/png" if data.startswith(b"\x89PNG") else "image/jpeg"
return f"data:{mime};base64,{base64.b64encode(data).decode()}"
def qi_request(messages: list, metadata: dict, session_id: str) -> dict:
"""调用 QI，返回本轮动作（dict，含 action_type）。
AK 签名 + Agent API Key + Instance ID 鉴权；session_id 贯穿一个任务。
"""
body = json.dumps({
"model": "",
"messages": messages,
"metadata": metadata,
"tools": [],
"stream": False,
}, ensure_ascii=False).encode("utf-8")
request_id = f"round-{uuid.uuid4().hex}"
# data=body 直接发送已签名的字节，不能再次 JSON 序列化。
endpoint = urllib.parse.urlsplit(ENDPOINT)
conn = http.client.HTTPSConnection(endpoint.hostname, endpoint.port, timeout=300)
try:
# urllib 会把 X-QI-* 改为 X-Qi-*；直接发送以保留网关要求的头名称。
conn.request("POST", endpoint.path or "/", body=body,
headers=sign_headers(body, session_id, request_id, stream=False))
resp = conn.getresponse()
raw = resp.read()
if resp.status >= 400:
detail = raw.decode("utf-8", errors="replace")
for name in ("QI_AK_ID", "QI_AK_SECRET", "GUI_API_KEY", "GUI_INSTANCE_ID", "QI_STS_TOKEN"):
value = os.environ.get(name)
if value:
detail = detail.replace(value, "[redacted]")
raise RuntimeError(f"QI HTTP {resp.status}: {detail}")
data = json.loads(raw)
finally:
conn.close()
arguments = data["choices"][0]["message"]["tool_calls"][0]["function"]["arguments"]
return json.loads(arguments)
# ======================================================================
# 2. ADB 执行器：把 QI 的动作落到设备上
# ======================================================================
class AdbExecutor:
def __init__(self, device: str):
self.device = device
def adb(self, *args: str, binary: bool = False):
out = subprocess.run(["adb", "-s", self.device, *args], capture_output=True)
return out.stdout if binary else out.stdout.decode(errors="replace").strip()
def screenshot(self) -> bytes:
return self.adb("exec-out", "screencap", "-p", binary=True)
def foreground_package(self) -> str | None:
"""当前前台 App 的包名：从 dumpsys 里抓 `包名/Activity` token。"""
out = self.adb("shell", "dumpsys", "activity", "activities")
for line in out.splitlines():
if "mResumedActivity" in line or "mFocusedApp" in line:
m = re.search(r"([A-Za-z][A-Za-z0-9_.]*)/[A-Za-z0-9_.$]+", line)
if m:
return m.group(1)
return None
def execute(self, action: dict) -> str:
"""执行一个 QI 动作，返回简短状态字符串。"""
t = action["action_type"]
if t == "click":
self.adb("shell", "input", "tap", str(action["x"]), str(action["y"]))
elif t == "double_click":
x, y = str(action["x"]), str(action["y"])
self.adb("shell", "input", "tap", x, y)
time.sleep(0.1)
self.adb("shell", "input", "tap", x, y)
elif t == "long_press":
x, y = str(action["x"]), str(action["y"])
self.adb("shell", "input", "swipe", x, y, x, y, "1000")
elif t == "drag":
self.adb("shell", "input", "swipe",
str(action["start_x"]), str(action["start_y"]),
str(action["end_x"]), str(action["end_y"]), "400")
elif t == "type":
b64 = base64.b64encode(action["text"].encode()).decode()
self.adb("shell", "am", "broadcast", "-a", "ADB_INPUT_B64", "--es", "msg", b64)
elif t == "open":
self.adb("shell", "monkey", "-p", resolve_package(action["text"]),
"-c", "android.intent.category.LAUNCHER", "1")
elif t == "system_button":
key = {"back": "KEYCODE_BACK", "home": "KEYCODE_HOME",
"enter": "KEYCODE_ENTER", "menu": "KEYCODE_MENU"}[action["button"]]
self.adb("shell", "input", "keyevent", key)
elif t == "wait":
time.sleep(action.get("duration", 1))
return f"{t} ok"
# ======================================================================
# 3. 主循环：截图 → 请求 QI → 执行动作 → 下一轮
# ======================================================================
def run_task(executor: AdbExecutor, task: str, apps: list, max_steps: int, delay: float):
session_id = f"sess-{uuid.uuid4().hex[:16]}"
print(f"session: {session_id} | task: {task} | apps: {apps}")
# next_text：下一轮 user 消息里可选携带的一段文本。仅两种情况写入：
# 1) open 后：前台 App 校验结果（已经/未成功打开 xx）；
# 2) take_over 恢复后：用户输入文本，或默认「用户已经完成操作，请继续执行」。
next_text = None
for step in range(1, max_steps + 1):
shot = executor.screenshot()
# 每轮只发一条 user 消息 + 一张当前截图，多轮上下文由 QI 按 session_id 维护。
# 首轮 text 是任务；后续轮仅在需要时带 next_text。
text = task if step == 1 else next_text
content = ([{"type": "text", "text": text}] if text else [])
content.append({"type": "image_url", "image_url": {"url": to_data_url(shot)}})
messages = [{"role": "user", "content": content}]
metadata = {"available_apps": apps} if step == 1 else {}
next_text = None # 本轮已消费，重置
action = qi_request(messages, metadata, session_id)
action_type = action["action_type"]
print(f"[step {step}] {action_type}: {action}")
if action_type == "terminate": # 任务结束
print(f"完成 status={action.get('status')} answer={action.get('answer')}")
return
if action_type == "take_over":
# 阻塞等待用户输入。有输入则回传该文本，否则用默认继续文案。
print(f"需要人工介入：{action.get('text')}")
user_input = input(">>> 请输入补充信息后回车继续（直接回车表示已完成操作）：").strip()
next_text = user_input or "用户已经完成操作，请继续执行"
print(f"已提交：{next_text}")
continue
executor.execute(action)
time.sleep(delay)
# open 动作后校验前台 App 是否已是目标，结果作为下一轮的文本回传。
if action_type == "open":
name = (action.get("text") or "").strip()
ok = executor.foreground_package() == resolve_package(name)
next_text = f"已经成功打开{name}" if ok else f"未成功打开{name}"
print(f"前台校验：{next_text}")
print(f"达到最大步数 {max_steps}")
def main():
p = argparse.ArgumentParser(description="QI GUI Agent 最小接入 demo")
p.add_argument("--task", required=True, help="任务指令，如：打开设置进入 WiFi 页面")
p.add_argument("--app", nargs="*", default=[], help="设备可用 App 名列表")
p.add_argument("--device", default=os.environ.get("ANDROID_DEVICE", "emulator-5554"))
p.add_argument("--max-steps", type=int, default=100)
p.add_argument("--step-delay", type=float, default=1.2)
p.add_argument("--screenshot", help="单轮测试：只发这张图，打印 QI 返回的动作后退出")
args = p.parse_args()
try:
validate_config()
except ValueError as exc:
p.error(str(exc))
if args.screenshot: # 无需真机的单轮 API 测试
messages = [{"role": "user", "content": [
{"type": "text", "text": args.task},
{"type": "image_url", "image_url": {"url": to_data_url(args.screenshot)}}]}]
print(json.dumps(qi_request(messages, {"available_apps": args.app},
f"sess-{uuid.uuid4().hex[:16]}"),
ensure_ascii=False, indent=2))
return
run_task(AdbExecutor(args.device), args.task, args.app, args.max_steps, args.step_delay)
if __name__ == "__main__":
main()
