#!/usr/bin/env python3 """小艺 ↔ Agent67 桥接 — PaaS agent: ag_e7e376960efe (lambda-xiaoyi)""" import json, time, hmac, hashlib, base64, ssl, uuid, threading, sys AK = 'd031c15a2c7e4ad7a8c71312c31816c9' SK = '8FB5757D89AA777C31EE542A7C0BF966347143004C3591766762CF429AC21BBB' AGENT_ID = 'agent935da52acaeb48d09f6adc858c7d3ae2' WS1 = 'wss://hag.cloud.huawei.com/openclaw/v1/ws/link' WS2 = 'wss://116.63.174.231/openclaw/v1/ws/link' PAAS_AGENT_ID = 'ag_e7e376960efe' PAAS_API_KEY = 'ap_8c8694e7f49d9af60c30a92ae17971d6' import os from pathlib import Path _project_root = Path(__file__).resolve().parent sys.path.insert(0, str(_project_root)) sys.path.insert(0, str(_project_root / 'agentexample')) _assistant = None LOG = open('/tmp/xiaoyi_test.log', 'w', buffering=1) def log(msg): line = f'[{time.strftime("%H:%M:%S")}] {msg}' print(line, flush=True) LOG.write(line + '\n') LOG.flush() def headers(): ts = str(int(time.time() * 1000)) sig = base64.b64encode(hmac.new(SK.encode(), ts.encode(), hashlib.sha256).digest()).decode() return {'x-access-key': AK, 'x-sign': sig, 'x-ts': ts, 'x-agent-id': AGENT_ID} def wrap(session_id, task_id, msg_id, result_obj): jsonrpc = {'jsonrpc': '2.0', 'id': msg_id, 'result': result_obj} return json.dumps({ 'msgType': 'agent_response', 'agentId': AGENT_ID, 'sessionId': session_id, 'taskId': task_id, 'msgDetail': json.dumps(jsonrpc), }) def run_server(name, url): import websocket sslopt = {} if '116.63' in url: sslopt = {'cert_reqs': ssl.CERT_NONE, 'check_hostname': False} ws = websocket.WebSocket(sslopt=sslopt) ws.timeout = 600 log(f'[{name}] Connecting...') ws.connect(url, header=headers()) log(f'[{name}] CONNECTED') # init ws.send(json.dumps({'msgType': 'clawd_bot_init', 'agentId': AGENT_ID})) log(f'[{name}] init sent') # heartbeat thread def heartbeat(): while True: time.sleep(25) try: ws.send(json.dumps({'msgType': 'heartbeat', 'agentId': AGENT_ID, 'timestamp': int(time.time()*1000)})) except: break t = threading.Thread(target=heartbeat, daemon=True) t.start() # listen while True: try: raw = ws.recv() if not raw: continue data = json.loads(raw) if data.get('msgType') == 'heartbeat': continue log(f'📩 [{name}] {json.dumps(data, ensure_ascii=False)[:300]}') params = data.get('params', {}) task_id = params.get('id', data.get('id', '')) session_id = params.get('sessionId', data.get('sessionId', '')) msg_id = data.get('id', '') parts = params.get('message', {}).get('parts', []) text = ' '.join(p.get('text', '') for p in parts if p.get('kind') == 'text').strip() if not text or not task_id: continue top_session = data.get('sessionId', '') log(f' User: {text}') log(f' param_session={session_id} top_session={top_session}') sid = top_session # 定期发 working 状态防止小艺超时 agent_done = threading.Event() agent_reply = [None] def keep_alive(): """每 8 秒发一次 reasoningText，保持连接活跃""" tick = 0 msgs = ['正在思考...', '分析问题中...', '调用工具中...', '处理数据...', '即将完成...'] while not agent_done.is_set(): time.sleep(8) if agent_done.is_set(): break tick += 1 # 用 reasoningText (原始插件的做法)，不用 status working ws.send(wrap(sid, task_id, msg_id, { 'taskId': task_id, 'kind': 'artifact-update', 'append': True, 'lastChunk': True, 'final': False, 'artifact': {'artifactId': f'reason_{int(time.time()*1000)}', 'parts': [{'kind': 'reasoningText', 'reasoningText': msgs[tick % len(msgs)] + '\n'}]}, })) log(f' [{name}] ♻ reasoning #{tick}') ka_thread = threading.Thread(target=keep_alive, daemon=True) ka_thread.start() t0 = time.time() # 调用 agent67 (Claude Code) try: from agent67.core.assistant import PersonalAssistant global _assistant if _assistant is None: _assistant = PersonalAssistant(model="sonnet", use_api=False) reply = _assistant.chat(text) except Exception as e: log(f' [{name}] agent error: {e}') reply = f'🐂 lambda67: 抱歉，处理出错了: {e}' finally: agent_done.set() art_id = f'artifact_{int(time.time()*1000)}' # 1. 内容 (append=true, final=false) ws.send(wrap(sid, task_id, msg_id, { 'taskId': task_id, 'kind': 'artifact-update', 'append': True, 'lastChunk': False, 'final': False, 'artifact': {'artifactId': art_id, 'parts': [{'kind': 'text', 'text': reply}]}, })) log(f' [{name}] → content (sid={sid[:16]})') # 2. 空结束标记 (append=true, final=true, text="") ws.send(wrap(sid, task_id, msg_id, { 'taskId': task_id, 'kind': 'artifact-update', 'append': True, 'lastChunk': True, 'final': True, 'artifact': {'artifactId': art_id, 'parts': [{'kind': 'text', 'text': ''}]}, })) log(f' [{name}] → final (sid={sid[:16]})') # 3. status completed — 告诉小艺结束了 status_id = f'done_{int(time.time()*1000)}' ws.send(wrap(sid, task_id, status_id, { 'taskId': task_id, 'kind': 'status-update', 'final': True, 'status': {'state': 'completed', 'message': {'role': 'agent', 'parts': []}}, })) log(f' [{name}] → status completed') # 记录到 PaaS try: import urllib.request as _ur _paas_body = json.dumps({ 'input': text[:500], 'output': reply[:2000], 'status': 'completed', 'duration_ms': int((time.time() - t0) * 1000), 'steps': 1, 'source': 'xiaoyi-openclaw', }).encode() _req = _ur.Request( f'http://127.0.0.1:8067/api/v1/agents/{PAAS_AGENT_ID}/runs/record', data=_paas_body, headers={'Content-Type': 'application/json', 'Authorization': f'Bearer {PAAS_API_KEY}'}, method='POST') _ur.urlopen(_req, timeout=5) except Exception: pass log(f' [{name}] DONE ✓') except Exception as e: log(f'[{name}] ERROR: {e}') break log('=== 小艺测试 v2 (websocket-client, 无自动 ping) ===') t1 = threading.Thread(target=run_server, args=('S1', WS1), daemon=True) t2 = threading.Thread(target=run_server, args=('S2', WS2), daemon=True) t1.start() t2.start() t1.join() t2.join()