Coverage for src/lilbee/cli/launchers/hermes.py: 100%

1"""hermes launcher: registers lilbee in the user's real ~/.hermes, then runs hermes.""" 

2 

3from __future__ import annotations 

4 

5import os 

6import shutil 

7from pathlib import Path 

8 

9import typer 

10import yaml 

11 

12from lilbee.app.agent_configs.hermes import hermes_config 

13from lilbee.app.agent_configs.merge import deep_merge, prune_lilbee 

14from lilbee.cli.launchers import config_file 

15from lilbee.cli.launchers.hermes_mcp import ensure_hermes_http_mcp 

16from lilbee.cli.launchers.launcher import LILBEE_TOKEN_ENV_VAR, run_launcher 

17from lilbee.cli.launchers.server import LOOPBACK, client_chat_ctx 

18from lilbee.cli.launchers.skill_install import install_bundled_skill 

19from lilbee.core.config import cfg 

20 

21_HERMES_INSTALL_HINT = ( 

22 "hermes binary not found on PATH. Install it from https://github.com/NousResearch/hermes-agent." 

23) 

24_TOKEN_REF = "${" + LILBEE_TOKEN_ENV_VAR + "}" 

25_MCP_CONTAINER_KEY = "mcp_servers" 

26_CONFIG_LABEL = "hermes config (config.yaml)" 

27# hermes refuses to start against a model whose window is under this, so a smaller 

28# one is worth naming here rather than leaving hermes to fail after the handoff. 

29_HERMES_MIN_CTX = 64_000 

30# hermes config keys gating its own auto-installs (security.allow_lazy_installs). 

31_SECURITY_KEY = "security" 

32_ALLOW_LAZY_INSTALLS_KEY = "allow_lazy_installs" 

33 

34 

35def _hermes_home() -> Path: 

36 """The user's real hermes home; never relocated, so memory and skills are shared.""" 

37 return Path.home() / ".hermes" 

38 

39 

40def _hermes_config_path() -> Path: 

41 return _hermes_home() / "config.yaml" 

42 

43 

44def _hermes_env_path() -> Path: 

45 return _hermes_home() / ".env" 

46 

47 

48def _hermes_skill_dest() -> Path: 

49 return _hermes_home() / "skills" / "lilbee-mcp" 

50 

51 

52def _upsert_env_token(path: Path, token: str) -> None: 

53 """Set ``LILBEE_TOKEN=<token>`` in the hermes ``.env`` (0600), preserving other lines.""" 

54 line = f"{LILBEE_TOKEN_ENV_VAR}={token}" 

55 existing = path.read_text(encoding="utf-8").splitlines() if path.exists() else [] 

56 kept = [ln for ln in existing if not ln.startswith(f"{LILBEE_TOKEN_ENV_VAR}=")] 

57 # atomic_write_text creates the file 0600 and keeps that mode across the 

58 # replace, so the token is never briefly readable and needs no chmod after. 

59 config_file.atomic_write_text(path, "\n".join([*kept, line]) + "\n") 

60 

61 

62def warn_hermes_ungrounded() -> None: 

63 """Say plainly that hermes will run ungrounded when its MCP search did not connect. 

64 

65 Without the ``mcp`` extra hermes never calls ``lilbee_search`` and answers from 

66 its own training, silently at exit 0, so the install hint alone is easy to miss. 

67 """ 

68 typer.secho( 

69 "Warning: hermes could not connect lilbee's search (MCP), so it will run " 

70 "WITHOUT grounding -- it will not call lilbee_search and its answers come " 

71 "from its own training, not your indexed docs. Install hermes's mcp extra " 

72 "(shown above) and relaunch to ground it.", 

73 err=True, 

74 fg=typer.colors.YELLOW, 

75 ) 

76 

77 

78def warn_if_below_hermes_minimum(chat_ctx: int | None) -> None: 

79 """Tell the user up front when hermes will reject the window lilbee serves.""" 

80 if chat_ctx is None or chat_ctx >= _HERMES_MIN_CTX: 

81 return 

82 typer.secho( 

83 f"Warning: hermes requires at least a {_HERMES_MIN_CTX:,}-token context and " 

84 f"lilbee serves {chat_ctx:,}, so hermes will refuse to start. Chat with a " 

85 "longer-context model, a smaller quantization, or a higher gpu_memory_fraction.", 

86 err=True, 

87 fg=typer.colors.YELLOW, 

88 ) 

89 

90 

91class HermesLauncher: 

92 """``Launcher`` implementation for hermes-agent.""" 

93 

94 name = "hermes" 

95 install_hint = _HERMES_INSTALL_HINT 

96 

97 def __init__(self, *, include_mcp: bool = True) -> None: 

98 self._include_mcp = include_mcp 

99 self._binary: str | None = None 

100 

101 def find_binary(self) -> str | None: 

102 self._binary = shutil.which("hermes") 

103 return self._binary 

104 

105 def prepare( 

106 self, *, token: str, port: int, model_refs: list[str] 

107 ) -> tuple[list[str], dict[str, str]]: 

108 config = config_file.load_config_dict( 

109 _hermes_config_path(), 

110 parse=yaml.safe_load, 

111 parse_error=yaml.YAMLError, 

112 label=_CONFIG_LABEL, 

113 ) 

114 chat_ctx = client_chat_ctx(port) 

115 warn_if_below_hermes_minimum(chat_ctx) 

116 fragment = hermes_config( 

117 base_url=f"http://{LOOPBACK}:{port}", 

118 api_key=_TOKEN_REF, 

119 model_refs=model_refs, 

120 default_ref=str(cfg.chat_model), 

121 chat_ctx=chat_ctx, 

122 include_mcp=self._include_mcp, 

123 ) 

124 deep_merge(config, fragment) 

125 if not self._include_mcp: 

126 prune_lilbee(config, _MCP_CONTAINER_KEY) 

127 config_file.atomic_write_text( 

128 _hermes_config_path(), yaml.safe_dump(config, sort_keys=False) 

129 ) 

130 _upsert_env_token(_hermes_env_path(), token) 

131 if self._include_mcp: 

132 install_bundled_skill(_hermes_skill_dest()) 

133 # hermes ships HTTP MCP behind the optional `mcp` extra; without it 

134 # lilbee's MCP search shows "0 connected". Set it up before launch, 

135 # honoring hermes's own auto-install security gate. 

136 # _binary is non-None: run_launcher called find_binary() and gated on it. 

137 if self._binary is not None: 

138 allow_lazy = bool( 

139 (config.get(_SECURITY_KEY) or {}).get(_ALLOW_LAZY_INSTALLS_KEY, True) 

140 ) 

141 mcp_ready = ensure_hermes_http_mcp( 

142 self._binary, allow_lazy_installs=allow_lazy, echo=typer.echo 

143 ) 

144 # Requested but not connected: don't hand off a silently ungrounded hermes. 

145 if not mcp_ready: 

146 warn_hermes_ungrounded() 

147 return ([], {**os.environ, LILBEE_TOKEN_ENV_VAR: token}) 

148 

149 

150def hermes_cmd( 

151 mcp: bool | None = typer.Option( 

152 None, 

153 "--mcp/--no-mcp", 

154 help="Register lilbee's MCP search tool into hermes. Defaults to the " 

155 "agent_mcp_enabled config; --mcp/--no-mcp overrides it for this launch.", 

156 ), 

157) -> None: 

158 """Launch hermes with lilbee registered as its model provider.""" 

159 include_mcp = cfg.agent_mcp_enabled if mcp is None else mcp 

160 run_launcher(HermesLauncher(include_mcp=include_mcp))