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

feat(cli): hermes connectors lists, connects, renames and removes accounts

Browse Source 
One app can now hold several named accounts. The CLI lists them (retired
ones folded behind --all), connects a new named account and waits for the
browser step, renames by name or app label, and removes one account after
a confirmation. Ctrl-C settles the open operation so the next connect of
that app is not blocked.
This commit is contained in:
Siddharth Balyan
2026-10-08 19:18:50 +05:30
committed by Siddharth Balyan
parent 08c049a0c0
commit af530e86d6
5 changed files with 339 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

238
hermes_cli/connectors_cmd.py Normal file
View File

@@ -0,0 +1,238 @@
"""``hermes connectors``: list, connect, rename and remove hosted connector accounts."""
from __future__ import annotations
import re
import sys
import time
from typing import Any, Callable, NoReturn
_POLL_SECONDS = 2.0
_DONE_STATES = frozenset({"connected", "skipped", "failed", "expired", "not_connected"})
_SIGN_IN = "Sign in with `hermes login` to manage connectors."
class _CommandError(Exception):
"""A user-facing failure: the message is printed and the command exits 1."""
def _fail(message: str) -> NoReturn:
raise _CommandError(message)
def _valid_alias(alias: str) -> str:
from tools.connectors.targets import ALIAS_PATTERN
if not re.match(ALIAS_PATTERN, alias):
_fail(f"Invalid name {alias!r}. Use 1 to 32 lowercase letters, digits or hyphens, "
"starting with a letter or digit.")
return alias
def _account_name(row: dict[str, Any]) -> str:
return str(row.get("alias") or row.get("label") or row.get("connectionId") or "")
def _display_name(row: dict[str, Any]) -> str:
return row["alias"] if row.get("alias") else f"{row.get('label') or row.get('connectionId')} (unnamed)"
def _app_accounts(app: str) -> list[dict[str, Any]]:
from tools.connectors.portal.client import PortalConnectorClient
return [r for r in PortalConnectorClient().list_accounts() if str(r.get("connector", "")).lower() == app]
def _names_hint(app: str, rows: list[dict[str, Any]]) -> str:
names = sorted({_account_name(r) for r in rows})
return f"Its accounts are: {', '.join(names)}." if names else f"You have no {app} account."
def _match_named(app: str, rows: list[dict[str, Any]], name: str) -> dict[str, Any]:
matches = [r for r in rows if r.get("alias") == name] or [
r for r in rows if not r.get("alias") and r.get("label") == name]
if len(matches) != 1:
what = "No" if not matches else "More than one"
_fail(f"{what} {app} account is called {name!r}. {_names_hint(app, rows)}")
return matches[0]
def _list(args) -> int:
from tools.connectors.managed import managed_client
rows = managed_client().list_connectors()
if not rows:
print("No connectors are available.")
return 0
width = max(len(str(r.get("connector", ""))) for r in rows)
for row in sorted(rows, key=lambda r: str(r.get("connector", ""))):
state = "connected" if row.get("connected") else "not connected"
print(f"{str(row.get('connector', '')).ljust(width)} {state}")
return 0
def _accounts(args) -> int:
from tools.connectors.portal.client import PortalConnectorClient
app = (args.app or "").strip().lower()
rows = [r for r in PortalConnectorClient().list_accounts()
if not app or str(r.get("connector", "")).lower() == app]
retired = [r for r in rows if r.get("disabled")]
shown = rows if args.all else [r for r in rows if not r.get("disabled")]
if not shown and not retired:
print(f"You have no {app} account." if app else "You have no connected accounts.")
return 0
table = [(str(r.get("connector", "")), _display_name(r), str(r.get("status", "")) + (" (retired)" if r.get("disabled") else ""))
for r in sorted(shown, key=lambda r: (str(r.get("connector", "")), _account_name(r)))]
if table:
app_width = max(len(row[0]) for row in table)
name_width = max(len(row[1]) for row in table)
for connector, name, status in table:
print(f"{connector.ljust(app_width)} {name.ljust(name_width)} {status}")
if retired and not args.all:
print(f"{len(retired)} retired (use --all to list)")
return 0
def _wait_for_link(start, app: str) -> dict[str, Any]:
"""Poll the operation until its one target settles; returns the last target snapshot."""
printed_url = False
while True:
snapshot = start.operation.snapshot()
target = snapshot["targets"][0]
url = target.get("connect_url")
if url and not printed_url:
print(url)
print("Open this link in your browser to finish.")
printed_url = True
if target.get("state") in _DONE_STATES or snapshot.get("settled_at") is not None:
return target
if time.time() >= start.operation.deadline_at:
return target
time.sleep(_POLL_SECONDS)
def _report_connect(app: str, alias: str | None, target: dict[str, Any]) -> int:
from tools.tool_labels import app_title
state = target.get("state")
if state == "connected":
if alias:
print(f"{app_title(app)} account {alias} is connected. "
f"Rename it anytime: hermes connectors rename {app} {alias} --to <new-name>")
else:
print(f"{app_title(app)} is connected.")
return 0
if state == "skipped":
print(f"Skipped connecting {app_title(app)}.")
return 0
detail = target.get("detail") or f"{app_title(app)} did not finish connecting."
print(detail, file=sys.stderr)
return 1
def _connect(args) -> int:
from tools.connectors import account, live
from tools.connectors.contract import SettleReason
app = args.app.strip().lower()
alias = _valid_alias(args.alias) if args.alias else None
start = account.find_or_start_operation(
[app], action="reconnect" if args.reconnect else "connect", profile_home=None, alias=alias)
try:
account.wait_for_prepare(start)
if start.failed:
_fail(f"Could not start connecting {app}. Try again in a moment.")
target = _wait_for_link(start, app)
except KeyboardInterrupt:
# The operation would otherwise stay open and block the next connect of this app.
start.operation.settle(SettleReason.continue_)
live.close(start.operation)
print("\nStopped waiting.")
return 130
return _report_connect(app, alias, target)
def _rename(args) -> int:
from tools.connectors.gateway.errors import GatewayUnavailable, IdempotencyConflict, ToolGatewayError
from tools.connectors.portal.client import PortalConnectorClient
app = args.app.strip().lower()
new = _valid_alias(args.to)
row = _match_named(app, _app_accounts(app), args.name)
try:
PortalConnectorClient().rename_account(row["connectionId"], new)
except IdempotencyConflict:
_fail(f"The name {new} is already used by another {app} account.")
except GatewayUnavailable as exc:
if exc.code == "connection_not_found":
_fail("That account no longer exists.")
_fail("This Nous Portal cannot rename connector accounts yet.")
except ToolGatewayError as exc:
if exc.status == 405:
_fail("This Nous Portal cannot rename connector accounts yet.")
raise
print(f"Renamed {app} account {args.name} to {new}.")
return 0
def _pick_for_disconnect(app: str, alias: str | None) -> dict[str, Any]:
rows = _app_accounts(app)
if alias:
return _match_named(app, rows, alias)
live_rows = [r for r in rows if not r.get("disabled")]
if len(live_rows) == 1:
return live_rows[0]
if not live_rows:
_fail(f"You have no {app} account.")
_fail(f"You have more than one {app} account. Pass --alias to pick one. {_names_hint(app, live_rows)}")
def _disconnect(args) -> int:
from tools.connectors.portal.client import PortalConnectorClient
app = args.app.strip().lower()
row = _pick_for_disconnect(app, args.alias)
name = _account_name(row)
if not args.yes:
if not sys.stdin.isatty():
_fail("Refusing to remove an account without confirmation. Pass --yes to confirm.")
answer = input(f"Remove {app} account {name}? This signs Hermes out of it. [y/N] ").strip().lower()
if answer not in ("y", "yes"):
print("Nothing removed.")
return 0
PortalConnectorClient().delete_account(row["connectionId"])
print(f"Removed {app} account {name}.")
return 0
_ACTIONS: dict[str, Callable[[Any], int]] = {
"list": _list,
"ls": _list,
"accounts": _accounts,
"connect": _connect,
"rename": _rename,
"disconnect": _disconnect,
}
def cmd_connectors(args) -> int:
from tools.connectors import connectors_available
from tools.connectors.gateway.errors import GatewayAuthError, ToolGatewayError
handler = _ACTIONS.get(getattr(args, "connectors_action", None) or "list")
if handler is None:
print(f"Unknown connectors action: {args.connectors_action}", file=sys.stderr)
return 2
if not connectors_available():
print("Connectors are not available. Sign in with `hermes login` first.", file=sys.stderr)
return 1
try:
return handler(args)
except _CommandError as exc:
print(str(exc), file=sys.stderr)
except GatewayAuthError:
print(_SIGN_IN, file=sys.stderr)
except ToolGatewayError as exc:
print(str(exc), file=sys.stderr)
return 1

4
hermes_cli/main.py
View File

@@ -389,6 +389,7 @@ from hermes_cli.subcommands.skills import build_skills_parser
from hermes_cli.subcommands.pairing import build_pairing_parser
from hermes_cli.subcommands.plugins import build_plugins_parser
from hermes_cli.subcommands.mcp import build_mcp_parser
from hermes_cli.subcommands.connectors import build_connectors_parser
from hermes_cli.subcommands.claw import build_claw_parser
from hermes_cli.subcommands.vault import build_vault_parser
from hermes_cli.subcommands.moa import build_moa_parser
@@ -2812,7 +2813,7 @@ def cmd_console(args):
_BUILTIN_SUBCOMMANDS = frozenset(
{
"acp", "approvals", "auth", "backup", "bundles", "checkpoints", "claw", "codex-runtime", "completion",
"computer-use",
"computer-use", "connectors",
"config", "console", "cron", "curator", "dashboard", "serve", "debug", "doctor",
"dump", "egress", "fallback", "gateway", "hooks", "import", "import-agent", "insights",
"gui", "desktop", "kanban", "login", "logout", "logs", "lsp", "mcp", "memory", "migrate", "moa",
@@ -3452,6 +3453,7 @@ def _build_cli_parser():
build_tools_parser(subparsers, cmd_tools=cmd_tools)
build_computer_use_parser(subparsers)
build_mcp_parser(subparsers, cmd_mcp=cmd_mcp)
build_connectors_parser(subparsers)
build_sessions_parser(subparsers, cmd_sessions=_cmd_sessions_lazy)
build_insights_parser(subparsers, cmd_insights=cmd_insights)
build_usage_parser(subparsers)

2
hermes_cli/profiles.py
View File

@@ -274,7 +274,7 @@ _RESERVED_NAMES = frozenset({"hermes", "default", "test", "tmp", "root", "sudo"}
_HERMES_SUBCOMMANDS = frozenset({
"chat", "model", "gateway", "setup", "whatsapp", "login", "logout",
"status", "cron", "doctor", "dump", "config", "pairing", "skills", "tools",
"mcp", "sessions", "insights", "version", "update", "uninstall", "profile", "plugins", "honcho", "acp",
"mcp", "connectors", "sessions", "insights", "version", "update", "uninstall", "profile", "plugins", "honcho", "acp",
})

36
hermes_cli/subcommands/connectors.py Normal file
View File

@@ -0,0 +1,36 @@
"""``hermes connectors`` subcommand parser."""
from __future__ import annotations
def build_connectors_parser(subparsers) -> None:
"""Attach the ``connectors`` subcommand to ``subparsers``."""
from hermes_cli.connectors_cmd import cmd_connectors
parser = subparsers.add_parser(
"connectors", help="Manage hosted app connectors and their accounts",
description="List, connect, rename and remove the app accounts Hermes uses through Nous Portal.")
actions = parser.add_subparsers(dest="connectors_action")
actions.add_parser("list", aliases=["ls"], help="Show every connector and whether it is connected")
accounts = actions.add_parser("accounts", help="Show connected accounts")
accounts.add_argument("app", nargs="?", help="Only show accounts of this app")
accounts.add_argument("--all", action="store_true", help="Also show retired accounts")
connect = actions.add_parser("connect", help="Connect an app account in your browser")
connect.add_argument("app", help="App to connect, for example gmail")
connect.add_argument("--alias", help="Name for this account, for example work")
connect.add_argument("--reconnect", action="store_true", help="Repair an account that stopped working")
rename = actions.add_parser("rename", help="Rename an account")
rename.add_argument("app", help="App the account belongs to")
rename.add_argument("name", help="Current name or label of the account")
rename.add_argument("--to", required=True, help="New name for the account")
disconnect = actions.add_parser("disconnect", help="Remove an account")
disconnect.add_argument("app", help="App the account belongs to")
disconnect.add_argument("--alias", help="Name or label of the account to remove")
disconnect.add_argument("--yes", "-y", action="store_true", help="Skip the confirmation prompt")
parser.set_defaults(func=cmd_connectors)

61
tests/hermes_cli/test_connectors_cmd.py Normal file
View File

@@ -0,0 +1,61 @@
"""Behaviour contracts for ``hermes connectors``."""
from __future__ import annotations
import argparse
import pytest
from hermes_cli.connectors_cmd import cmd_connectors
from hermes_cli.subcommands.connectors import build_connectors_parser
def _parse(*argv: str) -> argparse.Namespace:
parser = argparse.ArgumentParser()
build_connectors_parser(parser.add_subparsers(dest="command"))
return parser.parse_args(list(argv))
@pytest.fixture
def portal(monkeypatch):
calls: list[tuple[str, str]] = []
class FakePortal:
def list_accounts(self):
return [
{"connectionId": "conn-work", "connector": "gmail", "status": "active", "label": "a@x.test",
"alias": "work", "active": True},
{"connectionId": "conn-home", "connector": "gmail", "status": "active", "label": "b@x.test",
"alias": "home", "active": True},
]
def rename_account(self, connection_id, alias):
calls.append((connection_id, alias))
return {"connectionId": connection_id, "connector": "gmail", "alias": alias, "label": "a@x.test"}
monkeypatch.setattr("tools.connectors.portal.client.PortalConnectorClient", FakePortal)
monkeypatch.setattr("tools.connectors.connectors_available", lambda: True, raising=False)
return calls
def test_rename_parses_app_name_and_new_name():
args = _parse("connectors", "rename", "gmail", "work", "--to", "home")
assert (args.connectors_action, args.app, args.name, args.to) == ("rename", "gmail", "work", "home")
assert args.func is cmd_connectors
def test_rename_targets_the_named_account(portal, capsys):
rc = cmd_connectors(_parse("connectors", "rename", "gmail", "work", "--to", "office"))
assert rc == 0
assert portal == [("conn-work", "office")]
assert "office" in capsys.readouterr().out
def test_rename_to_an_invalid_name_calls_nothing(portal, capsys):
rc = cmd_connectors(_parse("connectors", "rename", "gmail", "work", "--to", "Bad Name"))
assert rc == 1
assert portal == []
assert "lowercase" in capsys.readouterr().err