chendelian
d2dbb5f61e
Some checks failed
Publish Python Package to Gitea / publish (push) Failing after 59s
83 lines
2.8 KiB
Python
83 lines
2.8 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""匠厂 Host HTTP API 的 Python 轻封装。
|
||
|
||
使用 stdlib urllib，避免为 SDK 引入新运行时依赖。当前只暴露本模块
|
||
真正需要的端点（健康探活、agent 列表），**不包含任何会话删除/会话
|
||
文件系统操作**——测试侧要保持"只通过 UI 操作，不走底层接口"的原则，
|
||
后续如需访问其他 host-api 端点请在此处按需添加读型方法。
|
||
|
||
Host API 默认端口：`13210`（见 jiangchang/electron/utils/config.ts 中 `CLAWX_HOST_API`）。
|
||
可通过环境变量 ``CLAWX_PORT_CLAWX_HOST_API`` 或本类的 ``port`` 参数覆盖。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import os
|
||
import urllib.error
|
||
import urllib.request
|
||
from typing import Any, Dict, Optional
|
||
|
||
DEFAULT_PORT = 13210
|
||
DEFAULT_HOST = "127.0.0.1"
|
||
|
||
logger = logging.getLogger("jiangchang_desktop_sdk.testing.host_api")
|
||
|
||
|
||
class HostAPIError(RuntimeError):
|
||
"""Host API 调用失败（网络错误或非 2xx 响应）。"""
|
||
|
||
|
||
class HostAPIClient:
|
||
def __init__(
|
||
self,
|
||
host: str = DEFAULT_HOST,
|
||
port: Optional[int] = None,
|
||
timeout: float = 10.0,
|
||
) -> None:
|
||
self.host = host
|
||
env_port = os.environ.get("CLAWX_PORT_CLAWX_HOST_API")
|
||
self.port = port or (int(env_port) if env_port else DEFAULT_PORT)
|
||
self.timeout = timeout
|
||
|
||
@property
|
||
def base_url(self) -> str:
|
||
return f"http://{self.host}:{self.port}"
|
||
|
||
def _request(self, method: str, path: str, body: Any = None) -> Any:
|
||
url = f"{self.base_url}{path}"
|
||
data = None
|
||
headers = {"Accept": "application/json"}
|
||
if body is not None:
|
||
data = json.dumps(body).encode("utf-8")
|
||
headers["Content-Type"] = "application/json"
|
||
req = urllib.request.Request(url, data=data, method=method, headers=headers)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=self.timeout) as resp:
|
||
raw = resp.read().decode("utf-8")
|
||
if not raw:
|
||
return None
|
||
try:
|
||
return json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
return raw
|
||
except urllib.error.HTTPError as exc:
|
||
payload: Any
|
||
try:
|
||
payload = json.loads(exc.read().decode("utf-8"))
|
||
except Exception: # noqa: BLE001
|
||
payload = str(exc)
|
||
raise HostAPIError(f"{method} {path} -> HTTP {exc.code}: {payload}") from exc
|
||
except urllib.error.URLError as exc:
|
||
raise HostAPIError(f"{method} {path} -> {exc}") from exc
|
||
|
||
def ping(self) -> bool:
|
||
try:
|
||
self.list_agents()
|
||
return True
|
||
except HostAPIError:
|
||
return False
|
||
|
||
def list_agents(self) -> Dict[str, Any]:
|
||
return self._request("GET", "/api/agents") or {}
|