Update 2026-07-14 11:17:55

agents/g1/sanad_api_g1.py

@@ -148,6 +148,9 @@ class Config:
     remote_ports: str
     remote_url: str
     remote_interval: float
+    ssh_enable: bool
+    ssh_user: str
+    ssh_port: int
     # project logs (e.g. the robot's Sanad app) shipped alongside agent logs
     project_log_container: str
     project_log_path: str
@@ -210,6 +213,9 @@ class Config:
             remote_ports=_env("REMOTE_PORTS", "8001,8014,8011,8012,8013,8000,8080"),
             remote_url=_env("REMOTE_URL", ""),   # explicit URL (e.g. a public tunnel) wins
             remote_interval=float(_env("REMOTE_INTERVAL", "60")),
+            ssh_enable=_env_bool("SSH_REGISTER", True),
+            ssh_user=_env("SSH_USER", "unitree"),
+            ssh_port=int(_env("SSH_PORT", "22")),
             project_log_container=_env("PROJECT_LOG_CONTAINER", "auto"),
             project_log_path=_env("PROJECT_LOG_PATH", ""),
             project_log_label=_env("PROJECT_LOG_LABEL", ""),
@@ -1318,7 +1324,8 @@ def logs_loop(cfg: Config, session: requests.Session) -> None:
 # fleet dashboard to embed ("full dashboard through the fleet API"). No changes
 # to the Sanad app: we only probe its port and POST the URL to /{sn}/remote.
 # --------------------------------------------------------------------------- #
-_REMOTE_STAT: Dict[str, Any] = {"url": None, "port": None, "kind": None, "ok": None}
+_REMOTE_STAT: Dict[str, Any] = {"url": None, "port": None, "kind": None, "ok": None,
+                                "ssh": None, "ssh_ok": None}
 
 
 def _primary_ip() -> str:
@@ -1353,27 +1360,49 @@ def discover_dashboard(cfg: Config) -> Optional[Dict[str, Any]]:
     return None
 
 
-def register_remote(cfg: Config, session: requests.Session) -> None:
+def _post_remote(cfg: Config, session: requests.Session, body: Dict[str, Any]) -> bool:
-    if not cfg.remote_enable:
-        return
-    d = discover_dashboard(cfg)
-    if not d:
-        _REMOTE_STAT.update(url=None, port=None, ok=None)
-        return
-    body = {"sn": cfg.sn, "name": cfg.name, "kind": cfg.remote_kind,
-            "url": d["url"], "label": f"{cfg.name} — Sanad Dashboard",
-            "port": d["port"], "ts": int(time.time())}
     try:
         r = session.post(cfg.remote_url_ep(), json=body, headers=cfg.auth_headers(),
                          timeout=cfg.http_timeout, verify=cfg.verify_tls)
-        _REMOTE_STAT.update(url=d["url"], port=d["port"], kind=cfg.remote_kind, ok=bool(r.ok))
+        if not r.ok:
-        if r.ok:
+            log.warning("remote(%s) register failed: HTTP %s %s",
-            log.info("remote dashboard registered: %s -> HTTP %s", d["url"], r.status_code)
+                        body.get("kind"), r.status_code, r.text[:120])
-        else:
+        return bool(r.ok)
-            log.warning("remote register failed: HTTP %s %s", r.status_code, r.text[:150])
     except requests.RequestException as e:
-        _REMOTE_STAT.update(url=d["url"], port=d["port"], kind=cfg.remote_kind, ok=False)
+        log.debug("remote(%s) register failed: %s", body.get("kind"), e)
-        log.debug("remote register failed: %s", e)
+        return False
+
+
+def register_remote(cfg: Config, session: requests.Session) -> None:
+    if not cfg.remote_enable:
+        return
+    host = cfg.remote_host or _primary_ip()
+    # 1) the Sanad dashboard (kind=web) — the exact UI, for the fleet to embed
+    d = discover_dashboard(cfg)
+    if d:
+        ok = _post_remote(cfg, session, {
+            "sn": cfg.sn, "name": cfg.name, "kind": cfg.remote_kind,
+            "url": d["url"], "label": f"{cfg.name} — Sanad Dashboard",
+            "port": d["port"], "ts": int(time.time())})
+        _REMOTE_STAT.update(url=d["url"], port=d["port"], kind=cfg.remote_kind, ok=ok)
+        if ok:
+            log.info("remote dashboard registered: %s -> HTTP 200", d["url"])
+    else:
+        _REMOTE_STAT.update(url=None, port=None, ok=None)
+    # 2) SSH access (kind=ssh) — "ssh unitree@<ip>"
+    if cfg.ssh_enable:
+        cmd = f"ssh {cfg.ssh_user}@{host}"
+        if cfg.ssh_port != 22:
+            cmd += f" -p {cfg.ssh_port}"
+        ok = _post_remote(cfg, session, {
+            "sn": cfg.sn, "name": cfg.name, "kind": "ssh",
+            "url": f"ssh://{cfg.ssh_user}@{host}:{cfg.ssh_port}",   # URL-valid form
+            "command": cmd,                                        # "ssh unitree@<ip>"
+            "host": host, "port": cfg.ssh_port, "user": cfg.ssh_user,
+            "label": f"{cfg.name} — SSH", "ts": int(time.time())})
+        _REMOTE_STAT.update(ssh=cmd, ssh_ok=ok)
+        if ok:
+            log.info("remote SSH registered: %s -> HTTP 200", cmd)
 
 
 def remote_loop(cfg: Config, session: requests.Session) -> None:

agents/go2/sanad_api_go2.py

@@ -130,6 +130,9 @@ class Config:
     remote_ports: str
     remote_url: str
     remote_interval: float
+    ssh_enable: bool
+    ssh_user: str
+    ssh_port: int
     # project logs (e.g. the robot's Sanad app) shipped alongside agent logs
     project_log_container: str
     project_log_path: str
@@ -188,6 +191,9 @@ class Config:
             remote_ports=_env("REMOTE_PORTS", "8001,8014,8011,8012,8013,8000,8080"),
             remote_url=_env("REMOTE_URL", ""),
             remote_interval=float(_env("REMOTE_INTERVAL", "60")),
+            ssh_enable=_env_bool("SSH_REGISTER", True),
+            ssh_user=_env("SSH_USER", "unitree"),
+            ssh_port=int(_env("SSH_PORT", "22")),
             project_log_container=_env("PROJECT_LOG_CONTAINER", "auto"),
             project_log_path=_env("PROJECT_LOG_PATH", ""),
             project_log_label=_env("PROJECT_LOG_LABEL", ""),
@@ -1234,7 +1240,8 @@ def logs_loop(cfg: Config, session: requests.Session) -> None:
 # --------------------------------------------------------------------------- #
 # remote dashboard (discover Sanad web UI, register its URL via /{sn}/remote)
 # --------------------------------------------------------------------------- #
-_REMOTE_STAT: Dict[str, Any] = {"url": None, "port": None, "kind": None, "ok": None}
+_REMOTE_STAT: Dict[str, Any] = {"url": None, "port": None, "kind": None, "ok": None,
+                                "ssh": None, "ssh_ok": None}
 
 
 def _primary_ip() -> str:
@@ -1269,27 +1276,49 @@ def discover_dashboard(cfg: Config) -> Optional[Dict[str, Any]]:
     return None
 
 
-def register_remote(cfg: Config, session: requests.Session) -> None:
+def _post_remote(cfg: Config, session: requests.Session, body: Dict[str, Any]) -> bool:
-    if not cfg.remote_enable:
-        return
-    d = discover_dashboard(cfg)
-    if not d:
-        _REMOTE_STAT.update(url=None, port=None, ok=None)
-        return
-    body = {"sn": cfg.sn, "name": cfg.name, "kind": cfg.remote_kind,
-            "url": d["url"], "label": f"{cfg.name} — Sanad Dashboard",
-            "port": d["port"], "ts": int(time.time())}
     try:
         r = session.post(cfg.remote_url_ep(), json=body, headers=cfg.auth_headers(),
                          timeout=cfg.http_timeout, verify=cfg.verify_tls)
-        _REMOTE_STAT.update(url=d["url"], port=d["port"], kind=cfg.remote_kind, ok=bool(r.ok))
+        if not r.ok:
-        if r.ok:
+            log.warning("remote(%s) register failed: HTTP %s %s",
-            log.info("remote dashboard registered: %s -> HTTP %s", d["url"], r.status_code)
+                        body.get("kind"), r.status_code, r.text[:120])
-        else:
+        return bool(r.ok)
-            log.warning("remote register failed: HTTP %s %s", r.status_code, r.text[:150])
     except requests.RequestException as e:
-        _REMOTE_STAT.update(url=d["url"], port=d["port"], kind=cfg.remote_kind, ok=False)
+        log.debug("remote(%s) register failed: %s", body.get("kind"), e)
-        log.debug("remote register failed: %s", e)
+        return False
+
+
+def register_remote(cfg: Config, session: requests.Session) -> None:
+    if not cfg.remote_enable:
+        return
+    host = cfg.remote_host or _primary_ip()
+    # 1) the Sanad dashboard (kind=web) — the exact UI, for the fleet to embed
+    d = discover_dashboard(cfg)
+    if d:
+        ok = _post_remote(cfg, session, {
+            "sn": cfg.sn, "name": cfg.name, "kind": cfg.remote_kind,
+            "url": d["url"], "label": f"{cfg.name} — Sanad Dashboard",
+            "port": d["port"], "ts": int(time.time())})
+        _REMOTE_STAT.update(url=d["url"], port=d["port"], kind=cfg.remote_kind, ok=ok)
+        if ok:
+            log.info("remote dashboard registered: %s -> HTTP 200", d["url"])
+    else:
+        _REMOTE_STAT.update(url=None, port=None, ok=None)
+    # 2) SSH access (kind=ssh) — "ssh unitree@<ip>"
+    if cfg.ssh_enable:
+        cmd = f"ssh {cfg.ssh_user}@{host}"
+        if cfg.ssh_port != 22:
+            cmd += f" -p {cfg.ssh_port}"
+        ok = _post_remote(cfg, session, {
+            "sn": cfg.sn, "name": cfg.name, "kind": "ssh",
+            "url": f"ssh://{cfg.ssh_user}@{host}:{cfg.ssh_port}",   # URL-valid form
+            "command": cmd,                                        # "ssh unitree@<ip>"
+            "host": host, "port": cfg.ssh_port, "user": cfg.ssh_user,
+            "label": f"{cfg.name} — SSH", "ts": int(time.time())})
+        _REMOTE_STAT.update(ssh=cmd, ssh_ok=ok)
+        if ok:
+            log.info("remote SSH registered: %s -> HTTP 200", cmd)
 
 
 def remote_loop(cfg: Config, session: requests.Session) -> None:

agents/r1/sanad_api_r1.py

@@ -148,6 +148,9 @@ class Config:
     remote_ports: str
     remote_url: str
     remote_interval: float
+    ssh_enable: bool
+    ssh_user: str
+    ssh_port: int
     # project logs (e.g. the robot's Sanad app) shipped alongside agent logs
     project_log_container: str
     project_log_path: str
@@ -210,6 +213,9 @@ class Config:
             remote_ports=_env("REMOTE_PORTS", "8001,8014,8011,8012,8013,8000,8080"),
             remote_url=_env("REMOTE_URL", ""),   # explicit URL (e.g. a public tunnel) wins
             remote_interval=float(_env("REMOTE_INTERVAL", "60")),
+            ssh_enable=_env_bool("SSH_REGISTER", True),
+            ssh_user=_env("SSH_USER", "unitree"),
+            ssh_port=int(_env("SSH_PORT", "22")),
             project_log_container=_env("PROJECT_LOG_CONTAINER", "auto"),
             project_log_path=_env("PROJECT_LOG_PATH", ""),
             project_log_label=_env("PROJECT_LOG_LABEL", ""),
@@ -1318,7 +1324,8 @@ def logs_loop(cfg: Config, session: requests.Session) -> None:
 # fleet dashboard to embed ("full dashboard through the fleet API"). No changes
 # to the Sanad app: we only probe its port and POST the URL to /{sn}/remote.
 # --------------------------------------------------------------------------- #
-_REMOTE_STAT: Dict[str, Any] = {"url": None, "port": None, "kind": None, "ok": None}
+_REMOTE_STAT: Dict[str, Any] = {"url": None, "port": None, "kind": None, "ok": None,
+                                "ssh": None, "ssh_ok": None}
 
 
 def _primary_ip() -> str:
@@ -1353,27 +1360,49 @@ def discover_dashboard(cfg: Config) -> Optional[Dict[str, Any]]:
     return None
 
 
-def register_remote(cfg: Config, session: requests.Session) -> None:
+def _post_remote(cfg: Config, session: requests.Session, body: Dict[str, Any]) -> bool:
-    if not cfg.remote_enable:
-        return
-    d = discover_dashboard(cfg)
-    if not d:
-        _REMOTE_STAT.update(url=None, port=None, ok=None)
-        return
-    body = {"sn": cfg.sn, "name": cfg.name, "kind": cfg.remote_kind,
-            "url": d["url"], "label": f"{cfg.name} — Sanad Dashboard",
-            "port": d["port"], "ts": int(time.time())}
     try:
         r = session.post(cfg.remote_url_ep(), json=body, headers=cfg.auth_headers(),
                          timeout=cfg.http_timeout, verify=cfg.verify_tls)
-        _REMOTE_STAT.update(url=d["url"], port=d["port"], kind=cfg.remote_kind, ok=bool(r.ok))
+        if not r.ok:
-        if r.ok:
+            log.warning("remote(%s) register failed: HTTP %s %s",
-            log.info("remote dashboard registered: %s -> HTTP %s", d["url"], r.status_code)
+                        body.get("kind"), r.status_code, r.text[:120])
-        else:
+        return bool(r.ok)
-            log.warning("remote register failed: HTTP %s %s", r.status_code, r.text[:150])
     except requests.RequestException as e:
-        _REMOTE_STAT.update(url=d["url"], port=d["port"], kind=cfg.remote_kind, ok=False)
+        log.debug("remote(%s) register failed: %s", body.get("kind"), e)
-        log.debug("remote register failed: %s", e)
+        return False
+
+
+def register_remote(cfg: Config, session: requests.Session) -> None:
+    if not cfg.remote_enable:
+        return
+    host = cfg.remote_host or _primary_ip()
+    # 1) the Sanad dashboard (kind=web) — the exact UI, for the fleet to embed
+    d = discover_dashboard(cfg)
+    if d:
+        ok = _post_remote(cfg, session, {
+            "sn": cfg.sn, "name": cfg.name, "kind": cfg.remote_kind,
+            "url": d["url"], "label": f"{cfg.name} — Sanad Dashboard",
+            "port": d["port"], "ts": int(time.time())})
+        _REMOTE_STAT.update(url=d["url"], port=d["port"], kind=cfg.remote_kind, ok=ok)
+        if ok:
+            log.info("remote dashboard registered: %s -> HTTP 200", d["url"])
+    else:
+        _REMOTE_STAT.update(url=None, port=None, ok=None)
+    # 2) SSH access (kind=ssh) — "ssh unitree@<ip>"
+    if cfg.ssh_enable:
+        cmd = f"ssh {cfg.ssh_user}@{host}"
+        if cfg.ssh_port != 22:
+            cmd += f" -p {cfg.ssh_port}"
+        ok = _post_remote(cfg, session, {
+            "sn": cfg.sn, "name": cfg.name, "kind": "ssh",
+            "url": f"ssh://{cfg.ssh_user}@{host}:{cfg.ssh_port}",   # URL-valid form
+            "command": cmd,                                        # "ssh unitree@<ip>"
+            "host": host, "port": cfg.ssh_port, "user": cfg.ssh_user,
+            "label": f"{cfg.name} — SSH", "ts": int(time.time())})
+        _REMOTE_STAT.update(ssh=cmd, ssh_ok=ok)
+        if ok:
+            log.info("remote SSH registered: %s -> HTTP 200", cmd)
 
 
 def remote_loop(cfg: Config, session: requests.Session) -> None: