""" ZQAgent 对外 API 的 Python 客户端（最小实现，只用标准库）。 用法： export ZQA_BASE=https://hub.example.com export ZQA_API_KEY=zqa_live_xxx python examples/api/python/zqagent_client.py 设计取舍： - 只用 `urllib`，不引入 requests：集成方往往有依赖审查，零依赖最容易过； - 不做自动重试：任务提交有副作用，重试策略必须由调用方按业务决定 （配合 Idempotency-Key 才安全）； - 只封装"提交 / 查询 / 批次 / 资源"四类，避免把整个 API 抄一遍后失同步。 """ from __future__ import annotations import json import os import time import urllib.error import urllib.parse import urllib.request from dataclasses import dataclass from typing import Any, Iterable class ZqAgentError(Exception): """服务端返回的结构化错误。 `code` 是稳定标识（按它写分支），`message` 是给人看的描述。 `retryable` 表示服务端认为"重试可能成功"。 """ def __init__(self, status: int, code: str, message: str, request_id: str | None = None): super().__init__(f"[{status}] {code}: {message}") self.status = status self.code = code self.message = message self.request_id = request_id @dataclass class JobView: """作业的只读视图（字段与 GET /v1/jobs/{id} 一致，只列常用项）。""" raw: dict[str, Any] @property def id(self) -> str: return self.raw["id"] @property def external_id(self) -> str | None: return self.raw.get("externalId") @property def status(self) -> str: return self.raw["status"] @property def is_terminal(self) -> bool: return self.status in ("done", "failed", "cancelled") @property def output(self) -> str | None: return (self.raw.get("result") or {}).get("summary") @property def produced_files(self) -> list[str]: return (self.raw.get("result") or {}).get("producedFiles") or [] @property def failure(self) -> dict[str, Any] | None: return self.raw.get("failure") class ZqAgentClient: def __init__(self, base_url: str, api_key: str, timeout: float = 30.0): self.base_url = base_url.rstrip("/") self.api_key = api_key self.timeout = timeout # ---------------- 底层 ---------------- def _request( self, method: str, path: str, body: dict[str, Any] | None = None, headers: dict[str, str] | None = None, ) -> Any: url = f"{self.base_url}{path}" data = json.dumps(body).encode("utf-8") if body is not None else None req = urllib.request.Request(url, data=data, method=method) req.add_header("Authorization", f"Bearer {self.api_key}") req.add_header("Accept", "application/json") if data is not None: req.add_header("Content-Type", "application/json") for k, v in (headers or {}).items(): req.add_header(k, v) try: with urllib.request.urlopen(req, timeout=self.timeout) as resp: payload = json.loads(resp.read().decode("utf-8")) # 成功响应统一是 {data: ..., requestId: ...}； # OpenAI 兼容层例外，直接返回顶层对象 return payload.get("data", payload) except urllib.error.HTTPError as e: raw = e.read().decode("utf-8", "replace") try: err = json.loads(raw).get("error", {}) except json.JSONDecodeError: err = {} raise ZqAgentError( status=e.code, code=err.get("code", "unknown"), message=err.get("message", raw[:200]), request_id=err.get("requestId"), ) from e # ---------------- 资源 ---------------- def overview(self, compact: bool = False) -> dict[str, Any]: q = "?format=compact" if compact else "" return self._request("GET", f"/v1/overview{q}") def projects(self) -> list[dict[str, Any]]: return self._request("GET", "/v1/projects")["items"] def nodes(self) -> list[dict[str, Any]]: return self._request("GET", "/v1/nodes")["items"] def node(self, node_id: str) -> dict[str, Any]: return self._request("GET", f"/v1/nodes/{urllib.parse.quote(node_id)}") def usage(self, days: int = 30) -> dict[str, Any]: return self._request("GET", f"/v1/usage?days={days}") # ---------------- 任务 ---------------- def submit( self, items: Iterable[dict[str, Any]], batch_id: str | None = None, idempotency_key: str | None = None, ) -> dict[str, Any]: """批量提交。`idempotency_key` 强烈建议传（网络重试不会重复建任务）。""" body: dict[str, Any] = {"items": list(items)} if batch_id: body["batchId"] = batch_id headers = {"Idempotency-Key": idempotency_key} if idempotency_key else None return self._request("POST", "/v1/jobs", body, headers) def submit_ai( self, project_id: str, prompt: str, external_id: str | None = None, node_id: str | None = None, callback_url: str | None = None, metadata: dict[str, Any] | None = None, idempotency_key: str | None = None, ) -> dict[str, Any]: """提交一条 AI 任务的便捷方法。""" item: dict[str, Any] = {"kind": "ai", "projectId": project_id, "prompt": prompt} if external_id: item["externalId"] = external_id if node_id: item["nodeId"] = node_id if callback_url: item["callbackUrl"] = callback_url if metadata: item["metadata"] = metadata return self.submit([item], idempotency_key=idempotency_key) def get_job(self, job_id: str) -> JobView: return JobView(self._request("GET", f"/v1/jobs/{urllib.parse.quote(job_id)}")) def list_jobs(self, **filters: str) -> list[JobView]: query = urllib.parse.urlencode({k: v for k, v in filters.items() if v}) path = "/v1/jobs" + (f"?{query}" if query else "") return [JobView(x) for x in self._request("GET", path)["items"]] def find_by_external_id(self, external_id: str) -> JobView | None: """用业务单号定位任务（比翻页可靠得多）。""" items = self.list_jobs(externalId=external_id) return items[0] if items else None def cancel(self, job_id: str) -> JobView: return JobView(self._request("POST", f"/v1/jobs/{urllib.parse.quote(job_id)}/cancel")) def retry(self, job_id: str) -> JobView: return JobView(self._request("POST", f"/v1/jobs/{urllib.parse.quote(job_id)}/retry")) def batch(self, batch_id: str) -> dict[str, Any]: return self._request("GET", f"/v1/batches/{urllib.parse.quote(batch_id)}") def cancel_batch(self, batch_id: str) -> int: return self._request("POST", f"/v1/batches/{urllib.parse.quote(batch_id)}/cancel")["cancelled"] def wait(self, job_id: str, timeout: float = 300.0, interval: float = 3.0) -> JobView: """轮询到终态。生产环境优先用 webhook，这个只适合脚本/兜底。""" deadline = time.time() + timeout while time.time() < deadline: job = self.get_job(job_id) if job.is_terminal: return job time.sleep(interval) raise TimeoutError(f"任务 {job_id} 在 {timeout}s 内未结束") def main() -> None: base = os.environ.get("ZQA_BASE", "http://127.0.0.1:8788") key = os.environ.get("ZQA_API_KEY", "") if not key: raise SystemExit("请先 export ZQA_API_KEY=zqa_live_xxx") client = ZqAgentClient(base, key) projects = client.projects() if not projects: raise SystemExit("没有可用项目：请先在控制台为某个节点创建项目") target = projects[0] print(f"目标项目：{target['name']}（{target['projectId']} @ {target['nodeId']}）") submitted = client.submit_ai( project_id=target["projectId"], prompt="列出当前目录下的文件，并说明每个文件的用途", external_id="DEMO-1", idempotency_key="demo-20261109-001", ) job_id = submitted["items"][0]["id"] print(f"已提交：{job_id}") job = client.wait(job_id, timeout=600) print(f"状态：{job.status}") if job.output: print(f"输出：\n{job.output}") if job.produced_files: print(f"改动文件：{', '.join(job.produced_files)}") if job.failure: print(f"失败：{job.failure}") if __name__ == "__main__": main()