seb/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Telegram QR onboarding to dashboard

Browse Source
This commit is contained in:
Shannon Sands
2026-06-04 16:55:27 -07:00
committed by Teknium
parent 5300727a08
commit 2f0c8e90e6
7 changed files with 1156 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/web_server.py
+335 -2
View File
@@ -14,11 +14,14 @@ from contextlib import asynccontextmanager
import asyncio
import base64
import binascii
from dataclasses import dataclass
from datetime import datetime, timezone
import hmac
import importlib.util
import json
import logging
import os
import re
import secrets
import stat
import subprocess
@@ -26,6 +29,7 @@ import sys
import tempfile
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
@@ -557,6 +561,14 @@ class MessagingPlatformUpdate(BaseModel):
clear_env: List[str] = []
class TelegramOnboardingStart(BaseModel):
bot_name: Optional[str] = None
class TelegramOnboardingApply(BaseModel):
allowed_user_ids: List[str]
class AudioTranscriptionRequest(BaseModel):
data_url: str
mime_type: Optional[str] = None
@@ -3050,6 +3062,329 @@ def _write_platform_enabled(platform_id: str, enabled: bool) -> None:
save_config(config)
_TELEGRAM_ONBOARDING_DEFAULT_URL = "https://setup.hermes-agent.nousresearch.com"
_TELEGRAM_USER_ID_RE = re.compile(r"^\d+$")
@dataclass
class _TelegramOnboardingPairing:
poll_token: str
expires_at: str
expires_at_ts: float
bot_token: str | None = None
bot_username: str | None = None
owner_user_id: str | None = None
_telegram_onboarding_pairings: dict[str, _TelegramOnboardingPairing] = {}
_telegram_onboarding_lock = threading.RLock()
def _telegram_onboarding_base_url() -> str:
return (
os.getenv("TELEGRAM_ONBOARDING_URL", _TELEGRAM_ONBOARDING_DEFAULT_URL)
.strip()
.rstrip("/")
)
def _parse_expiry_ts(value: str) -> float:
try:
normalized = value.replace("Z", "+00:00")
parsed = datetime.fromisoformat(normalized)
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
return parsed.timestamp()
except Exception:
return time.time() + 600
def _prune_telegram_onboarding_pairings() -> None:
now = time.time()
expired = [
pairing_id
for pairing_id, record in _telegram_onboarding_pairings.items()
if record.expires_at_ts <= now
]
for pairing_id in expired:
_telegram_onboarding_pairings.pop(pairing_id, None)
def _normalize_telegram_user_id(value: Any) -> str | None:
normalized = str(value or "").strip()
if _TELEGRAM_USER_ID_RE.fullmatch(normalized):
return normalized
return None
def _telegram_onboarding_error_message(error: str, fallback: str) -> str:
return {
"not_found": "Telegram pairing was not found. Start a new setup.",
"expired": "Telegram setup expired. Start a new setup.",
"claimed": "Telegram setup was already claimed. Start a new setup.",
"unauthorized": "Telegram setup service rejected this request.",
"telegram_manager_bot_token_not_configured": "Telegram setup service is not configured.",
"telegram_token_fetch_failed": "Telegram could not finish bot setup. Try again.",
}.get(error, fallback)
def _telegram_onboarding_request_sync(
method: str,
path: str,
*,
body: dict[str, Any] | None = None,
bearer_token: str | None = None,
) -> dict[str, Any]:
data = None
headers = {"Accept": "application/json"}
if body is not None:
data = json.dumps(body).encode("utf-8")
headers["Content-Type"] = "application/json"
if bearer_token:
headers["Authorization"] = f"Bearer {bearer_token}"
request = urllib.request.Request(
f"{_telegram_onboarding_base_url()}{path}",
data=data,
headers=headers,
method=method,
)
try:
with urllib.request.urlopen(request, timeout=10) as response:
payload = response.read()
except urllib.error.HTTPError as exc:
payload = exc.read()
try:
parsed = json.loads(payload.decode("utf-8"))
except Exception:
parsed = {}
error = str(parsed.get("error") or parsed.get("status") or "")
detail = _telegram_onboarding_error_message(
error,
"Telegram setup service returned an error.",
)
status_code = 404 if exc.code == 404 else 502
if error in {"expired", "claimed"}:
status_code = 410
raise HTTPException(status_code=status_code, detail=detail) from exc
except Exception as exc:
raise HTTPException(
status_code=502,
detail="Telegram setup service is unavailable. Try again shortly.",
) from exc
try:
parsed = json.loads(payload.decode("utf-8"))
except Exception as exc:
raise HTTPException(
status_code=502,
detail="Telegram setup service returned an invalid response.",
) from exc
if not isinstance(parsed, dict):
raise HTTPException(
status_code=502,
detail="Telegram setup service returned an invalid response.",
)
return parsed
async def _telegram_onboarding_request(
method: str,
path: str,
*,
body: dict[str, Any] | None = None,
bearer_token: str | None = None,
) -> dict[str, Any]:
return await asyncio.to_thread(
_telegram_onboarding_request_sync,
method,
path,
body=body,
bearer_token=bearer_token,
)
@app.post("/api/messaging/telegram/onboarding/start")
async def start_telegram_onboarding(body: TelegramOnboardingStart):
bot_name = (body.bot_name or "Hermes Agent").strip() or "Hermes Agent"
payload = await _telegram_onboarding_request(
"POST",
"/v1/telegram/pairings",
body={"bot_name": bot_name},
)
pairing_id = str(payload.get("pairing_id") or "").strip()
poll_token = str(payload.get("poll_token") or "").strip()
expires_at = str(payload.get("expires_at") or "").strip()
deep_link = str(payload.get("deep_link") or "").strip()
qr_payload = str(payload.get("qr_payload") or deep_link).strip()
suggested_username = str(payload.get("suggested_username") or "").strip()
if not pairing_id or not poll_token or not expires_at or not deep_link:
raise HTTPException(
status_code=502,
detail="Telegram setup service returned an incomplete response.",
)
with _telegram_onboarding_lock:
_prune_telegram_onboarding_pairings()
_telegram_onboarding_pairings[pairing_id] = _TelegramOnboardingPairing(
poll_token=poll_token,
expires_at=expires_at,
expires_at_ts=_parse_expiry_ts(expires_at),
)
return {
"pairing_id": pairing_id,
"suggested_username": suggested_username,
"deep_link": deep_link,
"qr_payload": qr_payload,
"expires_at": expires_at,
}
@app.get("/api/messaging/telegram/onboarding/{pairing_id}")
async def get_telegram_onboarding_status(pairing_id: str):
with _telegram_onboarding_lock:
_prune_telegram_onboarding_pairings()
record = _telegram_onboarding_pairings.get(pairing_id)
if not record:
raise HTTPException(
status_code=404,
detail="Telegram setup session was not found. Start a new setup.",
)
if record.bot_token:
return {
"status": "ready",
"bot_username": record.bot_username,
"owner_user_id": record.owner_user_id,
"expires_at": record.expires_at,
}
poll_token = record.poll_token
payload = await _telegram_onboarding_request(
"GET",
f"/v1/telegram/pairings/{urllib.parse.quote(pairing_id, safe='')}",
bearer_token=poll_token,
)
status = str(payload.get("status") or "").strip()
if status == "waiting":
with _telegram_onboarding_lock:
current = _telegram_onboarding_pairings.get(pairing_id)
expires_at = current.expires_at if current else ""
return {"status": "waiting", "expires_at": expires_at}
if status == "ready":
bot_token = str(payload.get("token") or "").strip()
bot_username = str(payload.get("bot_username") or "").strip()
if not bot_token:
raise HTTPException(
status_code=502,
detail="Telegram setup service returned an incomplete response.",
)
owner_user_id = _normalize_telegram_user_id(payload.get("owner_user_id"))
with _telegram_onboarding_lock:
record = _telegram_onboarding_pairings.get(pairing_id)
if not record:
raise HTTPException(
status_code=404,
detail="Telegram setup session was not found. Start a new setup.",
)
record.bot_token = bot_token
record.bot_username = bot_username or None
record.owner_user_id = owner_user_id
return {
"status": "ready",
"bot_username": record.bot_username,
"owner_user_id": record.owner_user_id,
"expires_at": record.expires_at,
}
if status in {"expired", "claimed"}:
with _telegram_onboarding_lock:
_telegram_onboarding_pairings.pop(pairing_id, None)
raise HTTPException(
status_code=410,
detail=_telegram_onboarding_error_message(
status,
"Telegram setup is no longer available. Start a new setup.",
),
)
raise HTTPException(
status_code=502,
detail="Telegram setup service returned an unknown status.",
)
@app.post("/api/messaging/telegram/onboarding/{pairing_id}/apply")
async def apply_telegram_onboarding(
pairing_id: str, body: TelegramOnboardingApply
):
allowed_user_ids = []
seen = set()
for raw_id in body.allowed_user_ids:
normalized = _normalize_telegram_user_id(raw_id)
if not normalized:
raise HTTPException(
status_code=400,
detail="Allowed Telegram user IDs must be numeric.",
)
if normalized not in seen:
seen.add(normalized)
allowed_user_ids.append(normalized)
if not allowed_user_ids:
raise HTTPException(
status_code=400,
detail="Add at least one allowed Telegram user ID.",
)
with _telegram_onboarding_lock:
_prune_telegram_onboarding_pairings()
record = _telegram_onboarding_pairings.get(pairing_id)
if not record:
raise HTTPException(
status_code=404,
detail="Telegram setup session was not found. Start a new setup.",
)
bot_token = record.bot_token
bot_username = record.bot_username
if not bot_token:
raise HTTPException(
status_code=409,
detail="Telegram setup is not ready yet.",
)
try:
save_env_value("TELEGRAM_BOT_TOKEN", bot_token)
save_env_value("TELEGRAM_ALLOWED_USERS", ",".join(allowed_user_ids))
_write_platform_enabled("telegram", True)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
except Exception as exc:
_log.exception("Telegram onboarding apply failed")
raise HTTPException(
status_code=500,
detail="Failed to save Telegram setup.",
) from exc
with _telegram_onboarding_lock:
_telegram_onboarding_pairings.pop(pairing_id, None)
return {
"ok": True,
"platform": "telegram",
"bot_username": bot_username,
"needs_restart": True,
}
@app.delete("/api/messaging/telegram/onboarding/{pairing_id}")
async def cancel_telegram_onboarding(pairing_id: str):
with _telegram_onboarding_lock:
_telegram_onboarding_pairings.pop(pairing_id, None)
return {"ok": True}
@app.get("/api/messaging/platforms")
async def get_messaging_platforms():
env_on_disk = load_env()
@@ -7078,8 +7413,6 @@ async def get_models_analytics(days: int = 30):
# though uvicorn binds to 127.0.0.1.
# ---------------------------------------------------------------------------
import re
# PTY bridge is POSIX-only (depends on fcntl/termios/ptyprocess). On native
# Windows the import raises; catch and leave PtyBridge=None so the rest of
# the dashboard (sessions, jobs, metrics, config editor) still loads and the