feat(auth): let model-provider plugins own hermes auth add/status/logout/refresh
A `kind: model-provider` plugin can register a ProviderProfile and a custom client but could not contribute an interactive login flow: its module is imported by provider discovery, and the generic command-plugin loader skips model-provider manifests on purpose, so `register(ctx)` is not a supported way to add commands. The core login implementations are also provider-name keyed, so a standards-based device-code provider had to ship a second standalone command plugin just for login/status/logout. Add an optional ProviderProfile.auth_handler(action, args) seam and consult it first from the existing `hermes auth` actions (add/status/logout/refresh) — before the known-provider gate and before the credential pool: - truthy return = the provider owned the action; falsy = per-action fallback. - sync or async handlers (a returned awaitable is awaited). - a handler exception becomes a readable SystemExit naming provider + action. - no handler at all = byte-for-byte unchanged built-in behavior, including the existing "Unknown provider" exit. - a registered profile that does not handle the requested action now says so instead of reporting its provider as unknown. No new top-level command, no manifest surface; credentials stay provider-owned. Tests: tests/hermes_cli/test_provider_auth_seam.py drives the real `hermes auth` parser with a fixture model-provider plugin — dispatch + argument pass-through for all four actions, per-action fallback, async handler, handler failure, registry lookup failure, duplicate registration (last-writer-wins), and an untouched built-in provider. Docs gain the new hook under the model-provider plugin guide.
This commit is contained in:
finn763
Teknium
@@ -3,6 +3,8 @@
|
||||
from __future__ import annotations
|
||||
from hermes_cli.cli_output import line_input
|
||||
|
||||
import asyncio
|
||||
import inspect
|
||||
import math
|
||||
import sys
|
||||
import time
|
||||
@@ -87,6 +89,83 @@ def _normalize_provider(provider: str) -> str:
|
||||
or auth_mod._plugin_aliases().get(normalized) or normalized)
|
||||
|
||||
|
||||
# ── Provider-owned interactive auth ──────────────────────────────────────────
|
||||
# A `kind: model-provider` plugin can set ``ProviderProfile.auth_handler`` and own
|
||||
# the interactive login flow for its provider (device-code, OIDC, …) without
|
||||
# shipping a second standalone command plugin — model-provider manifests are
|
||||
# skipped by the generic command-plugin loader on purpose. The four actions
|
||||
# below consult that handler FIRST; a provider without one (or a handler that
|
||||
# declines an action by returning falsy) keeps the built-in behavior untouched.
|
||||
|
||||
def _provider_auth_handler(provider: str) -> tuple[Any, Callable | None]:
|
||||
"""Return ``(profile, handler)`` when *provider*'s plugin owns its own auth.
|
||||
|
||||
``(None, None)`` when no profile is registered for the name, ``(profile,
|
||||
None)`` when one is registered without an auth handler.
|
||||
"""
|
||||
try:
|
||||
from providers import get_provider_profile
|
||||
except Exception: # provider layer unavailable — built-in path only
|
||||
return None, None
|
||||
profile = get_provider_profile(provider)
|
||||
handler = profile.auth_handler if profile is not None else None
|
||||
return (profile, handler) if callable(handler) else (profile, None)
|
||||
|
||||
|
||||
def _await_from_sync(coro):
|
||||
"""Run a provider handler's coroutine from the sync CLI path.
|
||||
|
||||
Safe inside a running loop too (gateway/embedded callers) — same idiom as
|
||||
``agent.context_references.preprocess_context_references``.
|
||||
"""
|
||||
try:
|
||||
asyncio.get_running_loop()
|
||||
except RuntimeError:
|
||||
return asyncio.run(coro)
|
||||
import concurrent.futures
|
||||
with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool:
|
||||
return pool.submit(asyncio.run, coro).result()
|
||||
|
||||
|
||||
def _dispatch_provider_auth(action: str, args, provider: str) -> bool:
|
||||
"""Offer ``hermes auth <action> <provider>`` to the provider's auth handler.
|
||||
|
||||
Returns True when the provider owned the action (core prints nothing more
|
||||
and returns); False when the caller must run the built-in path. A handler
|
||||
that fails becomes a readable ``SystemExit`` naming provider and action
|
||||
instead of a raw traceback.
|
||||
"""
|
||||
_profile, handler = _provider_auth_handler(provider)
|
||||
if handler is None:
|
||||
return False
|
||||
try:
|
||||
result = handler(action, args)
|
||||
if inspect.isawaitable(result):
|
||||
result = _await_from_sync(result)
|
||||
except SystemExit:
|
||||
raise
|
||||
except Exception as exc:
|
||||
raise SystemExit(
|
||||
f"{provider} auth handler failed for `{action}`: {type(exc).__name__}: {exc}"
|
||||
) from exc
|
||||
return bool(result)
|
||||
|
||||
|
||||
def _unknown_provider_error(provider: str, action: str) -> SystemExit:
|
||||
"""The existing ``Unknown provider`` exit, with a hint for plugin providers.
|
||||
|
||||
A registered model-provider profile that reaches this point offers no
|
||||
``auth_handler`` for the requested action (or declined it) — say so rather
|
||||
than pretending the provider is unknown.
|
||||
"""
|
||||
profile, _handler = _provider_auth_handler(provider)
|
||||
if profile is not None:
|
||||
return SystemExit(
|
||||
f"Unknown provider: {provider} — the `{profile.name}` model-provider plugin does "
|
||||
f"not provide auth handling for `hermes auth {action} {provider}`.")
|
||||
return _unknown_provider_exit(provider)
|
||||
|
||||
|
||||
def _migrate_legacy_custom_pool_key(provider: str, legacy_key: str) -> None:
|
||||
"""Move a keyed provider's old ``custom:`` pool into its runtime slug."""
|
||||
with auth_mod._auth_store_lock():
|
||||
@@ -374,9 +453,11 @@ def _add_api_key_credential(args, provider: str, pool) -> PooledCredential:
|
||||
|
||||
def auth_add_command(args) -> None:
|
||||
provider = _normalize_provider(getattr(args, "provider", ""))
|
||||
if _dispatch_provider_auth("add", args, provider):
|
||||
return
|
||||
configured_provider = _configured_provider_entry(provider)
|
||||
if not _is_known_provider(provider, configured_provider):
|
||||
raise _unknown_provider_exit(provider)
|
||||
raise _unknown_provider_error(provider, "add")
|
||||
if configured_provider is not None:
|
||||
_migrate_legacy_custom_pool_key(provider, configured_provider["pool_key"])
|
||||
|
||||
@@ -600,6 +681,8 @@ def auth_refresh_command(args) -> None:
|
||||
429s and benches it again. Failure leaves the pool's own verdict in place.
|
||||
"""
|
||||
provider = _normalize_provider(getattr(args, "provider", ""))
|
||||
if _dispatch_provider_auth("refresh", args, provider):
|
||||
return
|
||||
target = getattr(args, "target", None)
|
||||
pool = load_pool(provider)
|
||||
entries = pool.entries()
|
||||
@@ -648,6 +731,8 @@ def auth_status_command(args) -> None:
|
||||
provider = _normalize_provider(getattr(args, "provider", "") or "")
|
||||
if not provider:
|
||||
raise SystemExit("Provider is required. Example: `hermes auth status spotify`.")
|
||||
if _dispatch_provider_auth("status", args, provider):
|
||||
return
|
||||
status = auth_mod.get_auth_status(provider)
|
||||
if status.get("free_tier"):
|
||||
# Free tier: not an account login, so no account fields; point at the upgrade path.
|
||||
@@ -669,7 +754,12 @@ def auth_status_command(args) -> None:
|
||||
|
||||
|
||||
def auth_logout_command(args) -> None:
|
||||
auth_mod.logout_command(SimpleNamespace(provider=getattr(args, "provider", None)))
|
||||
# The built-in path keeps receiving the raw provider id (byte-for-byte
|
||||
# unchanged); the normalized alias is used only for the handler lookup.
|
||||
raw_provider = getattr(args, "provider", None)
|
||||
if _dispatch_provider_auth("logout", args, _normalize_provider(raw_provider or "")):
|
||||
return
|
||||
auth_mod.logout_command(SimpleNamespace(provider=raw_provider))
|
||||
|
||||
|
||||
def auth_spotify_command(args) -> None:
|
||||
@@ -776,9 +866,11 @@ def _pick_provider(prompt: str = "Provider") -> str:
|
||||
|
||||
def _interactive_add() -> None:
|
||||
provider = _pick_provider("Provider to add credential for")
|
||||
if _dispatch_provider_auth("add", SimpleNamespace(provider=provider), provider):
|
||||
return
|
||||
configured_provider = _configured_provider_entry(provider)
|
||||
if not _is_known_provider(provider, configured_provider):
|
||||
raise _unknown_provider_exit(provider)
|
||||
raise _unknown_provider_error(provider, "add")
|
||||
|
||||
auth_type = "api_key"
|
||||
if provider in _OAUTH_CAPABLE_PROVIDERS:
|
||||
|
||||
@@ -58,6 +58,16 @@ class ProviderProfile:
|
||||
# False → fetch_models returns None without a network call (catalog comes from an SDK/subprocess).
|
||||
supports_model_listing: bool = True
|
||||
|
||||
# ── Provider-owned interactive auth (optional) ────────────
|
||||
# Lets a `kind: model-provider` plugin own its own login flow instead of
|
||||
# shipping a second command plugin. When set, `hermes auth <action> <name>`
|
||||
# (action ∈ add|status|logout|refresh) calls it FIRST as
|
||||
# ``auth_handler(action, args)`` — ``args`` is the parsed CLI namespace — and
|
||||
# only falls back to the built-in credential-pool flow when the handler
|
||||
# returns falsy. Sync or async (a returned awaitable is awaited). The
|
||||
# handler owns its own credential storage; Hermes passes no secrets to it.
|
||||
auth_handler: Any = None
|
||||
|
||||
# ── Vision support ────────────────────────────────────────
|
||||
# True when the provider's API accepts image content inside
|
||||
# tool-result messages natively. Set on providers that expose
|
||||
|
||||
283
tests/hermes_cli/test_provider_auth_seam.py
Normal file
283
tests/hermes_cli/test_provider_auth_seam.py
Normal file
@@ -0,0 +1,283 @@
|
||||
"""`hermes auth <action> <provider>` prefers a model-provider plugin's auth handler.
|
||||
|
||||
The seam: ``ProviderProfile.auth_handler`` (an optional callable on the profile a
|
||||
``kind: model-provider`` plugin registers). These tests drive the real
|
||||
``hermes auth`` subcommand surface through the real argparse definitions, so they
|
||||
fail if the dispatch is dropped, reordered after the built-in paths, or stops
|
||||
passing the parsed arguments through.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
# A model-provider plugin that owns its own interactive auth. It appends one JSON
|
||||
# record per dispatch so the test can prove the action + arguments arrived.
|
||||
_PLUGIN_SOURCE = '''\
|
||||
"""Fixture provider plugin: owns its own interactive auth."""
|
||||
import json
|
||||
import os
|
||||
|
||||
from providers import register_provider
|
||||
from providers.base import ProviderProfile
|
||||
|
||||
|
||||
def _record(action, args):
|
||||
path = os.environ.get("FAKE_AUTH_LOG")
|
||||
if not path:
|
||||
return
|
||||
with open(path, "a", encoding="utf-8") as fh:
|
||||
fh.write(json.dumps({
|
||||
"action": action,
|
||||
"provider": getattr(args, "provider", None),
|
||||
"label": getattr(args, "label", None),
|
||||
"target": getattr(args, "target", None),
|
||||
"api_key": getattr(args, "api_key", None)}) + "\\n")
|
||||
|
||||
|
||||
def handler(action, args):
|
||||
_record(action, args)
|
||||
if action in (os.environ.get("FAKE_AUTH_DECLINE") or "").split(","):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
register_provider(ProviderProfile(name="__NAME__", auth_handler=handler))
|
||||
'''
|
||||
|
||||
|
||||
def _rediscover() -> None:
|
||||
"""Point the next profile lookup at the (new) HERMES_HOME user plugin dir.
|
||||
|
||||
Only the discovery flag is cleared: bundled plugin modules stay in
|
||||
``sys.modules`` (so their profiles stay registered) while the user dir is
|
||||
rescanned for the fixture. Fixture modules are evicted so the next
|
||||
``_import_plugin_dir`` actually re-executes them.
|
||||
"""
|
||||
import providers as _pkg
|
||||
|
||||
_pkg._discovered = False
|
||||
for mod in [m for m in sys.modules if m.startswith("_hermes_user_provider")]:
|
||||
del sys.modules[mod]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def install_provider(tmp_path, monkeypatch):
|
||||
"""Write a model-provider plugin into an isolated HERMES_HOME and discover it."""
|
||||
installed: list[str] = []
|
||||
|
||||
def _install(name: str = "fake-auth", *, with_handler: bool = True,
|
||||
async_handler: bool = False, raises: bool = False,
|
||||
reinstall: bool = False) -> Path:
|
||||
"""Write (or rewrite) the fixture plugin and re-run discovery."""
|
||||
plugin_dir = tmp_path / "hermes" / "plugins" / "model-providers" / name
|
||||
plugin_dir.mkdir(parents=True, exist_ok=True)
|
||||
(plugin_dir / "plugin.yaml").write_text(
|
||||
f"name: {name}\nkind: model-provider\nversion: 0.0.1\n"
|
||||
"description: provider auth seam fixture\n", encoding="utf-8")
|
||||
source = _PLUGIN_SOURCE.replace("__NAME__", name)
|
||||
if not with_handler:
|
||||
source = source.replace(", auth_handler=handler", "")
|
||||
if async_handler:
|
||||
source = source.replace("def handler(action, args):", "async def handler(action, args):")
|
||||
if raises:
|
||||
source = source.replace(" return True\n", ' raise ValueError("device flow exploded")\n')
|
||||
(plugin_dir / "__init__.py").write_text(source, encoding="utf-8")
|
||||
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes"))
|
||||
monkeypatch.setenv("FAKE_AUTH_LOG", str(tmp_path / "auth-log.jsonl"))
|
||||
_rediscover()
|
||||
installed.append(name)
|
||||
return plugin_dir
|
||||
|
||||
yield _install
|
||||
|
||||
# The provider registry is process-global: never leak the fixture profile.
|
||||
import providers as _pkg
|
||||
|
||||
for name in installed:
|
||||
_pkg._REGISTRY.pop(name, None)
|
||||
for alias, canonical in list(_pkg._ALIASES.items()):
|
||||
if canonical == name:
|
||||
_pkg._ALIASES.pop(alias, None)
|
||||
_pkg._PROVIDER_LIST_CACHE = None
|
||||
|
||||
|
||||
def _parse_auth_args(argv: list[str]) -> argparse.Namespace:
|
||||
"""Parse `hermes auth <argv>` through the real subcommand parser."""
|
||||
from hermes_cli.subcommands.auth import build_auth_parser
|
||||
|
||||
parser = argparse.ArgumentParser()
|
||||
subparsers = parser.add_subparsers(dest="command")
|
||||
build_auth_parser(subparsers, cmd_auth=lambda args: None)
|
||||
return parser.parse_args(["auth", *argv])
|
||||
|
||||
|
||||
def _log(tmp_path: Path) -> list[dict]:
|
||||
log = tmp_path / "auth-log.jsonl"
|
||||
if not log.exists():
|
||||
return []
|
||||
return [json.loads(line) for line in log.read_text(encoding="utf-8").splitlines()]
|
||||
|
||||
|
||||
def test_add_dispatches_to_provider_handler_with_arguments(tmp_path, install_provider):
|
||||
"""`hermes auth add <provider>` reaches the plugin handler, args included."""
|
||||
install_provider()
|
||||
|
||||
from hermes_cli.auth_commands import auth_command
|
||||
|
||||
args = _parse_auth_args(["add", "fake-auth", "--label", "work", "--api-key", "sk-fixture"])
|
||||
assert args.auth_action == "add" and args.provider == "fake-auth"
|
||||
|
||||
auth_command(args)
|
||||
|
||||
assert _log(tmp_path) == [{
|
||||
"action": "add", "provider": "fake-auth", "label": "work",
|
||||
"target": None, "api_key": "sk-fixture"}]
|
||||
# The built-in path never ran: nothing was written to the credential pool.
|
||||
assert not (tmp_path / "hermes" / "auth.json").exists()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("argv", "action"),
|
||||
[
|
||||
(["status", "fake-auth"], "status"),
|
||||
(["logout", "fake-auth"], "logout"),
|
||||
(["refresh", "fake-auth", "acct-2"], "refresh"),
|
||||
],
|
||||
)
|
||||
def test_status_logout_refresh_dispatch_to_same_handler(tmp_path, install_provider, capsys, argv, action):
|
||||
install_provider()
|
||||
|
||||
from hermes_cli.auth_commands import auth_command
|
||||
|
||||
args = _parse_auth_args(argv)
|
||||
auth_command(args)
|
||||
|
||||
recorded = _log(tmp_path)
|
||||
assert [r["action"] for r in recorded] == [action]
|
||||
# Core prints only when it owns the action; a dispatched action prints nothing here.
|
||||
assert capsys.readouterr().out == ""
|
||||
|
||||
|
||||
def test_declining_handler_falls_back_to_the_builtin_path(tmp_path, install_provider, monkeypatch):
|
||||
"""A handler may decline per action — core then behaves as it always did."""
|
||||
install_provider()
|
||||
monkeypatch.setenv("FAKE_AUTH_DECLINE", "add")
|
||||
|
||||
from hermes_cli.auth_commands import auth_command
|
||||
|
||||
with pytest.raises(SystemExit) as excinfo:
|
||||
auth_command(_parse_auth_args(["add", "fake-auth"]))
|
||||
|
||||
# Offered first, declined by the handler, then the built-in unknown-provider exit...
|
||||
assert [r["action"] for r in _log(tmp_path)] == ["add"]
|
||||
# ...which now names the plugin instead of pretending the provider is unknown.
|
||||
message = str(excinfo.value)
|
||||
assert message.startswith("Unknown provider: fake-auth")
|
||||
assert "does not provide auth handling" in message
|
||||
|
||||
|
||||
def test_builtin_provider_without_handler_is_unchanged(tmp_path, install_provider):
|
||||
"""A provider with no handler keeps the exact built-in credential-pool path."""
|
||||
install_provider()
|
||||
|
||||
from hermes_cli.auth_commands import auth_command
|
||||
|
||||
auth_command(_parse_auth_args(["add", "openrouter", "--api-key", "sk-or-fixture", "--label", "personal"]))
|
||||
|
||||
assert _log(tmp_path) == [] # no handler was ever consulted
|
||||
pool = json.loads((tmp_path / "hermes" / "auth.json").read_text(encoding="utf-8"))["credential_pool"]
|
||||
entry = next(e for e in pool["openrouter"] if e["access_token"] == "sk-or-fixture")
|
||||
assert entry["label"] == "personal"
|
||||
|
||||
|
||||
def test_provider_without_handler_still_reports_unknown_provider(tmp_path, install_provider):
|
||||
install_provider("handlerless", with_handler=False)
|
||||
|
||||
from hermes_cli.auth_commands import auth_command
|
||||
|
||||
with pytest.raises(SystemExit) as excinfo:
|
||||
auth_command(_parse_auth_args(["add", "handlerless"]))
|
||||
|
||||
message = str(excinfo.value)
|
||||
assert message.startswith("Unknown provider: handlerless")
|
||||
assert "does not provide auth handling" in message
|
||||
assert _log(tmp_path) == []
|
||||
|
||||
|
||||
def test_unregistered_provider_lookup_failure_falls_through(tmp_path, install_provider):
|
||||
"""Registry lookup failure (no profile at all) must never raise or dispatch."""
|
||||
install_provider()
|
||||
|
||||
from hermes_cli.auth_commands import _dispatch_provider_auth, _provider_auth_handler, auth_status_command
|
||||
|
||||
assert _provider_auth_handler("not-a-registered-provider") == (None, None)
|
||||
assert _dispatch_provider_auth("add", SimpleNamespace(provider="not-a-registered-provider"),
|
||||
"not-a-registered-provider") is False
|
||||
|
||||
auth_status_command(SimpleNamespace(provider="not-a-registered-provider"))
|
||||
assert _log(tmp_path) == []
|
||||
|
||||
|
||||
def test_duplicate_registration_last_writer_wins(tmp_path, install_provider):
|
||||
"""Two profiles under one name (a user plugin overriding a bundled one) resolve
|
||||
to the newest handler — the documented override semantics of register_provider."""
|
||||
install_provider()
|
||||
from hermes_cli.auth_commands import _provider_auth_handler
|
||||
|
||||
first, _ = _provider_auth_handler("fake-auth")
|
||||
assert callable(first.auth_handler)
|
||||
|
||||
install_provider(reinstall=True) # second registration for the same name
|
||||
|
||||
second, handler = _provider_auth_handler("fake-auth")
|
||||
assert handler is not None and handler is not first.auth_handler
|
||||
assert second is not first
|
||||
|
||||
|
||||
def test_async_handler_is_awaited(tmp_path, install_provider):
|
||||
install_provider(async_handler=True)
|
||||
|
||||
from hermes_cli.auth_commands import auth_command
|
||||
|
||||
auth_command(_parse_auth_args(["add", "fake-auth"]))
|
||||
|
||||
assert [r["action"] for r in _log(tmp_path)] == ["add"]
|
||||
|
||||
|
||||
def test_handler_failure_becomes_a_readable_exit(tmp_path, install_provider):
|
||||
install_provider(raises=True)
|
||||
|
||||
from hermes_cli.auth_commands import auth_command
|
||||
|
||||
with pytest.raises(SystemExit) as excinfo:
|
||||
auth_command(_parse_auth_args(["add", "fake-auth"]))
|
||||
|
||||
message = str(excinfo.value)
|
||||
assert "fake-auth auth handler failed for `add`" in message
|
||||
assert "ValueError: device flow exploded" in message
|
||||
|
||||
|
||||
def test_every_action_reaches_the_seam(tmp_path, install_provider):
|
||||
"""Guard against a future action being added to the core enum without dispatch."""
|
||||
install_provider()
|
||||
|
||||
from hermes_cli import auth_commands
|
||||
|
||||
for action, command in (
|
||||
("add", auth_commands.auth_add_command),
|
||||
("status", auth_commands.auth_status_command),
|
||||
("logout", auth_commands.auth_logout_command),
|
||||
("refresh", auth_commands.auth_refresh_command),
|
||||
):
|
||||
assert auth_commands._dispatch_provider_auth(action, SimpleNamespace(provider="fake-auth"),
|
||||
"fake-auth"), action
|
||||
|
||||
assert sorted(r["action"] for r in _log(tmp_path)) == ["add", "logout", "refresh", "status"]
|
||||
@@ -100,6 +100,7 @@ Full definition in `providers/base.py`. The most useful ones:
|
||||
| `base_url` | str | Default inference endpoint |
|
||||
| `models_url` | str | Explicit catalog URL (falls back to `{base_url}/models`) |
|
||||
| `auth_type` | str | `api_key` \| `oauth_device_code` \| `oauth_external` \| `copilot` \| `aws_sdk` \| `external_process` |
|
||||
| `auth_handler` | `Callable \| None` | Optional provider-owned interactive auth for `hermes auth add/status/logout/refresh <name>` — see [Provider-owned interactive auth](#provider-owned-interactive-auth-auth_handler) |
|
||||
| `fallback_models` | `tuple[str, ...]` | Curated list shown when live catalog fetch fails |
|
||||
| `default_headers` | `dict[str, str]` | Sent on every request (e.g. Copilot's `Editor-Version`) |
|
||||
| `fixed_temperature` | Any | `None` = use caller's value; `OMIT_TEMPERATURE` sentinel = don't send temperature at all (Kimi) |
|
||||
@@ -227,6 +228,53 @@ Set `profile.api_mode` to match the default your provider ships — it acts as a
|
||||
|
||||
`auth_type` gates which codepaths treat your provider as a "simple api-key provider" — if it's not `api_key`, the PluginManager still records the manifest but Hermes' CLI-level automation (doctor checks, `--provider` flag, setup wizard delegation) may skip over it.
|
||||
|
||||
## Provider-owned interactive auth (`auth_handler`)
|
||||
|
||||
`auth_type` describes *what kind* of credential a provider needs; `auth_handler` is how the
|
||||
plugin **acquires** it — its own device-code / OIDC / IdC flow inside the existing `hermes auth`
|
||||
command family, with no second standalone command plugin (model-provider manifests are skipped by
|
||||
the generic command-plugin loader, so `register(ctx)` is not the way to add commands).
|
||||
|
||||
```python
|
||||
from providers import register_provider
|
||||
from providers.base import ProviderProfile
|
||||
|
||||
|
||||
def kiro_auth(action: str, args) -> bool:
|
||||
"""action: "add" | "status" | "logout" | "refresh"; args: parsed CLI namespace."""
|
||||
if action == "add":
|
||||
start_url = input("IdC start URL: ").strip() # provider-specific inputs
|
||||
creds = run_device_code_flow(start_url)
|
||||
save_my_credentials(creds) # the plugin owns its own storage
|
||||
print("Signed in to Kiro.")
|
||||
return True
|
||||
if action == "status":
|
||||
print("kiro: " + ("logged in" if load_my_credentials() else "logged out"))
|
||||
return True
|
||||
return False # decline → this action stays with the built-in credential-pool handling
|
||||
|
||||
|
||||
register_provider(ProviderProfile(
|
||||
name="kiro", auth_type="oauth_device_code", auth_handler=kiro_auth))
|
||||
```
|
||||
|
||||
| Contract | |
|
||||
|---|---|
|
||||
| Signature | `auth_handler(action, args)` — `args` is the parsed `hermes auth` namespace |
|
||||
| Return | truthy = handled (Hermes prints nothing more, exit 0); falsy = fall back to the built-in path **for that action** |
|
||||
| Async | a returned awaitable is awaited, so `async def` handlers work |
|
||||
| Failure | an exception becomes `SystemExit("<provider> auth handler failed for `<action>`: ValueError: …")` |
|
||||
| No handler | `hermes auth <action> <provider>` behaves exactly as it did before the seam |
|
||||
|
||||
`hermes auth add|status|logout|refresh <provider>` consults the handler **first** — before the
|
||||
known-provider gate and before the credential pool — so a provider Hermes does not otherwise
|
||||
recognize (`oauth_device_code`, `oauth_external`) is fully drivable. Registering the same name
|
||||
twice is last-writer-wins, so a user plugin can replace a bundled provider's flow.
|
||||
|
||||
Hermes passes the parsed namespace, not provider-declared flags: ask for provider-specific values
|
||||
interactively (or read your own config/env). Credentials stay provider-owned — Hermes hands your
|
||||
handler no secrets and reads none back.
|
||||
|
||||
## Discovery timing
|
||||
|
||||
Provider discovery is **lazy** — triggered by the first `get_provider_profile()` or `list_providers()` call in the process. In practice this happens early at startup (`auth.py` module load extends `PROVIDER_REGISTRY` eagerly). If you need to verify your plugin loaded, run:
|
||||
|
||||
Reference in New Issue
Block a user