Python 实战:自开发程序通过API 操控 OpenClaw ,打造专属 AI 自动化工具

def send_chat(self, message: str, session_key: str = "agent:main:main",              wait_for_response: bool = True, timeout: int = 120) -> Optional[Dict]:    # 重置状态    self._current_response_text = ""    self._current_run_id = None    self._chat_responses.clear()    self._last_printed_len.clear()    # 发送聊天请求    result = self.send_request("chat.send", {        "sessionKey": session_key,        "message": message,        "deliver": False,        "idempotencyKey": str(uuid.uuid4())  # 唯一标识符    }, timeout=60)    # 等待响应    if result and wait_for_response:        run_id = result.get("runId")        # 等待直到收到完成状态        start_time = time.time()        while time.time() - start_time < timeout:            if run_id in self._chat_responses:                resp = self._chat_responses[run_id]                if resp.get("state") in ["finished", "done", "error"]:                    result["response_text"] = resp.get("text", "")                    result["response_state"] = resp.get("state")                    break            time.sleep(0.5)    return result

def _on_message(self, ws, message):    try:        data = json.loads(message)        msg_type = data.get("type")        if msg_type == "event":            event = data.get("event")            payload = data.get("payload", {})            if event == "connect.challenge":                self._send_auth_connect()            elif event == "hello":                self.connected = True                self.server_version = payload.get("server", {}).get("version")            elif event == "agent":                self._handle_agent_event(event, payload)            elif event == "chat":                self._handle_chat_event(event, payload)        elif msg_type == "res":            # 处理请求响应            req_id = data.get("id")            ok = data.get("ok")            if req_id in self.pending:                if ok:                    self.pending[req_id]["result"] = data.get("payload", {})                    self.pending[req_id]["ok"] = True                else:                    self.pending[req_id]["ok"] = False                    self.pending[req_id]["error"] = data.get("error", {})                self.pending[req_id]["event"].set()    except Exception as e:        print(f"[OpenClaw] Error parsing message: {e}")

def _print_incremental(self, run_id: str, full_text: str):    """只打印新增的文本部分"""    last_len = self._last_printed_len.get(run_id, 0)    if len(full_text) > last_len:        new_text = full_text[last_len:]        print(new_text, end="", flush=True)        self._last_printed_len[run_id] = len(full_text)

elif stream in ("assistant", "text"):    text = data.get("text", "")    if text and len(text) > 0:        # 检查是否是新的、更长的文本        current_text = self._chat_responses.get(run_id, {}).get("text", "")        if len(text) > len(current_text):            self._chat_responses[run_id]["text"] = text            self._current_response_text = text            # 增量打印            self._print_incremental(run_id, text)