hermesclaw.py
Rama main · réplica de lectura del repositorio original.
"""HermesClaw v4: triple-gateway proxy router for WeChat.
Takes over one iLink token, polls for messages, and distributes them
to two independent proxy servers -- one for OpenClaw's clawbot and one
for Hermes Agent's WeChat gateway. Each gateway believes it is talking
directly to the iLink API.
"""
import json
import logging
import base64
import mimetypes
import os
import queue
import secrets
import signal
import shutil
import subprocess
import sys
import threading
import time
from enum import Enum
from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler
from pathlib import Path
from urllib.parse import urlparse
import requests
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
T, IMG, VO, VIDEO, FILE = 1, 2, 3, 4, 5
ILINK_VER = "2.1.7"
ILINK_CV = "65547"
QUEUE_CAP = 200
DEFAULT_POLL_SEC = 35
MAX_OPENCODE_MEDIA_BYTES = 10 * 1024 * 1024
log = logging.getLogger("hermesclaw")
# ---------------------------------------------------------------------------
# Route enum & persistent state
# ---------------------------------------------------------------------------
class Route(str, Enum):
HERMES = "hermes"
OPENCLAW = "openclaw"
OPENCODE = "opencode"
BOTH = "both"
THREE = "three"
class State:
"""Per-user routing state, persisted to JSON."""
def __init__(self, fp):
self.fp = Path(fp)
self.d = {}
self.lock = threading.Lock()
if self.fp.exists():
try:
self.d = json.loads(self.fp.read_text())
log.info("Loaded %d route states", len(self.d))
except Exception:
pass
def save(self):
t = self.fp.with_suffix(".tmp")
t.write_text(json.dumps(self.d, indent=2, ensure_ascii=False))
t.replace(self.fp)
def get(self, uid):
with self.lock:
if uid not in self.d:
self.d[uid] = {"route": Route.HERMES.value, "status_shown": False}
entry = self.d[uid]
# Migrate v1 format {"b": ...} to v2 {"route": ...}
raw = entry.get("route") or entry.get("b", Route.HERMES.value)
return Route(raw)
def should_show_status(self, uid):
with self.lock:
if uid not in self.d:
self.d[uid] = {"route": Route.HERMES.value, "status_shown": False}
return not self.d[uid].get("status_shown", False)
def mark_status_shown(self, uid):
with self.lock:
if uid not in self.d:
self.d[uid] = {"route": Route.HERMES.value, "status_shown": True}
else:
self.d[uid]["status_shown"] = True
self.save()
def set(self, uid, route):
with self.lock:
status_shown = self.d.get(uid, {}).get("status_shown", False)
self.d[uid] = {"route": route.value, "status_shown": status_shown}
self.save()
log.info("User %s -> %s", uid[:16], route.value)
# ---------------------------------------------------------------------------
# Text extraction (voice -> transcription)
# ---------------------------------------------------------------------------
def extract_text(items):
"""Return combined text from iLink items.
Voice items contribute only their iLink transcription by design.
"""
parts = []
for it in items:
tp = it.get("type", 0)
if tp == T:
x = it.get("text_item", {}).get("text", "")
if x:
parts.append(x)
elif tp == VO:
x = it.get("voice_item", {}).get("text", "")
if x:
parts.append(
f'[The user sent a voice message. Here\'s what they said: "{x}"]'
)
return "\n".join(parts).strip()
def media_kind(tp):
return {
IMG: "image",
VIDEO: "video",
FILE: "file",
}.get(tp, f"type-{tp}")
def _walk_values(obj):
if isinstance(obj, dict):
for k, v in obj.items():
yield k, v
yield from _walk_values(v)
elif isinstance(obj, list):
for v in obj:
yield from _walk_values(v)
def _first_value_by_keys(obj, keys):
wanted = {k.lower() for k in keys}
for k, v in _walk_values(obj):
if str(k).lower() in wanted and v not in (None, ""):
return str(v)
return ""
def _first_url(obj):
for _, v in _walk_values(obj):
if isinstance(v, str) and v.startswith(("http://", "https://")):
return v
return ""
def _media_name(item, kind, url):
name = _first_value_by_keys(item, ("name", "filename", "file_name", "title"))
if name:
return name
if url:
tail = url.rstrip("/").rsplit("/", 1)[-1].split("?", 1)[0]
if tail:
return tail
return f"wechat-{kind}"
def _media_mime(item, kind, url):
mime = _first_value_by_keys(item, ("mime_type", "mimeType", "content_type", "contentType"))
if not mime:
guessed, _ = mimetypes.guess_type(url or _media_name(item, kind, url))
mime = guessed or {
"image": "image/jpeg",
"video": "video/mp4",
"file": "application/octet-stream",
}.get(kind, "application/octet-stream")
# File attachments must not carry a video/* MIME — the mixed signal
# (kind="file" but mimeType="video/mp4") confuses the LLM into treating
# the attachment as a video even when the user sent it as a document.
if kind == "file" and mime.startswith("video/"):
mime = "application/octet-stream"
return mime
def _download_media(url, token):
if not url:
return None, "", None
try:
headers = hdrs(token)
r = requests.get(url, headers=headers, timeout=20, stream=True)
r.raise_for_status()
chunks = []
size = 0
for chunk in r.iter_content(chunk_size=65536):
if not chunk:
continue
size += len(chunk)
if size > MAX_OPENCODE_MEDIA_BYTES:
return None, "", f"media exceeds {MAX_OPENCODE_MEDIA_BYTES} bytes"
chunks.append(chunk)
return b"".join(chunks), r.headers.get("Content-Type", "").split(";", 1)[0], None
except Exception as e:
return None, "", str(e)
def build_opencode_prompt(items, token=""):
"""Build ACP content blocks from iLink items for OpenCode."""
blocks = []
txt = extract_text(items)
if txt:
blocks.append({"type": "text", "text": txt})
for item in items:
tp = item.get("type", 0)
if tp in (T, VO) or tp == 0:
continue
# Prefer the actual sub-key structure over type number alone — some
# iLink clients send document files with type=4 (video), so looking
# at the key name is more reliable than the integer type.
if "image_item" in item:
kind, tp = "image", IMG
elif "file_item" in item:
kind, tp = "file", FILE
elif "video_item" in item:
kind, tp = "video", VIDEO
else:
kind = media_kind(tp)
url = _first_url(item)
name = _media_name(item, kind, url)
mime = _media_mime(item, kind, url)
metadata = {
"kind": kind,
"name": name,
"mimeType": mime,
"url": url or None,
"itemType": tp,
}
blocks.append({
"type": "text",
"text": (
f"The user sent a {kind}. Use the attached content or link if available. "
f"Metadata: {json.dumps(metadata, ensure_ascii=False)}"
),
})
if kind == "image" and url:
data, content_type, err = _download_media(url, token)
if data:
blocks.append({
"type": "image",
"data": base64.b64encode(data).decode(),
"mimeType": content_type or mime,
"uri": url,
})
continue
log.info("OpenCode image download unavailable (%s): %s", url[:80], err)
if url:
block = {
"type": "resource_link",
"name": name,
"uri": url,
"mimeType": mime,
"title": name,
}
size = _first_value_by_keys(item, ("size", "file_size", "fileSize"))
if size and size.isdigit():
block["size"] = int(size)
blocks.append(block)
return blocks
# ---------------------------------------------------------------------------
# OpenCode ACP bridge
# ---------------------------------------------------------------------------
class ACPSession:
"""A single ACP session backed by an opencode acp subprocess."""
def __init__(self, opencode_cmd, cwd, model, permission_strategy="allow_always"):
self.model = model
self.permission_strategy = permission_strategy
self.session_id = None
self.prompt_capabilities = {}
self.alive = False
self._req_id = 0
self._lock = threading.Lock()
self._write_lock = threading.Lock()
self._pending = {} # req_id -> threading.Event
self._results = {} # req_id -> msg
self._text_buf = [] # accumulates text chunks for current prompt
self._active_req_id = None # req_id of the in-flight session/prompt
self._stderr_tail = []
self._proc = subprocess.Popen(
[opencode_cmd, "acp"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
cwd=cwd,
)
self._reader = threading.Thread(target=self._read_loop, daemon=True)
self._reader.start()
self._stderr_reader = threading.Thread(target=self._read_stderr_loop, daemon=True)
self._stderr_reader.start()
# ACP handshake
r = self._send_wait("initialize", {
"protocolVersion": 1,
"clientInfo": {
"name": "hermesclaw",
"title": "HermesClaw",
"version": "0.3.0",
},
"clientCapabilities": {},
}, timeout=30)
if not r or "error" in r:
raise RuntimeError(f"ACP initialize failed: {r}")
self.prompt_capabilities = (
r.get("result", {})
.get("agentCapabilities", {})
.get("promptCapabilities", {})
)
r = self._send_wait("session/new", {
"cwd": cwd,
"mcpServers": [],
}, timeout=15)
if not r or "error" in r or "result" not in r:
raise RuntimeError(f"ACP session/new failed: {r}")
self.session_id = r["result"]["sessionId"]
self.alive = True
def _read_loop(self):
while True:
try:
line = self._proc.stdout.readline()
if not line:
break
line = line.decode().strip()
if not line:
continue
msg = json.loads(line)
except Exception:
break
if "method" in msg and "id" in msg:
self._handle_agent_request(msg)
elif "id" in msg:
with self._lock:
self._results[msg["id"]] = msg
ev = self._pending.pop(msg["id"], None)
if ev:
ev.set()
elif msg.get("method") == "session/update":
upd = msg.get("params", {}).get("update", {})
if upd.get("sessionUpdate") == "agent_message_chunk":
content = upd.get("content", {})
if content.get("type") == "text":
with self._lock:
if self._active_req_id is not None:
self._text_buf.append(content["text"])
# EOF or error — mark dead and wake all pending waiters so they
# return immediately instead of blocking until their timeout expires.
stderr = self._stderr_snapshot()
if stderr:
log.warning("ACP read loop exited (subprocess may have died). stderr tail: %s", stderr)
else:
log.warning("ACP read loop exited (subprocess may have died)")
with self._lock:
self.alive = False
message = "ACP process died"
if stderr:
message += f": {stderr}"
dead_error = {"jsonrpc": "2.0", "error": {"code": -32000, "message": message}}
for req_id, ev in list(self._pending.items()):
self._results[req_id] = dead_error
ev.set()
self._pending.clear()
def _read_stderr_loop(self):
while True:
try:
line = self._proc.stderr.readline()
if not line:
break
text = line.decode(errors="replace").strip()
if not text:
continue
with self._lock:
self._stderr_tail.append(text)
self._stderr_tail = self._stderr_tail[-20:]
except Exception:
break
def _stderr_snapshot(self):
with self._lock:
return " | ".join(self._stderr_tail[-8:])
def _write_message(self, msg):
raw = json.dumps(msg)
with self._write_lock:
self._proc.stdin.write((raw + "\n").encode())
self._proc.stdin.flush()
def _handle_agent_request(self, msg):
method = msg.get("method")
req_id = msg.get("id")
if method == "session/request_permission":
option_id = self._select_permission_option(
msg.get("params", {}).get("options", [])
)
if option_id:
result = {"outcome": {"outcome": "selected", "optionId": option_id}}
else:
result = {"outcome": {"outcome": "cancelled"}}
log.info("ACP permission request handled with option=%s", option_id or "cancelled")
try:
self._write_message({"jsonrpc": "2.0", "id": req_id, "result": result})
except Exception as e:
log.warning("ACP permission response failed: %s", e)
return
log.warning("ACP unsupported agent request: %s", method)
try:
self._write_message({
"jsonrpc": "2.0",
"id": req_id,
"error": {"code": -32601, "message": f"Unsupported client method: {method}"},
})
except Exception as e:
log.warning("ACP unsupported-request response failed: %s", e)
def _select_permission_option(self, options):
preferred = {
"allow_always": ("allow_always", "allow_once"),
"allow_once": ("allow_once", "allow_always"),
"reject_once": ("reject_once", "reject_always"),
"reject_always": ("reject_always", "reject_once"),
}.get(self.permission_strategy, ("allow_once", "allow_always"))
for kind in preferred:
for option in options:
if option.get("kind") == kind:
return option.get("optionId")
# No recognised kind found — cancel rather than blindly picking options[0],
# which could accidentally select a reject/unknown option.
if options:
unknown_kinds = {o.get("kind") for o in options}
log.warning("ACP permission: no recognised option kind in %s; cancelling", unknown_kinds)
return None
def _send_wait(self, method, params, timeout=10):
with self._lock:
self._req_id += 1
req_id = self._req_id
ev = threading.Event()
self._pending[req_id] = ev
try:
self._write_message({"jsonrpc": "2.0", "id": req_id, "method": method, "params": params})
except Exception as e:
with self._lock:
self._pending.pop(req_id, None)
self.alive = False
log.warning("ACP stdin write failed (%s): %s", method, e)
return {"error": {"code": -32000, "message": f"stdin write failed: {e}"}}
if ev.wait(timeout):
with self._lock:
return self._results.pop(req_id, None)
log.warning("ACP timeout waiting for %s (id=%d)", method, req_id)
with self._lock:
self._pending.pop(req_id, None)
return None
def prompt_blocks(self, blocks, timeout=120):
"""Send ACP content blocks; return accumulated response text."""
with self._lock:
self._req_id += 1
req_id = self._req_id
self._active_req_id = req_id
self._text_buf.clear()
ev = threading.Event()
self._pending[req_id] = ev
try:
self._write_message({"jsonrpc": "2.0", "id": req_id, "method": "session/prompt",
"params": {"sessionId": self.session_id,
"prompt": blocks,
"model": self.model}})
except Exception as e:
with self._lock:
self._pending.pop(req_id, None)
self._active_req_id = None
self.alive = False
return f"[OpenCode error: stdin write failed: {e}]"
if ev.wait(timeout):
with self._lock:
self._active_req_id = None
r = self._results.pop(req_id, None)
text_out = "".join(self._text_buf).strip()
if r is None or "error" in (r or {}):
err = (r or {}).get("error", {}).get("message", "unknown error")
return f"[OpenCode error: {err}]"
return text_out
with self._lock:
self._active_req_id = None
self._pending.pop(req_id, None)
self.alive = False # force fresh session on next message
log.warning("ACP timeout on session/prompt (id=%d)", req_id)
return "[OpenCode: timeout — please try again]"
def prompt(self, text, timeout=120):
"""Send a text prompt; return accumulated response text."""
return self.prompt_blocks([{"type": "text", "text": text}], timeout)
def close(self):
self.alive = False
try:
self._proc.terminate()
self._proc.wait(timeout=5)
except Exception:
try:
self._proc.kill()
except Exception:
pass
class OpenCodeBridge:
"""Manages per-user ACPSession instances for OpenCode integration."""
def __init__(self, opencode_cmd, model="opencode/minimax-m2.5-free",
cwd=None, permission_strategy="allow_always"):
self.opencode_cmd = opencode_cmd
self.model = model
self.cwd = cwd or str(Path.home())
self.permission_strategy = permission_strategy
self._sessions = {} # uid -> ACPSession
self._lock = threading.Lock()
def is_available(self):
"""Return True if the opencode binary is found."""
resolved = shutil.which(self.opencode_cmd)
return bool(resolved or os.access(self.opencode_cmd, os.X_OK))
def send(self, uid, text, timeout=120):
"""Route text to this user's ACP session; return response."""
session = self._get_or_create(uid)
return session.prompt(text, timeout)
def send_blocks(self, uid, blocks, timeout=120):
"""Route ACP content blocks to this user's ACP session; return response."""
session = self._get_or_create(uid)
return session.prompt_blocks(blocks, timeout)
def _get_or_create(self, uid):
with self._lock:
s = self._sessions.get(uid)
if s is not None and s.alive:
return s
# Create the session outside the lock so other users are not blocked
# during the (sometimes slow) opencode acp startup sequence.
# opencode acp can crash silently on the first few attempts while its
# SQLite database initialises — retry up to 3 times with 2-second delays.
last_exc = None
for attempt in range(3):
if attempt > 0:
time.sleep(2)
log.info("Retrying OpenCode session for %s (attempt %d/3)",
uid[:16], attempt + 1)
try:
log.info("Creating OpenCode session for %s", uid[:16])
s = ACPSession(
self.opencode_cmd,
self.cwd,
self.model,
self.permission_strategy,
)
with self._lock:
self._sessions[uid] = s
return s
except Exception as e:
last_exc = e
log.warning("ACP session creation attempt %d/3 failed for %s: %s",
attempt + 1, uid[:16], e)
raise RuntimeError(f"OpenCode failed to start after 3 attempts: {last_exc}")
def close_session(self, uid):
with self._lock:
s = self._sessions.pop(uid, None)
if s:
s.close()
def close_all(self):
with self._lock:
sessions = list(self._sessions.values())
self._sessions.clear()
for s in sessions:
s.close()
# ---------------------------------------------------------------------------
# Router commands
# ---------------------------------------------------------------------------
def route_label(r):
if r == Route.HERMES:
return "Hermes"
if r == Route.OPENCLAW:
return "OpenClaw"
if r == Route.OPENCODE:
return "OpenCode"
if r == Route.BOTH:
return "Hermes + OpenClaw"
return "Hermes + OpenClaw + OpenCode"
def cmd(state, uid, text, opencode_bridge=None):
"""Process a slash command. Returns reply text or None for passthrough."""
c = text.strip().lower()
if c == "/hermes":
state.set(uid, Route.HERMES)
return "Switched to **Hermes**."
if c == "/openclaw":
state.set(uid, Route.OPENCLAW)
return "Switched to **OpenClaw**."
if c == "/opencode":
if opencode_bridge is not None and not opencode_bridge.is_available():
return (
"❌ OpenCode is not installed.\n"
"Please install it with:\n"
" npm install -g opencode-ai\n"
"Then restart HermesClaw and try /opencode again."
)
state.set(uid, Route.OPENCODE)
return "Switched to **OpenCode** 🤖"
if c == "/both":
state.set(uid, Route.BOTH)
return "Switched to **Hermes + OpenClaw**."
if c == "/three":
if opencode_bridge is not None and not opencode_bridge.is_available():
return (
"❌ OpenCode is not installed — /three requires it.\n"
"Please install it with:\n"
" npm install -g opencode-ai\n"
"Then restart HermesClaw and try /three again.\n"
"Tip: use /both for Hermes + OpenClaw only."
)
state.set(uid, Route.THREE)
return "Switched to **Hermes + OpenClaw + OpenCode** 🔱"
if c == "/whoami":
route = state.get(uid)
return (
f"**HermesClaw v4**\n"
f"**Current route**: **{route_label(route)}**\n"
f"**/hermes** → Hermes only\n"
f"**/openclaw** → OpenClaw only\n"
f"**/opencode** → OpenCode only\n"
f"**/both** → Hermes + OpenClaw\n"
f"**/three** → all three\n"
f"**/whoami** → this status"
)
return None
# ---------------------------------------------------------------------------
# iLink helpers
# ---------------------------------------------------------------------------
def hdrs(tok, body=""):
return {
"Content-Type": "application/json",
"AuthorizationType": "ilink_bot_token",
"Content-Length": str(len(body.encode())),
"iLink-App-Id": "",
"iLink-App-ClientVersion": ILINK_CV,
"Authorization": "Bearer " + tok if tok else "",
}
def ilink_post(base_url, ep, bd, tok, to=30):
url = base_url.rstrip("/") + "/" + ep.lstrip("/")
bs = json.dumps(bd)
r = requests.post(url, headers=hdrs(tok, bs), data=bs.encode(), timeout=to)
r.raise_for_status()
return r.json()
def get_updates_real(base_url, tok, buf="", to=None):
if to is None:
to = DEFAULT_POLL_SEC
try:
return ilink_post(
base_url,
"ilink/bot/getupdates",
{"get_updates_buf": buf, "base_info": {"channel_version": ILINK_VER}},
tok,
to + 5, # HTTP timeout slightly longer than iLink long-poll
)
except requests.exceptions.Timeout:
return {"ret": 0, "msgs": [], "get_updates_buf": buf}
except Exception as e:
log.warning("getUpdates: %s", e)
return {"ret": -1, "msgs": [], "get_updates_buf": buf}
def send_text_ilink(base_url, tok, to_user, text, ctx=None):
"""Send a plain text reply through iLink."""
m = {
"from_user_id": "",
"to_user_id": to_user,
"client_id": "hc-" + secrets.token_hex(8),
"message_type": 2,
"message_state": 2,
"item_list": [{"type": T, "text_item": {"text": text}}],
}
if ctx:
m["context_token"] = ctx
return ilink_post(
base_url,
"ilink/bot/sendmessage",
{"msg": m, "base_info": {"channel_version": ILINK_VER}},
tok,
)
def get_typing_ticket(base_url, tok, to_user, ctx=None):
"""Fetch iLink typing ticket for a user."""
body = {
"ilink_user_id": to_user,
"base_info": {"channel_version": ILINK_VER},
}
if ctx:
body["context_token"] = ctx
resp = ilink_post(base_url, "ilink/bot/getconfig", body, tok, to=10)
return resp.get("typing_ticket", "")
def send_typing_ilink(base_url, tok, to_user, typing_ticket, status=1):
"""Send or cancel iLink typing status."""
return ilink_post(
base_url,
"ilink/bot/sendtyping",
{
"ilink_user_id": to_user,
"typing_ticket": typing_ticket,
"status": status,
"base_info": {"channel_version": ILINK_VER},
},
tok,
to=10,
)
def keep_typing_ilink(base_url, tok, to_user, ctx, stop_event, interval=5):
"""Keep iLink typing status alive until stop_event is set."""
ticket = ""
try:
ticket = get_typing_ticket(base_url, tok, to_user, ctx)
if not ticket:
return
while not stop_event.is_set():
send_typing_ilink(base_url, tok, to_user, ticket, status=1)
stop_event.wait(interval)
except Exception as e:
log.debug("typing keepalive failed for %s: %s", to_user[:16], e)
finally:
if ticket:
try:
send_typing_ilink(base_url, tok, to_user, ticket, status=2)
except Exception as e:
log.debug("typing cancel failed for %s: %s", to_user[:16], e)
# ---------------------------------------------------------------------------
# Message queue with event-based long-poll
# ---------------------------------------------------------------------------
class MessageQueue:
"""Thread-safe queue with blocking dequeue for long-poll simulation."""
def __init__(self, capacity=QUEUE_CAP):
self.lock = threading.Lock()
self.event = threading.Event()
self.msgs = []
self.capacity = capacity
def enqueue(self, msg):
with self.lock:
if len(self.msgs) >= self.capacity:
self.msgs.pop(0)
log.warning("Queue full (%d), dropped oldest", self.capacity)
self.msgs.append(msg)
self.event.set()
def dequeue_all(self, timeout=None):
"""Return all queued messages, blocking up to *timeout* seconds."""
if timeout is not None and timeout > 0:
self.event.wait(timeout=timeout)
with self.lock:
batch = list(self.msgs)
self.msgs.clear()
self.event.clear()
return batch
def size(self):
with self.lock:
return len(self.msgs)
# ---------------------------------------------------------------------------
# Gateway proxy handler
# ---------------------------------------------------------------------------
PROXY_ALLOWLIST = frozenset([
"ilink/bot/getupdates",
"ilink/bot/sendmessage",
"ilink/bot/getuploadurl",
"ilink/bot/sendtyping",
"ilink/bot/getconfig",
"ilink/bot/get_bot_qrcode",
"ilink/bot/get_qrcode_status",
])
def make_proxy_handler(queue, ilink_base_url, ilink_token, state, tag):
"""Factory: create a request handler class bound to a specific queue.
*state* -- State instance for route lookup.
*tag* -- e.g. "[Hermes Agent]"; prepended to text items in
sendmessage when the destination user's route is /both.
"""
class GatewayProxyHandler(BaseHTTPRequestHandler):
def do_POST(self):
length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(length) if length else b""
ep = urlparse(self.path).path.lstrip("/")
if ep not in PROXY_ALLOWLIST:
self.send_response(404)
self.end_headers()
self.wfile.write(b'{"ret":-1,"errmsg":"not allowed"}')
log.warning("Proxy blocked: %s", ep)
return
try:
if ep == "ilink/bot/getupdates":
self._handle_getupdates(body)
elif ep == "ilink/bot/sendmessage":
self._handle_sendmessage(body)
else:
self._proxy_passthrough(body)
except BrokenPipeError:
log.debug("Client disconnected (BrokenPipeError)")
except ConnectionResetError:
log.debug("Client disconnected (ConnectionResetError)")
# -- getupdates: return from queue with long-poll ----------------
def _handle_getupdates(self, body):
try:
bd = json.loads(body) if body else {}
except Exception:
bd = {}
client_buf = bd.get("get_updates_buf", "")
msgs = queue.dequeue_all(timeout=DEFAULT_POLL_SEC)
resp = {"ret": 0, "msgs": msgs, "get_updates_buf": client_buf}
self._write_json(200, resp)
if msgs:
log.info("Proxy [%s] getupdates -> %d msgs", tag or "?", len(msgs))
# -- sendmessage: forward to real iLink with optional tagging ----
def _handle_sendmessage(self, body):
try:
bd = json.loads(body) if body else {}
except Exception:
bd = {}
# Tag text items only in /both mode for attribution.
if tag:
msg_obj = bd.get("msg", {})
to_user = msg_obj.get("to_user_id", "")
try:
route = state.get(to_user) if to_user else Route.HERMES
except Exception:
route = Route.HERMES
if route in (Route.BOTH, Route.THREE):
for item in msg_obj.get("item_list", []):
if item.get("type") == T:
ti = item.get("text_item", {})
original = ti.get("text", "")
if original:
ti["text"] = f"{tag} {original}"
self._forward_to_ilink(
"ilink/bot/sendmessage",
json.dumps(bd).encode() if bd else body,
)
# -- other endpoints: passthrough to real iLink ------------------
def _proxy_passthrough(self, body):
ep = urlparse(self.path).path.lstrip("/")
self._forward_to_ilink(ep, body)
def _forward_to_ilink(self, ep, body):
url = ilink_base_url.rstrip("/") + "/" + ep
try:
resp = requests.post(
url,
headers={
"Content-Type": "application/json",
"AuthorizationType": "ilink_bot_token",
"iLink-App-Id": "",
"iLink-App-ClientVersion": ILINK_CV,
"Authorization": "Bearer " + ilink_token,
},
data=body,
timeout=30,
)
self.send_response(resp.status_code)
for k, v in resp.headers.items():
if k.lower() not in (
"transfer-encoding", "content-encoding", "connection",
):
self.send_header(k, v)
self.end_headers()
self.wfile.write(resp.content)
except BrokenPipeError:
# Gateway disconnected before we could write back the response.
# The upstream request already succeeded; nothing to retry.
log.debug("BrokenPipe on write-back (benign): %s", ep)
except Exception as e:
log.error("Proxy forward error: %s", e)
try:
err = json.dumps({"ret": -1, "errmsg": str(e)}).encode()
self.send_response(502)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(err)))
self.end_headers()
self.wfile.write(err)
except BrokenPipeError:
log.debug("BrokenPipe writing error response (benign)")
def _write_json(self, code, obj):
data = json.dumps(obj).encode()
self.send_response(code)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
self.wfile.write(data)
def log_message(self, format, *args):
pass # Suppress default HTTP log noise.
return GatewayProxyHandler
# ---------------------------------------------------------------------------
# Message routing
# ---------------------------------------------------------------------------
def route_message(uid, msg, route, hermes_q, openclaw_q, opencode_q=None):
"""Enqueue *msg* to the correct proxy queue(s) based on *route*."""
if route in (Route.HERMES, Route.BOTH, Route.THREE):
if hermes_q:
hermes_q.enqueue(msg)
else:
log.warning("Hermes queue not available for %s", uid[:16])
if route in (Route.OPENCLAW, Route.BOTH, Route.THREE):
if openclaw_q:
openclaw_q.enqueue(msg)
else:
log.warning("OpenClaw queue not available for %s", uid[:16])
if route in (Route.OPENCODE, Route.THREE):
if opencode_q is not None:
opencode_q.put((uid, msg))
else:
log.warning("OpenCode queue not available for %s", uid[:16])
def opencode_worker(bridge, q, base_url, token, state):
"""Dedicated thread: dequeue messages and send them to OpenCode."""
while True:
try:
uid, msg = q.get()
items = msg.get("item_list", [])
blocks = build_opencode_prompt(items, token)
route = state.get(uid)
if not blocks:
continue
stop_typing = threading.Event()
typing_thread = threading.Thread(
target=keep_typing_ilink,
args=(base_url, token, uid, msg.get("context_token"), stop_typing),
daemon=True,
)
typing_thread.start()
try:
reply = bridge.send_blocks(uid, blocks)
if reply:
tag = "[OpenCode] " if route == Route.THREE else ""
send_text_ilink(
base_url, token, uid,
tag + reply,
msg.get("context_token"),
)
except Exception as e:
log.error("OpenCode error for %s: %s", uid[:16], e)
try:
send_text_ilink(base_url, token, uid,
f"[OpenCode error: {e}]",
msg.get("context_token"))
except Exception:
pass
finally:
stop_typing.set()
except Exception as e:
log.error("opencode_worker: %s", e, exc_info=True)
def proc_msg(msg, state, base_url, token, hermes_q, openclaw_q,
opencode_q=None, opencode_bridge=None):
"""Process one inbound iLink message."""
uid = msg.get("from_user_id", "")
ctx = msg.get("context_token", "")
items = msg.get("item_list", [])
if msg.get("message_type", 1) != 1:
return
item_types = ",".join(str(it.get("type", 0)) for it in items)
log.info("Msg from=%s... items=%d types=%s", uid[:16], len(items), item_types)
txt = extract_text(items)
has_any = txt or any(it.get("type", 0) != 0 for it in items)
if not has_any:
return
# Show status on first contact.
if state.should_show_status(uid) and not txt.startswith("/"):
reply = cmd(state, uid, "/whoami", opencode_bridge)
send_text_ilink(base_url, token, uid, reply, ctx)
state.mark_status_shown(uid)
# Slash commands are text-only; never forwarded to gateways.
if txt.startswith("/"):
r = cmd(state, uid, txt, opencode_bridge)
if r:
send_text_ilink(base_url, token, uid, r, ctx)
return
route = state.get(uid)
route_message(uid, msg, route, hermes_q, openclaw_q, opencode_q)
# ---------------------------------------------------------------------------
# Poll loop
# ---------------------------------------------------------------------------
MAX_FAILS = 3
BACKOFF = 30
def poll_loop(base_url, token, state, hermes_q, openclaw_q,
opencode_q=None, opencode_bridge=None, poll_sec=None):
if poll_sec is None:
poll_sec = DEFAULT_POLL_SEC
buf = ""
fails = 0
while True:
try:
resp = get_updates_real(base_url, token, buf, poll_sec)
ret, ec = resp.get("ret"), resp.get("errcode")
if ret not in (0, None) or ec not in (0, None):
log.warning("getUpdates err: ret=%s ec=%s", ret, ec)
fails += 1
if fails >= MAX_FAILS:
time.sleep(BACKOFF)
fails = 0
else:
time.sleep(2)
continue
fails = 0
if resp.get("get_updates_buf"):
buf = resp["get_updates_buf"]
for m in resp.get("msgs", []):
try:
proc_msg(m, state, base_url, token, hermes_q, openclaw_q,
opencode_q, opencode_bridge)
except Exception as e:
log.error("proc: %s", e, exc_info=True)
except Exception as e:
log.error("Loop: %s", e, exc_info=True)
fails += 1
if fails >= MAX_FAILS:
time.sleep(BACKOFF)
fails = 0
else:
time.sleep(2)
# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------
def main():
from dotenv import load_dotenv
load_dotenv(Path(__file__).parent / ".env")
base_url = os.getenv("ILINK_BASE_URL", "https://ilinkai.weixin.qq.com")
token = os.getenv("ILINK_TOKEN", "")
hermes_port = int(os.getenv("HERMES_PROXY_PORT", "19998"))
oc_port = int(os.getenv("OPENCLAW_PROXY_PORT", "19999"))
state_file = os.getenv("STATE_FILE", str(Path(__file__).parent / "router_state.json"))
log_file = os.getenv("LOG_FILE", str(Path(__file__).parent / "hermesclaw.log"))
poll_sec = int(os.getenv("LONG_POLL_TIMEOUT", "35"))
hermes_on = os.getenv("HERMES_ENABLED", "true").lower() in ("true", "1", "yes")
oc_on = os.getenv("OPENCLAW_ENABLED", "true").lower() in ("true", "1", "yes")
opencode_on = os.getenv("OPENCODE_ENABLED", "true").lower() in ("true", "1", "yes")
opencode_model = os.getenv("OPENCODE_MODEL", "opencode/minimax-m2.5-free")
opencode_cmd = os.getenv("OPENCODE_CMD", "/home/ubuntu/.npm-global/bin/opencode")
opencode_cwd = os.getenv("OPENCODE_CWD", str(Path.home()))
opencode_permission_strategy = os.getenv("OPENCODE_PERMISSION_STRATEGY", "allow_always")
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
handlers=[
logging.StreamHandler(sys.stdout),
logging.FileHandler(log_file, encoding="utf-8"),
],
)
if not token:
log.error("No ILINK_TOKEN set!")
sys.exit(1)
state = State(state_file)
hermes_q = MessageQueue() if hermes_on else None
oc_q = MessageQueue() if oc_on else None
opencode_q = None
opencode_bridge = None
log.info("=" * 60)
log.info("HermesClaw v4 -- triple gateway proxy")
log.info("iLink: %s", base_url)
log.info("Hermes proxy: :%d (enabled=%s)", hermes_port, hermes_on)
log.info("OpenClaw proxy: :%d (enabled=%s)", oc_port, oc_on)
log.info("OpenCode bridge: enabled=%s", opencode_on)
log.info("Default route: %s", Route.HERMES.value)
log.info("=" * 60)
servers = []
if hermes_on:
h_handler = make_proxy_handler(
hermes_q, base_url, token, state, tag="[Hermes Agent]",
)
h_srv = ThreadingHTTPServer(("127.0.0.1", hermes_port), h_handler)
threading.Thread(target=h_srv.serve_forever, daemon=True).start()
servers.append(h_srv)
log.info("Hermes proxy started on :%d", hermes_port)
if oc_on:
oc_handler = make_proxy_handler(
oc_q, base_url, token, state, tag="[OpenClaw]",
)
oc_srv = ThreadingHTTPServer(("127.0.0.1", oc_port), oc_handler)
threading.Thread(target=oc_srv.serve_forever, daemon=True).start()
servers.append(oc_srv)
log.info("OpenClaw proxy started on :%d", oc_port)
if opencode_on:
opencode_bridge = OpenCodeBridge(
opencode_cmd,
opencode_model,
opencode_cwd,
opencode_permission_strategy,
)
if opencode_bridge.is_available():
opencode_q = queue.Queue()
threading.Thread(
target=opencode_worker,
args=(opencode_bridge, opencode_q, base_url, token, state),
daemon=True,
).start()
log.info("OpenCode bridge ready (model=%s)", opencode_model)
else:
log.warning(
"OpenCode binary not found at %s; /opencode and /three will "
"prompt users to install it.", opencode_cmd,
)
poll_thread = threading.Thread(
target=poll_loop,
args=(base_url, token, state, hermes_q, oc_q, opencode_q, opencode_bridge, poll_sec),
daemon=True,
)
poll_thread.start()
stop = threading.Event()
signal.signal(signal.SIGINT, lambda s, f: stop.set())
signal.signal(signal.SIGTERM, lambda s, f: stop.set())
while not stop.is_set():
time.sleep(1)
log.info("Shutting down...")
if opencode_bridge:
opencode_bridge.close_all()
for s in servers:
s.shutdown()
log.info("Stopped")
if __name__ == "__main__":
main()