lambdagent/link/test_xiaoyi2.py

#!/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()