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

refactor(tools): mcp_oauth inline provider-class builder + _fill_result, fold client-change guard; lazy_deps walrus guards, dist-roots/bytecode-warm folds, docstring compaction; gateway timestamp/token folds

Browse Source
This commit is contained in:
Teknium
2026-09-03 00:11:01 -07:00
parent c9f3d37e97
commit bb430e6990
4 changed files with 249 additions and 383 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/lazy_deps.py
+92 -154
View File
@@ -198,19 +198,14 @@ _SAFE_SPEC = re.compile(rf"^{_NAME_RE}(?:\[[A-Za-z0-9_,\-]+\])?(?:[<>=!~]=?[A-Za
class FeatureUnavailable(RuntimeError):
"""A lazily-installable feature is missing and cannot be made available
(lazy installs disabled, or the install attempt failed)."""
"""A lazily-installable feature is missing and cannot be made available (installs disabled or failed)."""
def __init__(self, feature: str, missing: tuple[str, ...], reason: str):
self.feature = feature
self.missing = missing
self.reason = reason
self.feature, self.missing, self.reason = feature, missing, reason
spec_list = " ".join(repr(s) for s in missing)
super().__init__(
f"Feature {feature!r} unavailable: {reason}. "
f"To enable manually: uv pip install {spec_list} "
f"(or: pip install {spec_list})."
)
f"To enable manually: uv pip install {spec_list} (or: pip install {spec_list}).")
@dataclass(frozen=True)
@@ -220,13 +215,11 @@ class _InstallResult:
stderr: str
# ---- Internals ---------------------------------------------------------------
# Internal bridge var (set by the Docker image, not user config) redirecting lazy installs
# from the sealed venv to a writable durable volume.
# Internal bridge var (set by the Docker image, not user config) redirecting lazy installs from the
# sealed venv to a writable durable volume.
_LAZY_TARGET_ENV = "HERMES_LAZY_INSTALL_TARGET"
# Stamp of the Python X.Y + ABI the target was populated for; a mismatch after an image
# rebuild wipes the store so stale .so files are never imported.
# Stamp of the Python X.Y + ABI the target was populated for; a mismatch after an image rebuild
# wipes the store so stale .so files are never imported.
_TARGET_STAMP_NAME = ".python-abi"
_SUBPROCESS_KW = dict(capture_output=True, text=True, encoding="utf-8", errors="replace", stdin=subprocess.DEVNULL)
@@ -244,8 +237,8 @@ def _lazy_install_target() -> Optional[Path]:
def _ensure_target_ready(target: Path) -> Optional[str]:
"""Create the target dir and validate its ABI stamp; a different-ABI stamp wipes the
contents first (stale .so must never import). None on success, else an error string."""
"""Create the target dir and validate its ABI stamp; a different-ABI stamp wipes the contents
first (stale .so must never import). None on success, else an error string."""
want = _python_abi_tag()
stamp = target / _TARGET_STAMP_NAME
try:
@@ -269,9 +262,8 @@ def _ensure_target_ready(target: Path) -> Optional[str]:
def _activate_target_on_syspath(target: Path) -> None:
"""Append the durable target to ``sys.path`` (idempotent). ``site.addsitedir`` honours
``.pth`` files but inserts near the front, so new entries are moved to the END — core venv
site-packages must win collisions."""
"""Append the durable target to ``sys.path`` (idempotent). ``site.addsitedir`` honours ``.pth``
files but inserts near the front, so new entries are moved to the END — core venv site-packages must win collisions."""
target_str = str(target)
before = list(sys.path)
if target_str not in before:
@@ -293,22 +285,20 @@ def _invalidate_import_caches() -> None:
def activate_durable_lazy_target() -> None:
"""Wire the durable target onto ``sys.path`` at startup so packages installed on a previous
run import on this one. No-op when unset or absent. Never raises."""
"""Wire the durable target onto ``sys.path`` at startup so packages installed on a previous run
import on this one. No-op when unset or absent. Never raises."""
target = _lazy_install_target()
if target is None:
return
try:
if target.exists():
if target is not None and target.exists():
_activate_target_on_syspath(target)
except Exception as e: # pragma: no cover - defensive
logger.debug("Failed to activate durable lazy target %s: %s", target, e)
def _allow_lazy_installs() -> bool:
"""Whether lazy installs are permitted: (1) ``security.allow_lazy_installs: false`` blocks
in BOTH modes; (2) the sealed venv (``HERMES_DISABLE_LAZY_INSTALLS=1``) blocks only without a
durable target to redirect into. Unreadable config fails OPEN — blocking is an explicit opt-in."""
"""Whether lazy installs are permitted: (1) ``security.allow_lazy_installs: false`` blocks in BOTH
modes; (2) the sealed venv (``HERMES_DISABLE_LAZY_INSTALLS=1``) blocks only without a durable
target to redirect into. Unreadable config fails OPEN — blocking is an explicit opt-in."""
cfg = None
with contextlib.suppress(Exception):
from hermes_cli.config import load_config
@@ -323,22 +313,17 @@ def _allow_lazy_installs() -> bool:
def _unsupported_feature_reason(feature: str) -> Optional[str]:
"""Platform capability gate (not policy): why a feature cannot work on this host, or None."""
if sys.platform == "win32" and feature == "platform.matrix":
return (
"unsupported on Windows: Matrix E2EE depends on python-olm, "
"which has no Windows wheel and requires make + libolm to build "
"from sdist. Run Hermes under WSL to use Matrix on Windows."
)
return ("unsupported on Windows: Matrix E2EE depends on python-olm, which has no Windows wheel and "
"requires make + libolm to build from sdist. Run Hermes under WSL to use Matrix on Windows.")
return None
def _spec_is_safe(spec: str) -> bool:
"""Reject pip specs that contain URLs, paths, or shell metacharacters."""
return bool(
spec and len(spec) <= 200
and not any(ch in spec for ch in (";", "|", "&", "`", "$", "\n", "\r", "\t", "\\"))
and not spec.startswith(("-", "/", ".")) and "://" not in spec and "@" not in spec
and _SAFE_SPEC.match(spec)
)
return bool(spec and len(spec) <= 200
and not any(ch in spec for ch in (";", "|", "&", "`", "$", "\n", "\r", "\t", "\\"))
and not spec.startswith(("-", "/", ".")) and "://" not in spec and "@" not in spec
and _SAFE_SPEC.match(spec))
def _pkg_name_from_spec(spec: str) -> str:
@@ -364,14 +349,13 @@ def _installed_version(spec: str) -> Optional[str]:
def _is_satisfied(spec: str) -> bool:
"""Present AND inside the spec's version range, so ``hermes update`` propagates pin bumps
to installed backends. Unparseable specs/versions or a missing ``packaging`` count as
satisfied — err toward "don't churn"."""
"""Present AND inside the spec's version range, so ``hermes update`` propagates pin bumps to
installed backends. Unparseable specs/versions or a missing ``packaging`` count as satisfied — err
toward "don't churn"."""
installed = _installed_version(spec)
if installed is None:
return False
spec_tail = _specifier_from_spec(spec)
if not spec_tail:
if not (spec_tail := _specifier_from_spec(spec)):
return True
try:
from packaging.specifiers import SpecifierSet
@@ -389,9 +373,9 @@ def _is_present(spec: str) -> bool:
def _core_constraints_file() -> Optional[Path]:
"""Temp ``--constraint`` file pinning every core-venv package to its installed version for
durable-target installs: shared deps resolve as satisfied (store stays minimal) and a
conflicting backend fails loudly instead of installing a shadowed copy that can never win
on sys.path. None if enumeration failed (install unconstrained)."""
durable-target installs: shared deps resolve as satisfied (store stays minimal) and a conflicting
backend fails loudly instead of installing a shadowed copy that can never win on sys.path. None if
enumeration failed (install unconstrained)."""
try:
import tempfile
from importlib.metadata import distributions
@@ -413,25 +397,20 @@ def _core_constraints_file() -> Optional[Path]:
def _installed_dist_roots(spec: str, target: Optional[Path]) -> set[Path]:
"""Package dirs a freshly installed *spec* owns, from the dist's file list
(``python-telegram-bot`` ships ``telegram``; some ship several)."""
"""Package dirs a freshly installed *spec* owns, from the dist's file list (``python-telegram-bot``
ships ``telegram``; some ship several)."""
name = _pkg_name_from_spec(spec)
roots: set[Path] = set()
try:
import importlib.metadata as _md
if target is not None:
dist = next(iter(_md.distributions(name=name, path=[str(target)])), None)
else:
dist = _md.distribution(name)
if dist is None:
return roots
for entry in dist.files or ():
parts = entry.parts
dist = next(iter(_md.distributions(name=name, path=[str(target)])), None) if target is not None else _md.distribution(name)
for entry in dist.files or () if dist is not None else ():
top = entry.parts[0] if entry.parts else ""
# Skip hidden entries, __pycache__ and metadata dirs (no importable code).
if not parts or parts[0].startswith(".") or parts[0] == "__pycache__" or parts[0].endswith((".dist-info", ".egg-info")):
if not top or top.startswith(".") or top == "__pycache__" or top.endswith((".dist-info", ".egg-info")):
continue
root = Path(dist.locate_file(parts[0]))
root = Path(dist.locate_file(top))
if root.is_dir():
roots.add(root)
except Exception:
@@ -440,9 +419,9 @@ def _installed_dist_roots(spec: str, target: Optional[Path]) -> set[Path]:
def _warm_installed_bytecode(specs: tuple[str, ...], target: Optional[Path]) -> None:
"""Byte-compile what was just installed: a fresh install writes no ``__pycache__``, so the
next import (often a user request, ~2-10s for a big SDK, reading as a hang) would pay the
compile. Pay it here while the caller already waits. Best-effort; never fails the install."""
"""Byte-compile what was just installed: a fresh install writes no ``__pycache__``, so the next
import (often a user request, ~2-10s for a big SDK, reading as a hang) would pay the compile. Pay
it here while the caller already waits. Best-effort; never fails the install."""
if sys.dont_write_bytecode:
return
try:
@@ -451,25 +430,22 @@ def _warm_installed_bytecode(specs: tuple[str, ...], target: Optional[Path]) ->
return
for spec in specs:
try:
roots = _installed_dist_roots(spec, target)
for root in _installed_dist_roots(spec, target):
try:
compileall.compile_dir(str(root), quiet=2, force=False, workers=1)
except Exception as exc:
logger.debug("Bytecode warm skipped for %s: %s", root, exc)
except Exception as exc:
logger.debug("Bytecode warm skipped for %s: %s", spec, exc)
continue
for root in roots:
try:
compileall.compile_dir(str(root), quiet=2, force=False, workers=1)
except Exception as exc:
logger.debug("Bytecode warm skipped for %s: %s", root, exc)
def _run_installer(cmd: list[str], **kw) -> subprocess.CompletedProcess:
# _SUBPROCESS_KW carries stdin=DEVNULL
return subprocess.run(cmd, **_SUBPROCESS_KW, creationflags=windows_hide_flags(), **kw)
def _uv_binary() -> Optional[str]:
"""Managed uv first ($HERMES_HOME/bin is never on PATH), then PATH. A lookup, not
ensure_uv(): downloading uv mid-turn is more than the caller asked for; pip covers no-uv."""
"""Managed uv first ($HERMES_HOME/bin is never on PATH), then PATH. A lookup, not ensure_uv():
downloading uv mid-turn is more than the caller asked for; pip covers no-uv."""
try:
from hermes_cli.managed_uv import resolve_uv
@@ -480,8 +456,8 @@ def _uv_binary() -> Optional[str]:
def _venv_pip_install(specs: tuple[str, ...], *, timeout: int = 300) -> _InstallResult:
"""Install ``specs`` via the uv -> pip -> ensurepip ladder, venv-scoped or into the durable
``--target`` (constrained to core versions) when :data:`_LAZY_TARGET_ENV` is set.
Independent of ``hermes_cli.tools_config._pip_install`` (no CLI dependency)."""
``--target`` (constrained to core versions) when :data:`_LAZY_TARGET_ENV` is set. Independent of
``hermes_cli.tools_config._pip_install`` (no CLI dependency)."""
if not specs:
return _InstallResult(True, "", "")
target = _lazy_install_target()
@@ -506,27 +482,22 @@ def _venv_pip_install(specs: tuple[str, ...], *, timeout: int = 300) -> _Install
from tools.environments.local import hermes_subprocess_env
uv_env = hermes_subprocess_env(inherit_credentials=False)
uv_env["VIRTUAL_ENV"] = str(Path(sys.executable).parent.parent)
# Tier 1: uv.
uv_bin = _uv_binary()
if uv_bin:
# Tier 1: uv. --compile-bytecode because uv writes no __pycache__ by default, so the first
# import would recompile the backend AND its transitives (_warm_installed_bytecode is the
# belt-and-braces pass for the spec's own roots on any tier).
if uv_bin := _uv_binary():
try:
# --compile-bytecode: uv writes no __pycache__ by default, so the first import
# would recompile the backend AND its transitives; _warm_installed_bytecode is
# the belt-and-braces pass for the spec's own roots on any tier.
r = _run_installer([uv_bin, "pip", "install", "--compile-bytecode", *extra_args, *specs], timeout=timeout, env=uv_env)
if r.returncode != 0:
logger.debug("uv pip install failed: %s", r.stderr)
# A uv resolver failure is authoritative: falling through to pip would discard
# uv policy (exclude-newer) and could install a quarantined release.
# A uv resolver failure is authoritative: falling through to pip would discard uv
# policy (exclude-newer) and could install a quarantined release.
return _finish(r)
except subprocess.TimeoutExpired as e:
logger.debug("uv invocation failed: %s", e)
return _InstallResult(False, "", f"uv pip install timed out: {e}")
except FileNotFoundError as e:
# uv vanished between lookup and spawn; it never evaluated the requirements.
except FileNotFoundError as e: # uv vanished between lookup and spawn; it never evaluated the requirements
logger.debug("uv invocation failed: %s", e)
# Tier 2: python -m pip (ensurepip bootstrap if needed)
pip_cmd = [sys.executable, "-m", "pip"]
try:
@@ -537,7 +508,6 @@ def _venv_pip_install(specs: tuple[str, ...], *, timeout: int = 300) -> _Install
_run_installer([sys.executable, "-m", "ensurepip", "--upgrade", "--default-pip"], timeout=120, check=True)
except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e:
return _InstallResult(False, "", f"pip not available and ensurepip failed: {e}")
try:
return _finish(_run_installer(pip_cmd + ["install", *extra_args, *specs], timeout=timeout))
except subprocess.TimeoutExpired as e:
@@ -550,9 +520,6 @@ def _venv_pip_install(specs: tuple[str, ...], *, timeout: int = 300) -> _Install
constraints.unlink()
# ---- Public API ---------------------------------------------------------------
def feature_missing(feature: str) -> tuple[str, ...]:
"""Return the subset of specs for ``feature`` not currently installed."""
if feature not in LAZY_DEPS:
@@ -561,36 +528,31 @@ def feature_missing(feature: str) -> tuple[str, ...]:
def _prompt_toolkit_active() -> bool:
"""A bare input() deadlocks while a prompt_toolkit app owns the terminal, so ensure() skips
the confirmation under the TUI — reaching it is already gated by security.allow_lazy_installs."""
"""A bare input() deadlocks while a prompt_toolkit app owns the terminal, so ensure() skips the
confirmation under the TUI — reaching it is already gated by security.allow_lazy_installs."""
if "prompt_toolkit.application.current" not in sys.modules:
return False
try:
from prompt_toolkit.application.current import get_app_or_none
app = get_app_or_none()
return app is not None and bool(getattr(app, "is_running", False))
return bool(getattr(get_app_or_none(), "is_running", False))
except Exception:
return False
def ensure(feature: str, *, prompt: bool = True) -> None:
"""Make every package for ``feature`` importable, installing if needed; raises
:class:`FeatureUnavailable` when installs are disabled or fail. ``prompt``: confirm on a
TTY first (non-interactive callers pass False and rely on the config gate)."""
:class:`FeatureUnavailable` when installs are disabled or fail. ``prompt``: confirm on a TTY first
(non-interactive callers pass False and rely on the config gate)."""
if feature not in LAZY_DEPS:
raise FeatureUnavailable(feature, (), f"feature {feature!r} not in LAZY_DEPS allowlist")
missing = feature_missing(feature)
if not missing:
return
unsupported = _unsupported_feature_reason(feature)
if unsupported:
if unsupported := _unsupported_feature_reason(feature):
raise FeatureUnavailable(feature, missing, unsupported)
# Package-manager installs (NixOS etc.) have read-only site-packages: fail fast instead of
# burning ~15s on ensurepip — unless a durable target is configured. The reason MUST start
# with "unsupported ": _refresh_features classifies skips by that prefix.
# Package-manager installs (NixOS etc.) have read-only site-packages: fail fast instead of burning
# ~15s on ensurepip — unless a durable target is configured. The reason MUST start with
# "unsupported ": _refresh_features classifies skips by that prefix.
if _lazy_install_target() is None:
managed_by = "" # config unreadable — proceed with the install
with contextlib.suppress(Exception):
@@ -599,43 +561,29 @@ def ensure(feature: str, *, prompt: bool = True) -> None:
if managed_by:
raise FeatureUnavailable(
feature, missing,
f"unsupported on {managed_by}-managed installs: this build's "
f"packages come from {managed_by}, so Hermes cannot install "
f"them at runtime. Add the dependencies for {feature!r} via "
f"{managed_by} (or run a pip/uv install of Hermes instead)."
)
f"unsupported on {managed_by}-managed installs: this build's packages come from {managed_by}, "
f"so Hermes cannot install them at runtime. Add the dependencies for {feature!r} via "
f"{managed_by} (or run a pip/uv install of Hermes instead).")
for spec in missing: # belt and braces on top of the allowlist
if not _spec_is_safe(spec):
raise FeatureUnavailable(feature, missing, f"refusing to install unsafe spec {spec!r}")
if not _allow_lazy_installs():
raise FeatureUnavailable(feature, missing, "lazy installs disabled (security.allow_lazy_installs=false)")
if prompt and not _prompt_toolkit_active() and sys.stdin.isatty() and sys.stdout.isatty():
spec_list = ", ".join(missing)
try:
answer = input(f"\nFeature {feature!r} requires: {spec_list}\nInstall into the active venv now? [Y/n] ").strip().lower()
answer = input(f"\nFeature {feature!r} requires: {', '.join(missing)}\nInstall into the active venv now? [Y/n] ").strip().lower()
except (EOFError, KeyboardInterrupt):
answer = "n"
if answer and answer not in {"y", "yes"}:
raise FeatureUnavailable(feature, missing, "user declined install at prompt")
logger.info("Lazy-installing %s for feature %r", " ".join(missing), feature)
result = _venv_pip_install(missing)
if not result.success:
# Surface pip's own error (quarantine 404, network), tail-clipped.
if not result.success: # surface pip's own error (quarantine 404, network), tail-clipped
snippet = (result.stderr or result.stdout or "").strip()[-2000:]
raise FeatureUnavailable(feature, missing, f"pip install failed: {snippet or 'no error output'}")
_invalidate_import_caches()
still_missing = feature_missing(feature)
if still_missing:
raise FeatureUnavailable(
feature, still_missing,
"install reported success but packages still not importable (may require Python restart)"
)
if still_missing := feature_missing(feature):
raise FeatureUnavailable(feature, still_missing, "install reported success but packages still not importable (may require Python restart)")
logger.info("Lazy install complete for feature %r", feature)
@@ -645,8 +593,8 @@ def is_available(feature: str) -> bool:
def feature_install_command(feature: str, *, venv_pip: bool = False) -> Optional[str]:
"""Manual install command for a feature, or None. ``venv_pip=True`` uses
``{sys.executable} -m pip`` — immune to PEP 668 failures a bare ``pip install`` invites."""
"""Manual install command for a feature, or None. ``venv_pip=True`` uses ``{sys.executable} -m pip``
— immune to PEP 668 failures a bare ``pip install`` invites."""
if feature not in LAZY_DEPS:
return None
joined = " ".join(repr(s) for s in LAZY_DEPS[feature])
@@ -655,10 +603,9 @@ def feature_install_command(feature: str, *, venv_pip: bool = False) -> Optional
@dataclass
class InstallSpecsResult:
"""Outcome of :func:`install_specs` for one batch of pip specs. ``blocked`` means installs
are gated off (config kill switch, sealed venv without a durable target) or a spec failed
validation — nothing was executed, ``reason`` says why. ``command`` is the human-readable
description of what ran (for UIs/logs)."""
"""Outcome of :func:`install_specs` for one batch of pip specs. ``blocked`` means installs are gated
off (config kill switch, sealed venv without a durable target) or a spec failed validation — nothing
was executed, ``reason`` says why. ``command`` is the human-readable description of what ran."""
ok: bool
blocked: bool = False
reason: str = ""
@@ -668,28 +615,23 @@ class InstallSpecsResult:
def install_specs(specs: list[str] | tuple[str, ...], *, timeout: int = 300) -> InstallSpecsResult:
"""Install data-driven pip specs (plugin manifest ``pip_dependencies``) with the same
routing and gating as :func:`ensure`, but unknown packages are allowed — the caller owns
manifest trust, this owns spec hygiene. Never raises; inspect the :class:`InstallSpecsResult`."""
"""Install data-driven pip specs (plugin manifest ``pip_dependencies``) with the same routing and
gating as :func:`ensure`, but unknown packages are allowed — the caller owns manifest trust, this
owns spec hygiene. Never raises; inspect the :class:`InstallSpecsResult`."""
cleaned = tuple(str(s).strip() for s in specs if str(s).strip())
if not cleaned:
return InstallSpecsResult(ok=True, command="")
for spec in cleaned:
if not _spec_is_safe(spec):
return InstallSpecsResult(ok=False, blocked=True, reason=f"refusing to install unsafe spec {spec!r}")
target = _lazy_install_target()
if not _allow_lazy_installs():
sealed = os.environ.get("HERMES_DISABLE_LAZY_INSTALLS") == "1" and target is None
reason = (
"runtime installs are disabled on this deployment: the agent environment is immutable "
"and no writable install target is configured (HERMES_LAZY_INSTALL_TARGET)"
) if sealed else "runtime installs disabled (security.allow_lazy_installs=false)"
reason = ("runtime installs are disabled on this deployment: the agent environment is immutable "
"and no writable install target is configured (HERMES_LAZY_INSTALL_TARGET)"
) if sealed else "runtime installs disabled (security.allow_lazy_installs=false)"
return InstallSpecsResult(ok=False, blocked=True, reason=reason)
display = "uv pip install " + (f"--target {target} " if target is not None else "") + " ".join(cleaned)
logger.info("Installing pip specs %s (target=%s)", " ".join(cleaned), target or "venv")
try:
result = _venv_pip_install(cleaned, timeout=timeout)
@@ -702,8 +644,8 @@ def install_specs(specs: list[str] | tuple[str, ...], *, timeout: int = 300) ->
def active_features() -> list[str]:
"""Features whose ANCHOR package (first spec) is present at any version — shared helpers
like asyncpg are deliberately not proof a backend was enabled. Drives ``hermes update``."""
"""Features whose ANCHOR package (first spec) is present at any version — shared helpers like
asyncpg are deliberately not proof a backend was enabled. Drives ``hermes update``."""
return [f for f, specs in LAZY_DEPS.items() if specs and _is_present(specs[0])]
@@ -727,28 +669,24 @@ def _refresh_features(features: list[str], *, prompt: bool, restoring: bool) ->
if not feature_missing(feature):
results[feature] = "current"
continue
unsupported = _unsupported_feature_reason(feature)
if unsupported:
if unsupported := _unsupported_feature_reason(feature):
results[feature] = f"skipped: {unsupported}"
continue
try:
ensure(feature, prompt=False if restoring else prompt)
results[feature] = "restored" if restoring else "refreshed"
except FeatureUnavailable as e:
# Opt-outs and platform-incompatible features are skips, not failures.
if "lazy installs disabled" in str(e) or "declined" in str(e) or e.reason.startswith("unsupported "):
results[feature] = f"skipped: {e.reason}"
else:
results[feature] = f"failed: {e.reason}"
except FeatureUnavailable as e: # opt-outs and platform-incompatible features are skips, not failures
skip = "lazy installs disabled" in str(e) or "declined" in str(e) or e.reason.startswith("unsupported ")
results[feature] = f"skipped: {e.reason}" if skip else f"failed: {e.reason}"
except Exception as e:
results[feature] = f"failed: {e}"
return results
def ensure_and_bind(feature: str, importer: Callable[[], dict[str, Any]], target_globals: dict, *, prompt: bool = False) -> bool:
""":func:`ensure` the feature, then ``target_globals.update(importer())`` so module-level
names are rebound after a lazy install (``importer`` returns ``{name: obj}`` and runs only
after ensure succeeds). Returns False (and logs) if deps could not be installed or imported."""
""":func:`ensure` the feature, then ``target_globals.update(importer())`` so module-level names are
rebound after a lazy install (``importer`` returns ``{name: obj}`` and runs only after ensure
succeeds). Returns False (and logs) if deps could not be installed or imported."""
try:
ensure(feature, prompt=prompt)
except FeatureUnavailable as exc:
tools/managed_tool_gateway.py
+16 -31
View File
@@ -44,39 +44,30 @@ def _read_nous_provider_state() -> Optional[dict]:
return None
providers = json.loads(path.read_text(encoding="utf-8-sig")).get("providers", {})
nous_provider = providers.get("nous", {}) if isinstance(providers, dict) else None
if isinstance(nous_provider, dict):
return nous_provider
return nous_provider if isinstance(nous_provider, dict) else None
except Exception:
pass
return None
return None
def _parse_timestamp(value: object) -> Optional[datetime]:
normalized = _clean(value)
if normalized is None:
return None
if normalized.endswith("Z"):
normalized = normalized[:-1] + "+00:00"
try:
parsed = datetime.fromisoformat(normalized)
parsed = datetime.fromisoformat(normalized[:-1] + "+00:00" if normalized.endswith("Z") else normalized)
except ValueError:
return None
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
return parsed.astimezone(timezone.utc)
return (parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=timezone.utc)).astimezone(timezone.utc)
def _access_token_is_expiring(expires_at: object, skew_seconds: int) -> bool:
expires = _parse_timestamp(expires_at)
if expires is None:
return True
return (expires - datetime.now(timezone.utc)).total_seconds() <= max(0, int(skew_seconds))
return expires is None or (expires - datetime.now(timezone.utc)).total_seconds() <= max(0, int(skew_seconds))
def _read_user_token_override() -> Optional[str]:
"""Read the TOOL_GATEWAY_USER_TOKEN override through the secret scope. Scope verdict is
authoritative when installed (a scoped miss must NOT borrow the process env under
multiplex); ``os.environ`` only when unscoped."""
"""Read the TOOL_GATEWAY_USER_TOKEN override through the secret scope. Scope verdict is authoritative
when installed (a scoped miss must NOT borrow the process env under multiplex); ``os.environ`` only when unscoped."""
try:
from agent.secret_scope import UnscopedSecretError, get_secret
@@ -90,16 +81,14 @@ def _read_user_token_override() -> Optional[str]:
def peek_nous_access_token() -> Optional[str]:
"""Cheap token probe: env override or cached auth-store token, no expiry check and no
network — availability scans must stay off the synchronous OAuth refresh path (that lives
in :func:`read_nous_access_token`)."""
"""Cheap token probe: env override or cached auth-store token, no expiry check and no network —
availability scans must stay off the synchronous OAuth refresh path (:func:`read_nous_access_token`)."""
return _read_user_token_override() or _clean((_read_nous_provider_state() or {}).get("access_token"))
def read_nous_access_token() -> Optional[str]:
"""Read a Nous Subscriber OAuth access token from auth store or env override."""
explicit = _read_user_token_override()
if explicit:
if explicit := _read_user_token_override():
return explicit
nous_provider = _read_nous_provider_state() or {}
cached_token = peek_nous_access_token()
@@ -108,8 +97,7 @@ def read_nous_access_token() -> Optional[str]:
try:
from hermes_cli.auth import resolve_nous_access_token
refreshed_token = _clean(resolve_nous_access_token(refresh_skew_seconds=_NOUS_ACCESS_TOKEN_REFRESH_SKEW_SECONDS))
if refreshed_token:
if refreshed_token := _clean(resolve_nous_access_token(refresh_skew_seconds=_NOUS_ACCESS_TOKEN_REFRESH_SKEW_SECONDS)):
return refreshed_token
except Exception as exc:
logger.debug("Nous access token refresh failed: %s", exc)
@@ -126,8 +114,7 @@ def get_tool_gateway_scheme() -> str:
def build_vendor_gateway_url(vendor: str) -> str:
"""Return the gateway origin for a specific vendor."""
explicit_vendor_url = os.getenv(f"{vendor.upper().replace('-', '_')}_GATEWAY_URL", "").strip().rstrip("/")
if explicit_vendor_url:
if explicit_vendor_url := os.getenv(f"{vendor.upper().replace('-', '_')}_GATEWAY_URL", "").strip().rstrip("/"):
return explicit_vendor_url
shared_domain = os.getenv("TOOL_GATEWAY_DOMAIN", "").strip().strip("/") or _DEFAULT_TOOL_GATEWAY_DOMAIN
return f"{get_tool_gateway_scheme()}://{vendor}-gateway.{shared_domain}"
@@ -135,8 +122,7 @@ def build_vendor_gateway_url(vendor: str) -> str:
def resolve_managed_tool_gateway(
vendor: str, gateway_builder: Optional[Callable[[str], str]] = None,
token_reader: Optional[Callable[[], Optional[str]]] = None,
) -> Optional[ManagedToolGatewayConfig]:
token_reader: Optional[Callable[[], Optional[str]]] = None) -> Optional[ManagedToolGatewayConfig]:
"""Resolve shared managed-tool gateway config for a vendor."""
if not managed_nous_tools_enabled():
return None
@@ -148,10 +134,9 @@ def resolve_managed_tool_gateway(
def is_managed_tool_gateway_ready(
vendor: str,
gateway_builder: Optional[Callable[[str], str]] = None,
vendor: str, gateway_builder: Optional[Callable[[str], str]] = None,
token_reader: Optional[Callable[[], Optional[str]]] = None) -> bool:
"""True when a gateway URL and a likely-usable Nous token are present. Defaults to
:func:`peek_nous_access_token` (no OAuth refresh); callers about to make a real request
use :func:`resolve_managed_tool_gateway` instead."""
:func:`peek_nous_access_token` (no OAuth refresh); callers about to make a real request use
:func:`resolve_managed_tool_gateway` instead."""
return resolve_managed_tool_gateway(vendor, gateway_builder=gateway_builder, token_reader=token_reader or peek_nous_access_token) is not None
tools/mcp_dashboard_oauth.py
+5 -12
View File
@@ -1,9 +1,6 @@
"""Dashboard-mediated callback bridge for MCP OAuth.
The MCP SDK remains responsible for discovery, DCR, PKCE, state validation and
token exchange. This module only moves the two human/browser callbacks from a
loopback listener into the already-authenticated dashboard session.
"""
"""Dashboard-mediated callback bridge for MCP OAuth: the SDK still does discovery, DCR, PKCE, state
validation and token exchange; this only moves the two human/browser callbacks from a loopback
listener into the already-authenticated dashboard session."""
from __future__ import annotations
@@ -115,12 +112,8 @@ class DashboardOAuthFlow:
def snapshot(self) -> dict:
with self._lock:
return {
"flow_id": self.flow_id,
"server_name": self.server_name,
"status": self.status,
"authorization_url": self.authorization_url,
"error": self.error}
return {"flow_id": self.flow_id, "server_name": self.server_name, "status": self.status,
"authorization_url": self.authorization_url, "error": self.error}
def mark_worker_done(self) -> None:
self._worker_done.set()
tools/mcp_oauth.py
+136 -186
View File
@@ -2,12 +2,11 @@
"""MCP OAuth 2.1 client support: browser authorization-code flow with PKCE.
The SDK's ``OAuthClientProvider`` does discovery, client identification, PKCE, exchange and
refresh; this module supplies ``HermesTokenStorage`` (on-disk persistence), the localhost
callback listener and ``build_oauth_auth()`` (legacy entry point). client_id is Hermes' Client
ID Metadata Document URL (CIMD) when the server supports it, else RFC 7591 DCR.
``mcp_servers.<name>.oauth`` keys (all optional): client_id, client_secret, scope, redirect_port,
redirect_uri (proxy callback), redirect_host, client_name, client_metadata_url, cimd, user_agent, timeout.
"""
refresh; this module supplies ``HermesTokenStorage`` (on-disk persistence), the localhost callback
listener and ``build_oauth_auth()`` (legacy entry point). client_id is Hermes' Client ID Metadata
Document URL (CIMD) when the server supports it, else RFC 7591 DCR. ``mcp_servers.<name>.oauth`` keys
(all optional): client_id, client_secret, scope, redirect_port, redirect_uri (proxy callback),
redirect_host, client_name, client_metadata_url, cimd, user_agent, timeout."""
import asyncio
import contextlib
@@ -49,8 +48,7 @@ _SDK_CLASSES: dict[str, Any] = {}
def _sdk_class(name: str) -> Any:
"""Return the SDK OAuth class *name*, importing the SDK on first call; None when unavailable
(the failure is remembered so a broken SDK is probed once)."""
"""SDK OAuth class *name*, importing the SDK on first call; None when unavailable (a broken SDK is probed once)."""
global _OAUTH_AVAILABLE
if not _SDK_CLASSES:
try:
@@ -77,17 +75,17 @@ class OAuthNonInteractiveError(RuntimeError):
"""Raised when OAuth requires browser interaction in a non-interactive env."""
# Port used by the most recent callback-port resolution. Legacy global; the per-flow
# closures are the real mechanism (concurrent flows must not share it).
# Port of the most recent callback-port resolution. Legacy global; per-flow closures are the
# real mechanism (concurrent flows must not share it).
_oauth_port: int | None = None
# Interactivity gates for OAuth stdin prompts. ContextVars, NOT threading.local: background
# discovery sets them on its own thread while connect+OAuth runs on the `mcp-event-loop`
# thread via run_coroutine_threadsafe, which copies the calling context into the coroutine.
# discovery sets them on its own thread while connect+OAuth runs on the `mcp-event-loop` thread
# via run_coroutine_threadsafe, which copies the calling context into the coroutine. `forced`
# pushes _is_interactive() past the TTY check for GUI-driven flows (dashboard/desktop REST; the
# paste fallback degrades harmlessly to EOF). Suppression wins — background discovery must never
# start a browser flow.
_oauth_interactive_enabled = contextvars.ContextVar("_oauth_interactive_enabled", default=True)
# Forces _is_interactive() past the TTY check for GUI-driven flows (dashboard/desktop REST);
# the paste fallback degrades harmlessly (EOF). Suppression wins — background discovery must
# never start a browser flow.
_oauth_interactive_forced = contextvars.ContextVar("_oauth_interactive_forced", default=False)
# Paste-prompt tokens that exit OAuth without auth; the waiter maps the sentinel to
@@ -108,17 +106,16 @@ def _safe_filename(name: str) -> str:
return re.sub(r"[^\w\-]", "_", name).strip("_")[:128] or "default"
# -- Callback-port reservation: bound-but-not-listening sockets keyed by port, held from
# selection until the waiter adopts them (closes the select→bind TOCTOU window). Bounded FIFO
# so reconnect loops cannot leak fds.
# Callback-port reservation: bound-but-not-listening sockets keyed by port, held from selection
# until the waiter adopts them (closes the select→bind TOCTOU window). Bounded so reconnect loops cannot leak fds.
_reserved_sockets: "dict[int, socket.socket]" = {}
_MAX_RESERVED_SOCKETS = 8
def _bind_reserved(port: int) -> int | None:
"""Bind ``127.0.0.1:port`` (0 = ephemeral) and park it until the waiter adopts it; None if
taken. The FIFO cap evicts ephemeral parks only: pinned CIMD ports are the only ones the
published document declares, so losing one mid-flow would reopen the race."""
"""Bind ``127.0.0.1:port`` (0 = ephemeral) and park it until the waiter adopts it; None if taken.
The cap evicts ephemeral parks only: losing a pinned CIMD port (the only ones the published
document declares) mid-flow would reopen the race."""
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
sock.bind(("127.0.0.1", port))
@@ -143,7 +140,6 @@ def _reserve_callback_port() -> int:
return _bind_reserved(0) # type: ignore[return-value] # port 0 never returns None
# -- Cached registration lookups ---------------------------------------------
def _cached_client_info(storage: "HermesTokenStorage | None") -> dict | None:
"""The on-disk client registration for *storage*, or None."""
try:
@@ -153,9 +149,9 @@ def _cached_client_info(storage: "HermesTokenStorage | None") -> dict | None:
def _cached_redirect(storage: "HermesTokenStorage | None") -> "tuple[str | None, int | None]":
"""``(https proxy URI, loopback callback port)`` from the cached client registration (None
when absent). A DCR ``client_id`` is bound to its registered redirect URI; a new random port
under it gets ``redirect_uri does not match any registered URIs``."""
"""``(https proxy URI, loopback callback port)`` from the cached client registration (None when
absent): a DCR ``client_id`` is bound to its registered redirect URI, so a new random port under
it gets ``redirect_uri does not match any registered URIs``."""
uri = port = None
for raw in (_cached_client_info(storage) or {}).get("redirect_uris") or []:
with contextlib.suppress(TypeError, ValueError):
@@ -168,7 +164,6 @@ def _cached_redirect(storage: "HermesTokenStorage | None") -> "tuple[str | None,
return uri, port
# -- Interactivity -----------------------------------------------------------
def _is_interactive() -> bool:
"""True if we can reasonably expect to interact with a user."""
if not _oauth_interactive_enabled.get():
@@ -182,8 +177,7 @@ def _is_interactive() -> bool:
def _raise_if_non_interactive(lead: str) -> None:
"""Raise ``OAuthNonInteractiveError`` unless interactive; *lead* is the boundary-specific
first sentence, the ``hermes mcp login`` next step is shared."""
"""Raise ``OAuthNonInteractiveError`` unless interactive; *lead* is the boundary-specific first sentence."""
if not _is_interactive():
raise OAuthNonInteractiveError(
f"{lead} Run `hermes mcp login <server>` interactively to (re)authorize, then restart or reload the gateway."
@@ -191,16 +185,14 @@ def _raise_if_non_interactive(lead: str) -> None:
def force_interactive_oauth():
"""Treat the current context as interactive despite no TTY (GUI-driven auth): the user IS
present, just not on stdin. Crosses the MCP event-loop thread like
``suppress_interactive_oauth``."""
"""Treat the context as interactive despite no TTY (GUI-driven auth: the user IS present, just not
on stdin). Crosses the MCP event-loop thread like ``suppress_interactive_oauth``."""
return _contextvar_set(_oauth_interactive_forced, True)
def suppress_interactive_oauth():
"""Disable stdin-based OAuth prompts for the current execution context; ContextVar-based
so a background-discovery thread's suppression reaches the coroutine scheduled on the
MCP event-loop thread."""
"""Disable stdin-based OAuth prompts for the current context; ContextVar-based so a
background-discovery thread's suppression reaches the coroutine on the MCP event-loop thread."""
return _contextvar_set(_oauth_interactive_enabled, False)
@@ -213,7 +205,6 @@ def _can_open_browser() -> bool:
return bool(os.environ.get("DISPLAY") or os.environ.get("WAYLAND_DISPLAY"))
# -- JSON file I/O ------------------------------------------------------------
def _read_json(path: Path) -> dict | None:
"""Read a JSON file, returning None if it doesn't exist or is invalid."""
if not path.exists():
@@ -226,9 +217,9 @@ def _read_json(path: Path) -> dict | None:
def _write_json(path: Path, data: dict) -> None:
"""Atomically write *data* as JSON created at 0o600 (``O_EXCL`` + mode avoids the
write-then-chmod window where the file inherits a world-readable umask); parent dir tightened
to 0o700. The random per-process tmp suffix avoids clashes with concurrent writers/crash leftovers."""
"""Atomically write *data* as JSON created at 0o600 (``O_EXCL`` + mode avoids the write-then-chmod
window where the file inherits a world-readable umask); parent dir tightened to 0o700. The random
per-process tmp suffix avoids clashes with concurrent writers/crash leftovers."""
path.parent.mkdir(parents=True, exist_ok=True)
secure_parent_dir(path)
tmp = path.with_suffix(f".tmp.{os.getpid()}.{secrets.token_hex(4)}")
@@ -250,7 +241,6 @@ def _model_json(model: Any) -> dict:
return model.model_dump(mode="json", exclude_none=True)
# -- HermesTokenStorage -- persistent token/client-info on disk --------------
class HermesTokenStorage:
"""Persist OAuth state as ``HERMES_HOME/mcp-tokens/<server_name>`` + ``.json`` (tokens),
``.client.json`` (client info), ``.meta.json`` (server metadata), ``.cimd-off`` (CIMD refused)."""
@@ -286,12 +276,11 @@ class HermesTokenStorage:
logger.warning("Corrupt %s at %s -- ignoring: %s", label, path, exc)
return None
# -- tokens --
def _rebase_expires_in(self, data: dict) -> None:
"""Rewrite ``expires_in`` to seconds remaining from the stored absolute ``expires_at``
(not an SDK field, so stripped): a relative value reloaded after restart would make
``is_token_valid()`` True for tokens that expired while down. Legacy files without it
use the file mtime, clamped to zero (self-heals on the next ``set_tokens``)."""
"""Rewrite ``expires_in`` to seconds remaining from the stored absolute ``expires_at`` (not an
SDK field, so stripped): a relative value reloaded after restart would make ``is_token_valid()``
True for tokens that expired while down. Legacy files without it use the file mtime, clamped
to zero (self-heals on the next ``set_tokens``)."""
absolute_expiry = data.pop("expires_at", None)
if absolute_expiry is not None:
data["expires_in"] = int(max(absolute_expiry - time.time(), 0))
@@ -305,20 +294,18 @@ class HermesTokenStorage:
async def set_tokens(self, tokens: "OAuthToken") -> None:
payload = _model_json(tokens)
# Absolute ``expires_at``: a relative ``expires_in`` reloaded after restart has no
# wall-clock reference and leaves ``is_token_valid()`` falsely True.
# Absolute ``expires_at``: see _rebase_expires_in.
if payload.get("expires_in") is not None:
with contextlib.suppress(TypeError, ValueError): # mock tokens / odd shapes: skip, don't fail persistence
payload["expires_at"] = time.time() + int(payload["expires_in"])
_write_json(self._tokens_path(), payload)
logger.debug("OAuth tokens saved for %s", self._server_name)
# -- client info --
@staticmethod
def _coerce_secret_auth_method(data: dict) -> bool:
"""Set ``client_secret_post`` when a secret is present but no method is: some DCR
providers (Supabase) omit ``token_endpoint_auth_method``, the SDK defaults it to ``none``
and the exchange fails without the secret."""
"""Set ``client_secret_post`` when a secret is present but no method is: some DCR providers
(Supabase) omit ``token_endpoint_auth_method``, the SDK defaults it to ``none`` and the
exchange fails without the secret."""
if data.get("client_secret") and data.get("token_endpoint_auth_method") in (None, "none", ""):
data["token_endpoint_auth_method"] = "client_secret_post"
return True
@@ -339,19 +326,18 @@ class HermesTokenStorage:
_write_json(self._client_info_path(), data)
logger.debug("OAuth client info saved for %s", self._server_name)
# -- oauth server metadata -- persisted so a restarted process can refresh without
# re-discovery; otherwise the SDK guesses ``{server_url}/token`` (404) and forces re-auth.
def save_oauth_metadata(self, metadata: "OAuthMetadata") -> None:
"""Persist server metadata so a restarted process can refresh without re-discovery;
otherwise the SDK guesses ``{server_url}/token`` (404) and forces re-auth."""
_write_json(self._meta_path(), _model_json(metadata))
logger.debug("OAuth metadata saved for %s", self._server_name)
def load_oauth_metadata(self) -> "OAuthMetadata | None":
return self._load_model(self._meta_path(), "OAuthMetadata", "OAuth metadata")
# -- CIMD refusal --
def mark_cimd_rejected(self) -> None:
"""Durably record that this server refused our CIMD document, so a restart does not
re-present the refused client_id. Cleared by ``remove()`` so a fixed document gets a retry."""
"""Durably record that this server refused our CIMD document so a restart does not re-present
the refused client_id. Cleared by ``remove()`` so a fixed document gets a retry."""
path = self._cimd_rejected_path()
try:
path.parent.mkdir(parents=True, exist_ok=True)
@@ -363,15 +349,14 @@ class HermesTokenStorage:
"""True when this server has refused our metadata document before."""
return self._cimd_rejected_path().exists()
# -- cleanup --
def remove(self) -> None:
"""Delete all stored OAuth state for this server."""
for p in (*self._state_paths(), self._cimd_rejected_path()):
p.unlink(missing_ok=True)
def snapshot(self) -> dict[str, bytes]:
"""filename -> bytes for the existing state files; feed to ``restore()`` to undo a
``remove()`` after a failed re-auth so a valid token survives."""
"""filename -> bytes of the existing state files; ``restore()`` it to undo a ``remove()`` after
a failed re-auth so a valid token survives."""
snap: dict[str, bytes] = {}
for p in self._state_paths():
with contextlib.suppress(OSError):
@@ -397,9 +382,9 @@ class HermesTokenStorage:
logger.warning("Failed to restore OAuth state %s: %s", fname, exc)
def poison_client_registration(self) -> bool:
"""Discard a dead DCR client (``invalid_client`` at the token endpoint) plus stale
``meta.json`` so the SDK re-registers next flow; tokens are kept (a valid refresh token
survives if re-registration never completes). Keeps one ``.bak``. True if a client file was removed."""
"""Discard a dead DCR client (``invalid_client`` at the token endpoint) plus stale ``meta.json``
so the SDK re-registers next flow; tokens are kept (a valid refresh token survives if
re-registration never completes). Keeps one ``.bak``. True if a client file was removed."""
client_path = self._client_info_path()
if not client_path.exists():
return False
@@ -421,11 +406,10 @@ class HermesTokenStorage:
return self._tokens_path().exists()
# -- Callback capture -- HTTP listener and stdin paste share one result dict --
# Callback capture: the HTTP listener and the stdin paste reader share one result dict.
def _authorization_code_result(code: str, state: "str | None", iss: "str | None" = None):
"""Package redirect parameters in the shape the installed SDK expects: mcp 2.0's
``callback_handler`` returns an ``AuthorizationCodeResult`` (the SDK reads ``.state`` /
``.iss`` off it); older SDKs take a tuple."""
"""Redirect parameters in the shape the installed SDK expects: mcp 2.0's ``callback_handler``
returns an ``AuthorizationCodeResult`` (the SDK reads ``.state``/``.iss`` off it); older SDKs take a tuple."""
try:
from mcp.shared.auth import AuthorizationCodeResult
except ImportError: # mcp < 2.0
@@ -434,9 +418,8 @@ def _authorization_code_result(code: str, state: "str | None", iss: "str | None"
def _parse_redirect_query(query: str) -> dict[str, Any]:
"""Extract code/state/error/iss from a redirect query string. ``iss`` is the RFC 9207
issuer: mcp 2.0 rejects a response that omits it when the server advertised
``authorization_response_iss_parameter_supported``, so keep it."""
"""code/state/error/iss from a redirect query string. ``iss`` (RFC 9207 issuer) is kept: mcp 2.0
rejects a response omitting it when the server advertised ``authorization_response_iss_parameter_supported``."""
params = parse_qs(query)
return {k: params.get(k, [None])[0] for k in ("code", "state", "error", "iss")}
@@ -445,10 +428,6 @@ def _result_taken(result: dict) -> bool:
return result.get("auth_code") is not None or result.get("error") is not None
def _fill_result(result: dict, parsed: dict[str, Any]) -> None:
result.update(auth_code=parsed["code"], state=parsed["state"], error=parsed["error"], iss=parsed["iss"])
def _make_callback_handler() -> tuple[type, dict]:
"""Fresh ``(HandlerClass, result_dict)`` per flow so concurrent flows don't stomp on each other."""
result: dict[str, Any] = {"auth_code": None, "state": None, "error": None, "iss": None}
@@ -456,7 +435,7 @@ def _make_callback_handler() -> tuple[type, dict]:
class _Handler(BaseHTTPRequestHandler):
def do_GET(self) -> None: # noqa: N802
parsed = _parse_redirect_query(urlparse(self.path).query)
_fill_result(result, parsed)
result.update(auth_code=parsed["code"], state=parsed["state"], error=parsed["error"], iss=parsed["iss"])
body = ("<h2>Authorization Successful</h2><p>You can close this tab and return to Hermes.</p>" if parsed["code"]
else f"<h2>Authorization Failed</h2><p>Error: {parsed['error'] or 'unknown'}</p>")
self.send_response(200)
@@ -471,9 +450,9 @@ def _make_callback_handler() -> tuple[type, dict]:
def _paste_callback_reader(result: dict) -> None:
"""Read one stdin line as an OAuth redirect (full URL, bare query, or a ``_SKIP_TOKENS``
word that exits without auth) into *result*. Parse failures, EOF and interrupts are
swallowed — best-effort fallback racing the HTTP listener, which stays primary."""
"""Read one stdin line as an OAuth redirect (full URL, bare query, or a ``_SKIP_TOKENS`` word that
exits without auth) into *result*. Parse failures, EOF and interrupts are swallowed — best-effort
fallback racing the HTTP listener, which stays primary."""
try:
line = sys.stdin.readline()
except (KeyboardInterrupt, OSError, ValueError):
@@ -500,15 +479,13 @@ def _paste_callback_reader(result: dict) -> None:
return
if _result_taken(result): # one more race-check before writing
return
_fill_result(result, parsed)
result.update(auth_code=parsed["code"], state=parsed["state"], error=parsed["error"], iss=parsed["iss"])
if parsed["code"]:
print(" Got authorization code from paste — completing flow.", file=sys.stderr)
# -- Async redirect + callback handlers for OAuthClientProvider --------------
# Remote-session hints printed under the authorization URL: a proxy callback forwards the
# redirect here (no tunnel needed); on loopback the redirect misses this machine, so the user
# pastes the URL back or SSH-forwards the port.
# Remote-session hints printed under the authorization URL: a proxy callback forwards the redirect
# here (no tunnel needed); on loopback it misses this machine, so the user pastes the URL back or SSH-forwards the port.
_SSH_HINT_PROXY = (
" Remote session detected. After you authorize, the provider redirects to\n"
" {redirect_uri}\n"
@@ -549,17 +526,15 @@ def _announce_authorization_url(authorization_url: str, port: int, redirect_uri:
def _make_redirect_handler(port: int, redirect_uri: str | None = None):
"""Return a redirect handler closing over this flow's port (a closure, not the module-level
``_oauth_port``, keeps concurrent server flows isolated). ``redirect_uri`` is a configured
proxy callback (None for loopback) and only tailors the remote-session hint."""
"""Redirect handler closing over this flow's port (a closure, not ``_oauth_port``, keeps concurrent
flows isolated). ``redirect_uri`` is a configured proxy callback (None for loopback) and only tailors the hint."""
async def _redirect_handler(authorization_url: str) -> None:
dashboard_flow = get_dashboard_oauth_flow()
if dashboard_flow is not None:
await dashboard_flow.publish_authorization_url(authorization_url)
return
# Fail fast in non-interactive contexts: a cached-but-unusable token makes the SDK
# fall through to the authorization-code flow past the token-file guard, and we would
# otherwise block in the waiter for the full timeout.
# Fail fast when non-interactive: a cached-but-unusable token makes the SDK fall through to the
# authorization-code flow past the token-file guard, and the waiter would block for the full timeout.
_raise_if_non_interactive(
"MCP OAuth requires browser authorization but no interactive session is available (non-interactive/background context)."
)
@@ -569,9 +544,9 @@ def _make_redirect_handler(port: int, redirect_uri: str | None = None):
def _start_callback_server(port: int, handler_cls: type) -> HTTPServer:
"""Bind the callback listener on *port*, adopting a parked reserved socket (closes the
select→bind TOCTOU window). ``allow_reuse_address`` is set BEFORE binding (a no-op
afterwards) so a lingering TIME_WAIT socket from a previous flow cannot block the next."""
"""Bind the callback listener on *port*, adopting a parked reserved socket (closes the select→bind
TOCTOU window). ``allow_reuse_address`` is set BEFORE binding (a no-op afterwards) so a lingering
TIME_WAIT socket from a previous flow cannot block the next."""
try:
server = HTTPServer(("127.0.0.1", port), handler_cls, bind_and_activate=False)
reserved = _reserved_sockets.pop(port, None)
@@ -582,9 +557,7 @@ def _start_callback_server(port: int, handler_cls: type) -> HTTPServer:
server.allow_reuse_address = True
server.server_bind()
server.server_activate()
except OSError as exc:
# Genuinely in use: a concurrent flow, leftover listener, or colliding fixed
# `oauth.redirect_port`. Nothing to poll — say so, not "timed out".
except OSError as exc: # genuinely in use (concurrent flow / leftover listener / colliding redirect_port): say so, not "timed out"
raise OAuthNonInteractiveError(
f"OAuth callback port {port} is already in use ({exc}). Close any other in-progress login, "
"or set a free `oauth.redirect_port` in the server config, then retry."
@@ -611,19 +584,17 @@ def _callback_outcome(result: dict, cimd_url: str | None):
def _make_callback_waiter(port: int, cimd_url: str | None = None, timeout: float = 300.0):
"""Return a callback waiter bound to one flow's port. ``timeout`` is where ``oauth.timeout``
applies (mcp 2.0 dropped the provider's own). ``cimd_url`` only tailors the timeout message:
a server refusing the document aborts at the authorization endpoint, so no redirect arrives
and a bare "timed out" would hide the cause. Raises ``OAuthNonInteractiveError`` on timeout
or when non-interactive."""
"""Callback waiter bound to one flow's port. ``timeout`` is where ``oauth.timeout`` applies (mcp 2.0
dropped the provider's own). ``cimd_url`` only tailors the timeout message: a server refusing the
document aborts at the authorization endpoint, so no redirect arrives and a bare "timed out" would
hide the cause. Raises ``OAuthNonInteractiveError`` on timeout or when non-interactive."""
async def _wait():
dashboard_flow = get_dashboard_oauth_flow()
if dashboard_flow is not None:
# Dashboard flow speaks the legacy tuple; normalize to one shape.
return _authorization_code_result(*await dashboard_flow.wait_for_callback())
# The SDK entered the authorization-code flow, so any cached token is unusable. Reject
# BEFORE binding: binding would block for the full timeout and collide with the
# TIME_WAIT port on retry.
# The SDK entered the authorization-code flow, so any cached token is unusable. Reject BEFORE
# binding: binding would block for the full timeout and collide with the TIME_WAIT port on retry.
_raise_if_non_interactive(
"OAuth callback requires an interactive session but none is available (non-interactive/background "
"context); skipping browser authorization without binding a callback listener.")
@@ -649,45 +620,31 @@ def _make_callback_waiter(port: int, cimd_url: str | None = None, timeout: float
return _wait
# -- OAuth provider class (legacy build_oauth_auth path) ---------------------
# Legacy build_oauth_auth provider class, built lazily (SDK) and cached here.
HermesOAuthClientProvider: Any = None
def _get_hermes_oauth_provider_class() -> type | None:
"""Build (once) and cache ``HermesOAuthClientProvider``; None without the SDK."""
global HermesOAuthClientProvider
base = _sdk_class("OAuthClientProvider")
if HermesOAuthClientProvider is None and base is not None:
from tools.mcp_oauth_provider import HermesProviderMixin
HermesOAuthClientProvider = type("HermesOAuthClientProvider", (HermesProviderMixin, base), {
"__doc__": "SDK provider plus Hermes' token-endpoint fixes (see ``HermesProviderMixin``).",
"__module__": __name__, "_hermes_logger": logger})
return HermesOAuthClientProvider
def remove_oauth_tokens(server_name: str, *, hermes_home: str | Path | None = None) -> None:
"""Delete stored OAuth tokens and client info for a server."""
HermesTokenStorage(server_name, hermes_home=hermes_home).remove()
logger.info("OAuth tokens removed for '%s'", server_name)
# -- CIMD -- OAuth Client ID Metadata Documents: the client_id IS an HTTPS URL the server
# fetches for our name/logo/redirect URIs, replacing per-install DCR. The SDK does the protocol;
# Hermes only decides eligibility. Published from ``website/static/oauth/client-metadata.json``;
# the github.io origin is deliberate — servers MUST NOT follow redirects when fetching it, and
# hermes-agent.nousresearch.com/docs/* 301s here.
# CIMD (OAuth Client ID Metadata Documents): the client_id IS an HTTPS URL the server fetches for our
# name/logo/redirect URIs, replacing per-install DCR. The SDK does the protocol; Hermes only decides
# eligibility. Published from ``website/static/oauth/client-metadata.json``; the github.io origin is
# deliberate — servers MUST NOT follow redirects when fetching it, and hermes-agent.nousresearch.com/docs/* 301s here.
_CIMD_CLIENT_METADATA_URL = "https://nousresearch.github.io/hermes-agent/docs/oauth/client-metadata.json"
# Loopback ports/hosts declared in that document (exact match, so no ephemeral port under
# CIMD); below Linux's 32768 ephemeral floor. tests/tools/test_mcp_cimd.py keeps them in sync.
# Loopback ports/hosts declared in that document (exact match, so no ephemeral port under CIMD);
# below Linux's 32768 ephemeral floor. tests/tools/test_mcp_cimd.py keeps them in sync.
_CIMD_PORTS = (27890, 27891, 27892, 27893, 27894)
_CIMD_REDIRECT_HOSTS = frozenset({"127.0.0.1", "localhost"})
def _is_valid_cimd_url(url: str) -> bool:
"""True when *url* is usable as a CIMD client_id on the installed SDK (ImportError = SDK
predates CIMD → DCR only). The SDK checks only https + non-root path; userinfo, fragments
and dot segments are rejected here because they fail mid-browser-flow as an opaque invalid-client page."""
"""True when *url* is usable as a CIMD client_id on the installed SDK (ImportError = SDK predates
CIMD → DCR only). The SDK checks only https + non-root path; userinfo, fragments and dot segments
are rejected here because they fail mid-browser-flow as an opaque invalid-client page."""
try:
from mcp.client.auth.utils import is_valid_client_metadata_url
if not is_valid_client_metadata_url(url):
@@ -699,17 +656,16 @@ def _is_valid_cimd_url(url: str) -> bool:
return not (has_userinfo or parsed.fragment or any(seg in {".", ".."} for seg in parsed.path.split("/")))
# Pinned ports this process has committed to (never released: a provider keeps its port for
# the process lifetime), including ports restored from a cached registration.
# Pinned ports this process committed to (never released: a provider keeps its port for the process
# lifetime), including ports restored from a cached registration.
_assigned_cimd_ports: "list[int]" = []
def _pick_cimd_port() -> int | None:
"""Reserve a pinned CIMD callback port, or None when none is usable. Holding the bound
socket makes contention cooperative: a sibling finds the bind refused and moves down the
range. Once every pinned port belongs to this process the range wraps rather than falling
back to DCR — a reused port only bites if both servers authorize at the same moment
(reported by the waiter), whereas DCR may be unsupported entirely."""
"""Reserve a pinned CIMD callback port, or None when none is usable. Holding the bound socket makes
contention cooperative: a sibling finds the bind refused and moves down the range. Once every pinned
port belongs to this process the range wraps rather than falling back to DCR — a reused port only
bites if both servers authorize at the same moment (reported by the waiter); DCR may be unsupported entirely."""
for port in _CIMD_PORTS:
if port not in _assigned_cimd_ports and _bind_reserved(port) is not None:
_assigned_cimd_ports.append(port)
@@ -718,9 +674,9 @@ def _pick_cimd_port() -> int | None:
def _server_declined_cimd(storage: "HermesTokenStorage | None") -> bool:
"""True when cached metadata shows this server doesn't advertise CIMD. The SDK decides
CIMD vs DCR in its 401 branch — after Hermes must fix the redirect URI — so cached metadata
closes the gap; only a genuinely unknown server pays the optimistic pin."""
"""True when cached metadata shows this server doesn't advertise CIMD. The SDK decides CIMD vs DCR
in its 401 branch — after Hermes must fix the redirect URI — so cached metadata closes the gap;
only a genuinely unknown server pays the optimistic pin."""
try:
metadata = storage.load_oauth_metadata() if storage is not None else None
except (AttributeError, TypeError, ValueError):
@@ -729,9 +685,9 @@ def _server_declined_cimd(storage: "HermesTokenStorage | None") -> bool:
def _maybe_use_cimd(cfg: dict, storage: "HermesTokenStorage | None" = None) -> "tuple[str, int] | None":
"""Return ``(client_id URL, pinned callback port)``, or None to use DCR. Each ineligibility
case means the redirect URI is not one the document declares, the client identity is already
settled, or the server is known not to want a document — a metadata URL would be rejected."""
"""``(client_id URL, pinned callback port)``, or None to use DCR. Each ineligibility case means the
redirect URI is not one the document declares, the client identity is already settled, or the
server is known not to want a document — a metadata URL would be rejected."""
url = cfg.get("client_metadata_url") or _CIMD_CLIENT_METADATA_URL
ineligible = (
cfg.get("cimd") is False
@@ -753,26 +709,25 @@ def _maybe_use_cimd(cfg: dict, storage: "HermesTokenStorage | None" = None) -> "
def cimd_provider_kwargs(cfg: dict) -> dict[str, Any]:
"""``client_metadata_url=`` for ``OAuthClientProvider``, when CIMD applies. Returned as
kwargs so the argument is omitted entirely on a DCR flow: an SDK too old for CIMD rejects
the keyword outright."""
"""``client_metadata_url=`` kwargs for ``OAuthClientProvider`` when CIMD applies; omitted entirely
on a DCR flow because an SDK too old for CIMD rejects the keyword outright."""
url = cfg.get("_cimd_url")
return {"client_metadata_url": url} if url else {}
def token_request_user_agent(cfg: dict) -> str | None:
"""Configured ``oauth.user_agent`` for token-endpoint requests (exchange + refresh only,
never MCP traffic or discovery), or None; a null/empty YAML value never sends a blank
header. No other headers are configurable (secrets would land in config.yaml)."""
"""Configured ``oauth.user_agent`` for token-endpoint requests (exchange + refresh only, never MCP
traffic or discovery), or None; a null/empty YAML value never sends a blank header. No other
headers are configurable (secrets would land in config.yaml)."""
ua = cfg.get("user_agent")
return ua.strip() if isinstance(ua, str) and ua.strip() else None
def _configure_callback_port(cfg: dict, storage: "HermesTokenStorage | None" = None) -> int:
"""Resolve the callback port into ``cfg['_resolved_port']`` (0 = non-loopback URI).
Precedence: dashboard flow / cached https redirect URI → CIMD pinned port (sets
``cfg['_cimd_url']``) → ``oauth.redirect_port`` → cached registration port → fresh
ephemeral port (the only parked one). Also sets the legacy ``_oauth_port``."""
"""Resolve the callback port into ``cfg['_resolved_port']`` (0 = non-loopback URI). Precedence:
dashboard flow / cached https redirect URI → CIMD pinned port (sets ``cfg['_cimd_url']``) →
``oauth.redirect_port`` → cached registration port → fresh ephemeral port (the only parked one).
Also sets the legacy ``_oauth_port``."""
global _oauth_port
dashboard_flow = get_dashboard_oauth_flow()
if dashboard_flow is not None:
@@ -798,34 +753,31 @@ def _configure_callback_port(cfg: dict, storage: "HermesTokenStorage | None" = N
def _resolve_redirect_uri(cfg: dict, port: int) -> str:
"""Configured ``redirect_uri`` (proxy) or ``http://<redirect_host>:<port>/callback``; the
single derivation so client metadata and pre-registered info stay identical.
``redirect_host`` only changes the hostname (some WAFs reject a literal ``127.0.0.1``);
the listener still binds ``127.0.0.1``."""
"""Configured ``redirect_uri`` (proxy) or ``http://<redirect_host>:<port>/callback``; the single
derivation so client metadata and pre-registered info stay identical. ``redirect_host`` only changes
the hostname (some WAFs reject a literal ``127.0.0.1``); the listener still binds ``127.0.0.1``."""
return cfg.get("redirect_uri") or f"http://{cfg.get('redirect_host') or '127.0.0.1'}:{port}/callback"
# Figma's remote MCP allowlists DCR by client_name ("Claude Code"/"Codex" register, others
# 403); register under an allowlisted name so the flow can start. oauth.client_name overrides.
# Figma's remote MCP allowlists DCR by client_name ("Claude Code"/"Codex" register, others 403);
# register under an allowlisted name so the flow can start. oauth.client_name overrides.
_FIGMA_DCR_CLIENT_NAME = "Claude Code"
_FIGMA_DEFAULT_SCOPE = "mcp:connect"
def _is_figma_remote_mcp(server_name: str | None = None, server_url: str | None = None) -> bool:
"""True when this MCP server is Figma's hosted remote endpoint."""
url = (server_url or "").lower()
name = (server_name or "").lower()
from utils import base_url_host_matches, base_url_hostname
url = (server_url or "").lower()
if base_url_host_matches(url, "mcp.figma.com") or (base_url_host_matches(url, "figma.com") and "/mcp" in url):
return True
# Name-only match only when the URL isn't some other host called figma-*.
return "figma" in name and (not url or "figma" in base_url_hostname(url))
return "figma" in (server_name or "").lower() and (not url or "figma" in base_url_hostname(url))
def apply_oauth_provider_defaults(cfg: dict, *, server_name: str = "", server_url: str | None = None) -> dict:
"""Mutate *cfg* with provider-specific OAuth workarounds; returns *cfg*. Call before
building client metadata / pre-registering. Only fills keys the user left unset —
explicit ``oauth.client_name`` / ``oauth.scope`` win."""
"""Mutate *cfg* with provider-specific OAuth workarounds (before building client metadata /
pre-registering); returns *cfg*. Only fills keys the user left unset — explicit values win."""
if _is_figma_remote_mcp(server_name, server_url):
if not cfg.get("client_name"):
cfg["client_name"] = _FIGMA_DCR_CLIENT_NAME
@@ -834,8 +786,8 @@ def apply_oauth_provider_defaults(cfg: dict, *, server_name: str = "", server_ur
server_name or server_url, _FIGMA_DCR_CLIENT_NAME)
if not cfg.get("scope"):
cfg["scope"] = _FIGMA_DEFAULT_SCOPE
# Figma advertises auth_method=none yet demands the returned client_secret at the
# token endpoint; request a confidential registration so the SDK posts it.
# Figma advertises auth_method=none yet demands the returned client_secret at the token
# endpoint; request a confidential registration so the SDK posts it.
cfg["token_endpoint_auth_method"] = cfg.get("token_endpoint_auth_method") or "client_secret_post"
return cfg
@@ -846,8 +798,7 @@ def _build_client_metadata(cfg: dict) -> "OAuthClientMetadata":
if port is None:
raise ValueError("_configure_callback_port() must be called before _build_client_metadata()")
metadata_cls = _sdk_class("OAuthClientMetadata")
# Public client by default; confidential only with a known secret or a provider (e.g.
# Figma) that needs confidential-style token posts.
# Public client by default; confidential only with a known secret or a provider (Figma) needing confidential-style token posts.
auth_method = cfg.get("token_endpoint_auth_method") or ("client_secret_post" if cfg.get("client_secret") else "none")
metadata_kwargs: dict[str, Any] = {
"client_name": cfg.get("client_name", "Hermes Agent"),
@@ -855,30 +806,27 @@ def _build_client_metadata(cfg: dict) -> "OAuthClientMetadata":
"grant_types": ["authorization_code", "refresh_token"],
"response_types": ["code"],
"token_endpoint_auth_method": auth_method,
# SEP-837: OIDC-strict servers need application_type to accept loopback redirects;
# "native" for a CLI/desktop app, overridable for a hosted https dashboard.
# SEP-837: OIDC-strict servers need application_type to accept loopback redirects; "native"
# for a CLI/desktop app, overridable for a hosted https dashboard.
"application_type": cfg.get("application_type", "native")}
if cfg.get("scope"):
metadata_kwargs["scope"] = cfg["scope"]
try:
return metadata_cls.model_validate(metadata_kwargs)
except Exception:
# mcp 1.x metadata models predate SEP-837 and reject the unknown field.
except Exception: # mcp 1.x metadata models predate SEP-837 and reject the unknown field
metadata_kwargs.pop("application_type", None)
return metadata_cls.model_validate(metadata_kwargs)
def _invalidate_tokens_on_client_change(
storage: "HermesTokenStorage", new_client_id: str, new_client_secret: str | None) -> None:
"""Drop cached tokens when the configured client identity changes: tokens minted under the
old ``client_id`` fail refresh with ``invalid_client``, and pre-registered clients are exempt
from auto-poison, so stale tokens would wedge every request until a manual wipe. Compares
on-disk ``client.json`` BEFORE it is overwritten; a matching identity is a no-op."""
"""Drop cached tokens when the configured client identity changes: tokens minted under the old
``client_id`` fail refresh with ``invalid_client``, and pre-registered clients are exempt from
auto-poison, so stale tokens would wedge every request until a manual wipe. Compares on-disk
``client.json`` BEFORE it is overwritten; a matching identity is a no-op."""
existing = _read_json(storage._client_info_path())
old_client_id = existing.get("client_id") if isinstance(existing, dict) else None
if not old_client_id:
return
if old_client_id == new_client_id and (existing.get("client_secret") or None) == (new_client_secret or None):
if not old_client_id or (old_client_id == new_client_id and (existing.get("client_secret") or None) == (new_client_secret or None)):
return
removed = False
for path in (storage._tokens_path(), storage._meta_path()):
@@ -916,9 +864,9 @@ def _maybe_preregister_client(storage: "HermesTokenStorage", cfg: dict, client_m
def humanize_oauth_registration_error(
server_name: str, exc: BaseException | str, *, server_url: str | None = None) -> str | None:
"""Turn a DCR 403/Forbidden into a useful next step; None for anything else so the caller
keeps the original text. Figma gates DCR on exact ``client_name`` (auto-set to ``Claude
Code``), so this fires when the user overrode it or an older Hermes is running."""
"""Turn a DCR 403/Forbidden into a useful next step; None for anything else so the caller keeps the
original text. Figma gates DCR on exact ``client_name`` (auto-set to ``Claude Code``), so this fires
when the user overrode it or an older Hermes is running."""
msg = str(exc)
lowered = msg.lower()
looks_like_registration = ("403" in msg or "forbidden" in lowered) and (
@@ -941,9 +889,9 @@ def humanize_oauth_registration_error(
def build_oauth_auth(server_name: str, server_url: str, oauth_config: dict | None = None) -> "OAuthClientProvider | None":
"""Build an ``httpx.Auth`` OAuth handler for an MCP server; None if the SDK lacks OAuth.
Legacy API — new code uses :func:`tools.mcp_oauth_manager.get_manager` so state is shared
across config-time, runtime and reconnect paths."""
"""``httpx.Auth`` OAuth handler for an MCP server; None if the SDK lacks OAuth. Legacy API — new code
uses :func:`tools.mcp_oauth_manager.get_manager` so state is shared across config-time, runtime and reconnect paths."""
global HermesOAuthClientProvider
if not _OAUTH_AVAILABLE or _sdk_class("OAuthClientProvider") is None:
logger.warning("MCP OAuth requested for '%s' but SDK auth types are not available. Install with: pip install 'mcp>=1.26.0'", server_name)
return None
@@ -956,8 +904,10 @@ def build_oauth_auth(server_name: str, server_url: str, oauth_config: dict | Non
f"requires browser authorization. Run `hermes mcp login {server_name}` interactively first to complete "
"initial authorization, then cached tokens will be reused.")
kwargs = build_provider_kwargs(cfg, storage, ssh_proxy_hint=True)
provider_class = _get_hermes_oauth_provider_class()
if provider_class is None:
logger.warning("MCP OAuth requested for '%s' but the provider class is unavailable", server_name)
return None
return provider_class(server_url=server_url, **kwargs)
if HermesOAuthClientProvider is None:
from tools.mcp_oauth_provider import HermesProviderMixin
HermesOAuthClientProvider = type("HermesOAuthClientProvider", (HermesProviderMixin, _sdk_class("OAuthClientProvider")), {
"__doc__": "SDK provider plus Hermes' token-endpoint fixes (see ``HermesProviderMixin``).",
"__module__": __name__, "_hermes_logger": logger})
return HermesOAuthClientProvider(server_url=server_url, **kwargs)