diff --git a/mcp_serve.py b/mcp_serve.py index d757b9cf2e..95eebded68 100644 --- a/mcp_serve.py +++ b/mcp_serve.py @@ -1,30 +1,10 @@ """ -Hermes MCP Server — expose messaging conversations as MCP tools. +Hermes MCP Server — expose messaging conversations as MCP tools (`hermes mcp serve`). -Starts a stdio MCP server that lets any MCP client (Claude Code, Cursor, Codex, -etc.) list conversations, read message history, send messages, poll for live -events, and manage approval requests across all connected platforms. - -Matches OpenClaw's 9-tool MCP channel bridge surface: - conversations_list, conversation_get, messages_read, attachments_fetch, - events_poll, events_wait, messages_send, permissions_list_open, - permissions_respond - -Plus: channels_list (Hermes-specific extra) - -Usage: - hermes mcp serve - hermes mcp serve --verbose - -MCP client config (e.g. claude_desktop_config.json): - { - "mcpServers": { - "hermes": { - "command": "hermes", - "args": ["mcp", "serve"] - } - } - } +A stdio MCP server letting any MCP client (Claude Code, Cursor, Codex, ...) list +conversations, read history, send messages, poll live events, and manage approvals. +Matches OpenClaw's 9-tool channel bridge surface plus the Hermes-specific +channels_list. Client config: {"mcpServers": {"hermes": {"command": "hermes", "args": ["mcp", "serve"]}}} """ from __future__ import annotations @@ -43,13 +23,9 @@ from typing import Dict, List, Optional logger = logging.getLogger("hermes.mcp_serve") -# --------------------------------------------------------------------------- -# Lazy MCP SDK import -# --------------------------------------------------------------------------- - -# mcp 2.0 removed `mcp.server.fastmcp`; its decorator-driven server is now -# `mcp.server.MCPServer` with the same `@server.tool()` / `run_stdio_async()` -# surface (docstring -> tool description, signature -> input schema). +# mcp 2.0 removed `mcp.server.fastmcp`; `mcp.server.MCPServer` keeps the same +# `@server.tool()` / `run_stdio_async()` surface (docstring -> description, +# signature -> input schema). _MCP_SERVER_AVAILABLE = False try: from mcp.server import MCPServer @@ -59,21 +35,49 @@ except ImportError: MCPServer = None # type: ignore[assignment,misc] -# --------------------------------------------------------------------------- -# Helpers -# --------------------------------------------------------------------------- +# --- Helpers ----------------------------------------------------------------- -def _get_sessions_dir() -> Path: - """Return the sessions directory using HERMES_HOME.""" +def _hermes_home() -> Path: try: from hermes_constants import get_hermes_home - return get_hermes_home() / "sessions" + return get_hermes_home() except ImportError: - return Path(os.environ.get("HERMES_HOME", Path.home() / ".hermes")) / "sessions" + return Path(os.environ.get("HERMES_HOME", Path.home() / ".hermes")) + + +def _get_sessions_dir() -> Path: + return _hermes_home() / "sessions" + + +def _read_state_db_mtime() -> float: + db_file = _hermes_home() / "state.db" + try: + return db_file.stat().st_mtime if db_file.exists() else 0.0 + except OSError: + return 0.0 + + +def _read_json(path: Path): + """Parsed JSON file, or {} when missing/unreadable.""" + if not path.exists(): + return {} + try: + with open(path, "r", encoding="utf-8") as f: + return json.load(f) + except Exception as e: + logger.debug("Failed to load %s: %s", path.name, e) + return {} + + +def _close_quietly(db, what: str) -> None: + try: + db.close() + except Exception: + logger.debug("Failed to close MCP %s SessionDB", what, exc_info=True) def _get_session_db(): - """Get a SessionDB instance for reading message transcripts.""" + """SessionDB instance for reading message transcripts, or None.""" try: from hermes_state import get_shared_session_db return get_shared_session_db() @@ -83,7 +87,7 @@ def _get_session_db(): def _load_session_messages(session_id: str): - """Read one session and close the temporary database handle.""" + """(messages, error) for one session; closes the temporary database handle.""" db = _get_session_db() if db is None: return None, "Session database unavailable" @@ -100,62 +104,45 @@ def _load_session_messages(session_id: str): def _load_sessions_index() -> dict: - """Load the gateway session routing index. + """Gateway routing index: session_key -> entry dict. - Returns a dict of session_key -> entry_dict with platform routing info. - - state.db is the primary source (#9006): gateway sessions persist their - routing metadata (session_key, chat/thread ids, display_name, origin) on - the durable session row, so a single database read replaces the old - dual-file sessions.json dependency. Falls back to sessions.json for - pre-migration databases where no gateway rows carry a session_key yet. + state.db is primary (gateway session rows carry session_key/origin metadata); + sessions.json is the fallback for pre-migration databases without session_keys. """ - entries = _load_sessions_index_from_db() - if entries: - return entries - return _load_sessions_index_from_json() + return _load_sessions_index_from_db() or _load_sessions_index_from_json() + + +def _iso(ts) -> str: + try: + return datetime.fromtimestamp(float(ts)).isoformat() if ts else "" + except (TypeError, ValueError, OSError): + return "" def _row_to_index_entry(row: dict) -> dict: """Convert a state.db gateway session row to the sessions.json entry shape.""" origin = {} - origin_json = row.get("origin_json") - if origin_json: + if row.get("origin_json"): try: - parsed = json.loads(origin_json) + parsed = json.loads(row["origin_json"]) if isinstance(parsed, dict): origin = parsed except (TypeError, ValueError): pass - if not origin: - # Pre-origin_json rows: synthesize the minimal origin from columns. - origin = { - "platform": row.get("source", ""), - "chat_id": row.get("chat_id"), - "chat_type": row.get("chat_type"), - "thread_id": row.get("thread_id"), - "user_id": row.get("user_id"), - } - - def _iso(ts) -> str: - try: - return datetime.fromtimestamp(float(ts)).isoformat() if ts else "" - except (TypeError, ValueError, OSError): - return "" + if not origin: # pre-origin_json rows: synthesize the minimal origin from columns + origin = {"platform": row.get("source", ""), **{k: row.get(k) for k in ("chat_id", "chat_type", "thread_id", "user_id")}} input_tokens = int(row.get("input_tokens") or 0) output_tokens = int(row.get("output_tokens") or 0) return { - "session_id": str(row.get("id", "")), - "session_key": row.get("session_key", ""), + "session_id": str(row.get("id", "")), "session_key": row.get("session_key", ""), "platform": row.get("source", ""), "chat_type": row.get("chat_type") or origin.get("chat_type", ""), "display_name": row.get("display_name") or origin.get("chat_name") or "", "origin": origin, "created_at": _iso(row.get("started_at")), "updated_at": _iso(row.get("last_active") or row.get("started_at")), - "input_tokens": input_tokens, - "output_tokens": output_tokens, + "input_tokens": input_tokens, "output_tokens": output_tokens, "total_tokens": input_tokens + output_tokens, } @@ -169,14 +156,7 @@ def _load_sessions_index_from_db() -> dict: lister = getattr(db, "list_gateway_sessions", None) if not callable(lister): return {} - rows = lister(active_only=True) - entries = {} - for row in rows: - key = row.get("session_key") - if not key: - continue - entries[key] = _row_to_index_entry(row) - return entries + return {row["session_key"]: _row_to_index_entry(row) for row in lister(active_only=True) if row.get("session_key")} except Exception as e: logger.debug("Failed to load gateway sessions from state.db: %s", e) return {} @@ -188,62 +168,22 @@ def _load_sessions_index_from_db() -> dict: def _load_sessions_index_from_json() -> dict: - """Legacy fallback: load the gateway sessions.json index directly. - - Used only for pre-migration databases whose gateway rows don't carry a - session_key yet. This avoids importing the full SessionStore which - needs GatewayConfig. - """ - sessions_file = _get_sessions_dir() / "sessions.json" - if not sessions_file.exists(): - return {} - try: - with open(sessions_file, "r", encoding="utf-8") as f: - data = json.load(f) - # Drop documentation/metadata sentinels (keys starting with "_", e.g. - # the "_README" note the gateway writes into the index). They are not - # session entries and would break consumers that treat every value as - # an entry dict. - if isinstance(data, dict): - return {k: v for k, v in data.items() if not str(k).startswith("_")} - return {} - except Exception as e: - logger.debug("Failed to load sessions.json: %s", e) - return {} + """Legacy fallback: read sessions.json directly (avoids importing SessionStore, + which needs GatewayConfig). Keys starting with "_" are metadata sentinels + (e.g. "_README"), not session entries.""" + data = _read_json(_get_sessions_dir() / "sessions.json") + if isinstance(data, dict): + return {k: v for k, v in data.items() if not str(k).startswith("_")} + return {} def _load_channel_directory() -> dict: """Load the cached channel directory for available targets.""" - try: - from hermes_constants import get_hermes_home - directory_file = get_hermes_home() / "channel_directory.json" - except ImportError: - directory_file = Path( - os.environ.get("HERMES_HOME", Path.home() / ".hermes") - ) / "channel_directory.json" - - if not directory_file.exists(): - return {} - try: - with open(directory_file, "r", encoding="utf-8") as f: - return json.load(f) - except Exception as e: - logger.debug("Failed to load channel_directory.json: %s", e) - return {} + return _read_json(_hermes_home() / "channel_directory.json") -def _coerce_int( - value, - *, - default: int, - minimum: int, - maximum: int, -) -> int: - """Coerce value to int with fallback and clamping. - - Used at MCP tool boundaries to handle invalid types from external clients. - Returns default if value cannot be converted to int. - """ +def _coerce_int(value, *, default: int, minimum: int, maximum: int) -> int: + """Clamped int for MCP tool boundaries; *default* when the client sent an unconvertible value.""" try: coerced = int(value) except (TypeError, ValueError): @@ -255,55 +195,42 @@ def _extract_message_content(msg: dict) -> str: """Extract text content from a message, handling multi-part content.""" content = msg.get("content", "") if isinstance(content, list): - text_parts = [ + return "\n".join( p.get("text", "") for p in content if isinstance(p, dict) and p.get("type") == "text" - ] - return "\n".join(text_parts) + ) return str(content) if content else "" def _extract_attachments(msg: dict) -> List[dict]: - """Extract non-text attachments from a message. - - Finds: multi-part image/file content blocks, MEDIA: tags in text, - image URLs, and file references. - """ + """Non-text attachments: image/file content blocks plus MEDIA: tags in the text.""" attachments = [] content = msg.get("content", "") - # Multi-part content blocks (image_url, file, etc.) - if isinstance(content, list): - for part in content: - if not isinstance(part, dict): - continue - ptype = part.get("type", "") - if ptype == "image_url": - url = part.get("image_url", {}).get("url", "") if isinstance(part.get("image_url"), dict) else "" - if url: - attachments.append({"type": "image", "url": url}) - elif ptype == "image": - url = part.get("url", part.get("source", {}).get("url", "")) - if url: - attachments.append({"type": "image", "url": url}) - elif ptype not in {"text",}: - # Unknown non-text content type + for part in content if isinstance(content, list) else (): + if not isinstance(part, dict): + continue + ptype = part.get("type", "") + if ptype == "image_url": + url = part.get("image_url", {}).get("url", "") if isinstance(part.get("image_url"), dict) else "" + elif ptype == "image": + url = part.get("url", part.get("source", {}).get("url", "")) + else: + if ptype != "text": attachments.append({"type": ptype, "data": part}) + continue + if url: + attachments.append({"type": "image", "url": url}) - # MEDIA: tags in text content text = _extract_message_content(msg) if text: - media_pattern = re.compile(r'MEDIA:\s*(\S+)') - for match in media_pattern.finditer(text): - path = match.group(1) - attachments.append({"type": "media", "path": path}) + for match in re.finditer(r'MEDIA:\s*(\S+)', text): + attachments.append({"type": "media", "path": match.group(1)}) return attachments -# --------------------------------------------------------------------------- -# Event Bridge — polls SessionDB for new messages, maintains event queue -# --------------------------------------------------------------------------- +# --- Event Bridge — polls SessionDB for new messages, maintains event queue --- QUEUE_LIMIT = 1000 POLL_INTERVAL = 0.2 # seconds between DB polls (200ms) @@ -317,29 +244,34 @@ class QueueEvent: session_key: str = "" data: dict = field(default_factory=dict) + def as_dict(self) -> dict: + return {"cursor": self.cursor, "type": self.type, "session_key": self.session_key, **self.data} + def _ts_float(ts) -> float: """Normalize a message timestamp (epoch int/float or ISO string) to float.""" if isinstance(ts, (int, float)): return float(ts) - if isinstance(ts, str) and ts: + if not (isinstance(ts, str) and ts): + return 0.0 + try: + return float(ts) + except ValueError: try: - return float(ts) - except ValueError: - try: - return datetime.fromisoformat(ts).timestamp() - except Exception: - return 0.0 - return 0.0 + return datetime.fromisoformat(ts).timestamp() + except Exception: + return 0.0 + + +def _latest_ts(messages) -> float: + """Newest normalized timestamp among *messages* (0.0 when none).""" + return max((_ts_float(m.get("timestamp", 0)) for m in (messages or ())), default=0.0) class EventBridge: - """Background poller that watches SessionDB for new messages and - maintains an in-memory event queue with waiter support. - - This is the Hermes equivalent of OpenClaw's WebSocket gateway bridge. - Instead of WebSocket events, we poll the SQLite database for changes. - """ + """Background poller watching SessionDB for new messages, feeding an in-memory + event queue with waiter support (the Hermes analogue of OpenClaw's WebSocket + gateway bridge, polling SQLite instead).""" def __init__(self): self._queue: List[QueueEvent] = [] @@ -349,22 +281,17 @@ class EventBridge: self._running = False self._thread: Optional[threading.Thread] = None self._last_poll_timestamps: Dict[str, float] = {} # session_key -> unix timestamp - # In-memory approval tracking (populated from events) - self._pending_approvals: Dict[str, dict] = {} - # mtime cache — skip expensive work when state.db hasn't changed - self._state_db_mtime: float = 0.0 + self._pending_approvals: Dict[str, dict] = {} # populated from events + self._state_db_mtime: float = 0.0 # skip polling work when state.db is unchanged self._cached_sessions_index: dict = {} def start(self): """Start the background polling thread.""" if self._running: return - # Snapshot existing history BEFORE the poll loop starts so pre-existing - # messages are not replayed as new events on startup (#13414). Sessions - # that first appear afterwards are absent from the baseline and default - # to last_seen=0.0 in _poll_once, so new-conversation delivery is - # preserved. Unit tests that drive _poll_once directly bypass start() - # and still observe first-poll delivery. + # Baseline existing history BEFORE polling so startup never replays old + # messages as events; sessions appearing later default to last_seen=0.0 + # in _poll_once, so new-conversation delivery is preserved. self._establish_baseline() self._running = True self._thread = threading.Thread(target=self._poll_loop, daemon=True) @@ -372,132 +299,79 @@ class EventBridge: logger.debug("EventBridge started") def stop(self): - """Stop the background polling thread.""" + """Stop the background polling thread and wake any waiters.""" self._running = False - self._new_event.set() # Wake any waiters + self._new_event.set() if self._thread: self._thread.join(timeout=5) logger.debug("EventBridge stopped") - def poll_events( - self, - after_cursor: int = 0, - session_key: Optional[str] = None, - limit: int = 20, - ) -> dict: - """Return events since after_cursor, optionally filtered by session_key.""" + def _matching(self, after_cursor: int, session_key: Optional[str], limit: int) -> List[dict]: with self._lock: - events = [ - e for e in self._queue - if e.cursor > after_cursor - and (not session_key or e.session_key == session_key) + return [ + e.as_dict() for e in self._queue + if e.cursor > after_cursor and (not session_key or e.session_key == session_key) ][:limit] - next_cursor = events[-1].cursor if events else after_cursor - return { - "events": [ - {"cursor": e.cursor, "type": e.type, - "session_key": e.session_key, **e.data} - for e in events - ], - "next_cursor": next_cursor, - } + def poll_events(self, after_cursor: int = 0, session_key: Optional[str] = None, limit: int = 20) -> dict: + """Return events since after_cursor, optionally filtered by session_key.""" + events = self._matching(after_cursor, session_key, limit) + return {"events": events, "next_cursor": events[-1]["cursor"] if events else after_cursor} - def wait_for_event( - self, - after_cursor: int = 0, - session_key: Optional[str] = None, - timeout_ms: int = 30000, - ) -> Optional[dict]: + def wait_for_event(self, after_cursor: int = 0, session_key: Optional[str] = None, timeout_ms: int = 30000) -> Optional[dict]: """Block until a matching event arrives or timeout expires.""" deadline = time.monotonic() + (timeout_ms / 1000.0) - while time.monotonic() < deadline: - with self._lock: - for e in self._queue: - if e.cursor > after_cursor and ( - not session_key or e.session_key == session_key - ): - return { - "cursor": e.cursor, "type": e.type, - "session_key": e.session_key, **e.data, - } - + found = self._matching(after_cursor, session_key, 1) + if found: + return found[0] remaining = deadline - time.monotonic() if remaining <= 0: break self._new_event.clear() self._new_event.wait(timeout=min(remaining, POLL_INTERVAL)) - return None def list_pending_approvals(self) -> List[dict]: """List approval requests observed during this bridge session.""" with self._lock: - return sorted( - self._pending_approvals.values(), - key=lambda a: a.get("created_at", ""), - ) + return sorted(self._pending_approvals.values(), key=lambda a: a.get("created_at", "")) def respond_to_approval(self, approval_id: str, decision: str) -> dict: """Resolve a pending approval (best-effort without gateway IPC).""" with self._lock: approval = self._pending_approvals.pop(approval_id, None) - if not approval: return {"error": f"Approval not found: {approval_id}"} - - self._enqueue(QueueEvent( - cursor=0, # Will be set by _enqueue - type="approval_resolved", - session_key=approval.get("session_key", ""), - data={"approval_id": approval_id, "decision": decision}, + self._enqueue(QueueEvent( # cursor is assigned by _enqueue + 0, "approval_resolved", approval.get("session_key", ""), + {"approval_id": approval_id, "decision": decision}, )) - return {"resolved": True, "approval_id": approval_id, "decision": decision} def _enqueue(self, event: QueueEvent) -> None: - """Add an event to the queue and wake any waiters.""" + """Add an event to the queue (trimmed to QUEUE_LIMIT) and wake any waiters.""" with self._lock: self._cursor += 1 event.cursor = self._cursor self._queue.append(event) - # Trim queue to limit while len(self._queue) > QUEUE_LIMIT: self._queue.pop(0) self._new_event.set() def _establish_baseline(self) -> None: db = _get_session_db() - if not db: - return - try: - self._establish_baseline_with_db(db) - finally: + if db: try: - db.close() - except Exception: - logger.debug("Failed to close MCP baseline SessionDB", exc_info=True) + self._establish_baseline_with_db(db) + finally: + _close_quietly(db, "baseline") def _establish_baseline_with_db(self, db) -> None: - """Record the latest per-session message timestamp and the current - state.db mtime WITHOUT emitting events, so startup does not replay - history (#13414). - - Only sessions that already exist at startup are baselined; a session - that first appears afterwards is absent here and defaults to - last_seen=0.0 in _poll_once, so a brand-new conversation's first - message is still delivered on its state.db-change tick. - """ - try: - from hermes_constants import get_hermes_home - db_file = get_hermes_home() / "state.db" - except ImportError: - db_file = Path(os.environ.get("HERMES_HOME", Path.home() / ".hermes")) / "state.db" - try: - self._state_db_mtime = db_file.stat().st_mtime if db_file.exists() else 0.0 - except OSError: - self._state_db_mtime = 0.0 + """Record per-session latest timestamps and the state.db mtime WITHOUT + emitting events. Only sessions existing now are baselined; later ones + default to last_seen=0.0 in _poll_once, so their first message is delivered.""" + self._state_db_mtime = _read_state_db_mtime() try: self._cached_sessions_index = _load_sessions_index() except Exception: @@ -507,21 +381,18 @@ class EventBridge: if not session_id: continue try: - messages = db.get_messages(session_id) + latest = _latest_ts(db.get_messages(session_id)) except Exception: continue - all_ts = [_ts_float(m.get("timestamp", 0)) for m in (messages or ())] - if all_ts: - latest = max(all_ts) - if latest > 0.0: - self._last_poll_timestamps[session_key] = latest + if latest > 0.0: + self._last_poll_timestamps[session_key] = latest + def _poll_loop(self): """Background loop: poll SessionDB for new messages.""" db = _get_session_db() if not db: logger.warning("EventBridge: SessionDB unavailable, event polling disabled") return - try: while self._running: try: @@ -530,124 +401,84 @@ class EventBridge: logger.debug("EventBridge poll error: %s", e) time.sleep(POLL_INTERVAL) finally: - try: - db.close() - except Exception: - logger.debug("Failed to close MCP polling SessionDB", exc_info=True) + _close_quietly(db, "polling") def _poll_once(self, db): """Check for new messages across all sessions. - Uses a single mtime check on state.db to skip work when nothing - has changed — makes 200ms polling essentially free. Since #9006 - the routing index itself lives in state.db (session rows carry - session_key/origin metadata), so a new conversation and its first - message land in the SAME file and one mtime check covers both — - eliminating the old dual-file (sessions.json + state.db) race that - could drop brand-new conversations (#8925). + One state.db mtime check gates all work, making 200ms polling nearly free. + The routing index lives in the same file as the messages, so a new + conversation and its first message land under a single mtime change (no + dual-file race that could drop brand-new conversations). """ - try: - from hermes_constants import get_hermes_home - db_file = get_hermes_home() / "state.db" - except ImportError: - db_file = Path(os.environ.get("HERMES_HOME", Path.home() / ".hermes")) / "state.db" - - try: - db_mtime = db_file.stat().st_mtime if db_file.exists() else 0.0 - except OSError: - db_mtime = 0.0 - + db_mtime = _read_state_db_mtime() if db_mtime == self._state_db_mtime: - return # Nothing changed since last poll — skip entirely - + return self._state_db_mtime = db_mtime - # Refresh the routing index from state.db on every change tick — - # it's a single indexed query and it can never lag the messages - # table (both live in the same database file). + # Refresh the index on every change tick: one indexed query, never lags messages. self._cached_sessions_index = _load_sessions_index() - entries = self._cached_sessions_index - for session_key, entry in entries.items(): + for session_key, entry in self._cached_sessions_index.items(): session_id = entry.get("session_id", "") if not session_id: continue - last_seen = self._last_poll_timestamps.get(session_key, 0.0) - try: messages = db.get_messages(session_id) except Exception: continue - if not messages: continue - # Find messages newer than our last seen timestamp (see the - # module-level _ts_float helper for timestamp normalization). - new_messages = [] for msg in messages: - ts = _ts_float(msg.get("timestamp", 0)) - role = msg.get("role", "") - if role not in {"user", "assistant"}: + if msg.get("role", "") not in {"user", "assistant"}: + continue + if _ts_float(msg.get("timestamp", 0)) <= last_seen: continue - if ts > last_seen: - new_messages.append(msg) - - for msg in new_messages: content = _extract_message_content(msg) if not content: continue - self._enqueue(QueueEvent( - cursor=0, - type="message", - session_key=session_key, - data={ - "role": msg.get("role", ""), - "content": content[:500], - "timestamp": str(msg.get("timestamp", "")), - "message_id": str(msg.get("id", "")), - }, - )) + self._enqueue(QueueEvent(0, "message", session_key, { + "role": msg.get("role", ""), "content": content[:500], + "timestamp": str(msg.get("timestamp", "")), "message_id": str(msg.get("id", "")), + })) - # Update last seen to the most recent message timestamp - all_ts = [_ts_float(m.get("timestamp", 0)) for m in messages] - if all_ts: - latest = max(all_ts) - if latest > last_seen: - self._last_poll_timestamps[session_key] = latest + latest = _latest_ts(messages) + if latest > last_seen: + self._last_poll_timestamps[session_key] = latest -# --------------------------------------------------------------------------- -# MCP Server -# --------------------------------------------------------------------------- +# --- MCP Server --------------------------------------------------------------- -def create_mcp_server(event_bridge: Optional[EventBridge] = None) -> "MCPServer": - """Create and return the Hermes MCP server with all tools registered.""" - if not _MCP_SERVER_AVAILABLE: - raise ImportError( - "MCP server requires the 'mcp' package. " - f"Install with: {sys.executable} -m pip install 'mcp'" - ) +def _conversation_messages(session_key: str): + """(messages, error_json) for a conversation; exactly one is None.""" + entry = _load_sessions_index().get(session_key) + if not entry: + return None, json.dumps({"error": f"Conversation not found: {session_key}"}) + session_id = entry.get("session_id", "") + if not session_id: + return None, json.dumps({"error": "No session ID for this conversation"}) + messages, error = _load_session_messages(session_id) + if error: + return None, json.dumps({"error": error}) + return messages, None - mcp = MCPServer( - "hermes", - instructions=( - "Hermes Agent messaging bridge. Use these tools to interact with " - "conversations across Telegram, Discord, Slack, WhatsApp, Signal, " - "Matrix, and other connected platforms." - ), - ) - bridge = event_bridge or EventBridge() +def _platform_matches(wanted: Optional[str], actual: str) -> bool: + return not wanted or actual.lower() == wanted.lower() - # -- conversations_list ------------------------------------------------ - @mcp.tool() - def conversations_list( - platform: Optional[str] = None, - limit: int = 50, - search: Optional[str] = None, - ) -> str: +class _ToolHandlers: + """The MCP tool handlers; each method named in _TOOL_NAMES is registered as one tool. + + Method docstrings are the wire-format tool descriptions and signatures the + input schemas — do not reword or reflow them. + """ + + def __init__(self, bridge: EventBridge): + self.bridge = bridge + + def conversations_list(self, platform: Optional[str] = None, limit: int = 50, search: Optional[str] = None) -> str: """List active messaging conversations across connected platforms. Returns conversations with their session keys (needed for messages_read), @@ -659,59 +490,36 @@ def create_mcp_server(event_bridge: Optional[EventBridge] = None) -> "MCPServer" search: Optional text to filter conversations by name """ limit = _coerce_int(limit, default=50, minimum=1, maximum=200) - entries = _load_sessions_index() conversations = [] - - for key, entry in entries.items(): + for key, entry in _load_sessions_index().items(): origin = entry.get("origin", {}) entry_platform = entry.get("platform") or origin.get("platform", "") - - if platform and entry_platform.lower() != platform.lower(): + if not _platform_matches(platform, entry_platform): continue - display_name = entry.get("display_name", "") chat_name = origin.get("chat_name", "") - if search: - search_lower = search.lower() - if (search_lower not in display_name.lower() - and search_lower not in chat_name.lower() - and search_lower not in key.lower()): - continue - + if search and not any(search.lower() in s.lower() for s in (display_name, chat_name, key)): + continue conversations.append({ - "session_key": key, - "session_id": entry.get("session_id", ""), - "platform": entry_platform, + "session_key": key, "session_id": entry.get("session_id", ""), "platform": entry_platform, "chat_type": entry.get("chat_type", origin.get("chat_type", "")), - "display_name": display_name, - "chat_name": chat_name, - "user_name": origin.get("user_name", ""), - "updated_at": entry.get("updated_at", ""), + "display_name": display_name, "chat_name": chat_name, + "user_name": origin.get("user_name", ""), "updated_at": entry.get("updated_at", ""), }) conversations.sort(key=lambda c: c.get("updated_at", ""), reverse=True) conversations = conversations[:limit] + return json.dumps({"count": len(conversations), "conversations": conversations}, indent=2) - return json.dumps({ - "count": len(conversations), - "conversations": conversations, - }, indent=2) - - # -- conversation_get -------------------------------------------------- - - @mcp.tool() - def conversation_get(session_key: str) -> str: + def conversation_get(self, session_key: str) -> str: """Get detailed info about one conversation by its session key. Args: session_key: The session key from conversations_list """ - entries = _load_sessions_index() - entry = entries.get(session_key) - + entry = _load_sessions_index().get(session_key) if not entry: return json.dumps({"error": f"Conversation not found: {session_key}"}) - origin = entry.get("origin", {}) return json.dumps({ "session_key": session_key, @@ -719,24 +527,14 @@ def create_mcp_server(event_bridge: Optional[EventBridge] = None) -> "MCPServer" "platform": entry.get("platform") or origin.get("platform", ""), "chat_type": entry.get("chat_type", origin.get("chat_type", "")), "display_name": entry.get("display_name", ""), - "user_name": origin.get("user_name", ""), - "chat_name": origin.get("chat_name", ""), - "chat_id": origin.get("chat_id", ""), - "thread_id": origin.get("thread_id"), - "updated_at": entry.get("updated_at", ""), - "created_at": entry.get("created_at", ""), - "input_tokens": entry.get("input_tokens", 0), - "output_tokens": entry.get("output_tokens", 0), + "user_name": origin.get("user_name", ""), "chat_name": origin.get("chat_name", ""), + "chat_id": origin.get("chat_id", ""), "thread_id": origin.get("thread_id"), + "updated_at": entry.get("updated_at", ""), "created_at": entry.get("created_at", ""), + "input_tokens": entry.get("input_tokens", 0), "output_tokens": entry.get("output_tokens", 0), "total_tokens": entry.get("total_tokens", 0), }, indent=2) - # -- messages_read ----------------------------------------------------- - - @mcp.tool() - def messages_read( - session_key: str, - limit: int = 50, - ) -> str: + def messages_read(self, session_key: str, limit: int = 50) -> str: """Read recent messages from a conversation. Returns the message history in chronological order with role, content, @@ -747,48 +545,25 @@ def create_mcp_server(event_bridge: Optional[EventBridge] = None) -> "MCPServer" limit: Maximum number of messages to return (default 50, most recent) """ limit = _coerce_int(limit, default=50, minimum=1, maximum=200) - entries = _load_sessions_index() - entry = entries.get(session_key) - if not entry: - return json.dumps({"error": f"Conversation not found: {session_key}"}) - - session_id = entry.get("session_id", "") - if not session_id: - return json.dumps({"error": "No session ID for this conversation"}) - - all_messages, error = _load_session_messages(session_id) + all_messages, error = _conversation_messages(session_key) if error: - return json.dumps({"error": error}) - + return error filtered = [] for msg in all_messages: role = msg.get("role", "") - if role in {"user", "assistant"}: - content = _extract_message_content(msg) - if content: - filtered.append({ - "id": str(msg.get("id", "")), - "role": role, - "content": content[:2000], - "timestamp": msg.get("timestamp", ""), - }) - + content = _extract_message_content(msg) if role in {"user", "assistant"} else "" + if content: + filtered.append({ + "id": str(msg.get("id", "")), "role": role, + "content": content[:2000], "timestamp": msg.get("timestamp", ""), + }) messages = filtered[-limit:] - return json.dumps({ - "session_key": session_key, - "count": len(messages), - "total_in_session": len(filtered), - "messages": messages, + "session_key": session_key, "count": len(messages), + "total_in_session": len(filtered), "messages": messages, }, indent=2) - # -- attachments_fetch ------------------------------------------------- - - @mcp.tool() - def attachments_fetch( - session_key: str, - message_id: str, - ) -> str: + def attachments_fetch(self, session_key: str, message_id: str) -> str: """List non-text attachments for a message in a conversation. Extracts images, media files, and other non-text content blocks @@ -798,45 +573,16 @@ def create_mcp_server(event_bridge: Optional[EventBridge] = None) -> "MCPServer" session_key: The session key from conversations_list message_id: The message ID from messages_read """ - entries = _load_sessions_index() - entry = entries.get(session_key) - if not entry: - return json.dumps({"error": f"Conversation not found: {session_key}"}) - - session_id = entry.get("session_id", "") - if not session_id: - return json.dumps({"error": "No session ID for this conversation"}) - - all_messages, error = _load_session_messages(session_id) + all_messages, error = _conversation_messages(session_key) if error: - return json.dumps({"error": error}) - - # Find the target message - target_msg = None - for msg in all_messages: - if str(msg.get("id", "")) == message_id: - target_msg = msg - break - + return error + target_msg = next((m for m in all_messages if str(m.get("id", "")) == message_id), None) if not target_msg: return json.dumps({"error": f"Message not found: {message_id}"}) - attachments = _extract_attachments(target_msg) + return json.dumps({"message_id": message_id, "count": len(attachments), "attachments": attachments}, indent=2) - return json.dumps({ - "message_id": message_id, - "count": len(attachments), - "attachments": attachments, - }, indent=2) - - # -- events_poll ------------------------------------------------------- - - @mcp.tool() - def events_poll( - after_cursor: int = 0, - session_key: Optional[str] = None, - limit: int = 20, - ) -> str: + def events_poll(self, after_cursor: int = 0, session_key: Optional[str] = None, limit: int = 20) -> str: """Poll for new conversation events since a cursor position. Returns events that have occurred since the given cursor. Use the @@ -851,21 +597,10 @@ def create_mcp_server(event_bridge: Optional[EventBridge] = None) -> "MCPServer" """ after_cursor = _coerce_int(after_cursor, default=0, minimum=0, maximum=10**18) limit = _coerce_int(limit, default=20, minimum=1, maximum=200) - result = bridge.poll_events( - after_cursor=after_cursor, - session_key=session_key, - limit=limit, - ) + result = self.bridge.poll_events(after_cursor=after_cursor, session_key=session_key, limit=limit) return json.dumps(result, indent=2) - # -- events_wait ------------------------------------------------------- - - @mcp.tool() - def events_wait( - after_cursor: int = 0, - session_key: Optional[str] = None, - timeout_ms: int = 30000, - ) -> str: + def events_wait(self, after_cursor: int = 0, session_key: Optional[str] = None, timeout_ms: int = 30000) -> str: """Wait for the next conversation event (long-poll). Blocks until a matching event arrives or the timeout expires. @@ -877,28 +612,13 @@ def create_mcp_server(event_bridge: Optional[EventBridge] = None) -> "MCPServer" timeout_ms: Maximum wait time in milliseconds (default 30000) """ after_cursor = _coerce_int(after_cursor, default=0, minimum=0, maximum=10**18) - timeout_ms = _coerce_int( - timeout_ms, - default=30000, - minimum=0, - maximum=300000, - ) # Cap at 5 minutes - event = bridge.wait_for_event( - after_cursor=after_cursor, - session_key=session_key, - timeout_ms=timeout_ms, - ) + timeout_ms = _coerce_int(timeout_ms, default=30000, minimum=0, maximum=300000) # cap 5 min + event = self.bridge.wait_for_event(after_cursor=after_cursor, session_key=session_key, timeout_ms=timeout_ms) if event: return json.dumps({"event": event}, indent=2) return json.dumps({"event": None, "reason": "timeout"}, indent=2) - # -- messages_send ----------------------------------------------------- - - @mcp.tool() - def messages_send( - target: str, - message: str, - ) -> str: + def messages_send(self, target: str, message: str) -> str: """Send a message to a platform conversation. The target format is "platform:chat_id" — same format used by the @@ -916,22 +636,15 @@ def create_mcp_server(event_bridge: Optional[EventBridge] = None) -> "MCPServer" """ if not target or not message: return json.dumps({"error": "Both target and message are required"}) - try: from tools.send_message_tool import send_message_tool - result_str = send_message_tool( - {"action": "send", "target": target, "message": message} - ) - return result_str + return send_message_tool({"action": "send", "target": target, "message": message}) except ImportError: return json.dumps({"error": "Send message tool not available"}) except Exception as e: return json.dumps({"error": f"Send failed: {e}"}) - # -- channels_list ----------------------------------------------------- - - @mcp.tool() - def channels_list(platform: Optional[str] = None) -> str: + def channels_list(self, platform: Optional[str] = None) -> str: """List available messaging channels and targets across platforms. Returns channels that you can send messages to. The target strings @@ -942,24 +655,21 @@ def create_mcp_server(event_bridge: Optional[EventBridge] = None) -> "MCPServer" """ directory = _load_channel_directory() if not directory: - entries = _load_sessions_index() + # No cached directory: derive send targets from the routing index. targets = [] seen = set() - for key, entry in entries.items(): + for key, entry in _load_sessions_index().items(): origin = entry.get("origin", {}) p = entry.get("platform") or origin.get("platform", "") chat_id = origin.get("chat_id", "") - if not p or not chat_id: - continue - if platform and p.lower() != platform.lower(): + if not p or not chat_id or not _platform_matches(platform, p): continue target_str = f"{p}:{chat_id}" if target_str in seen: continue seen.add(target_str) targets.append({ - "target": target_str, - "platform": p, + "target": target_str, "platform": p, "name": entry.get("display_name") or origin.get("chat_name", ""), "chat_type": entry.get("chat_type", origin.get("chat_type", "")), }) @@ -967,44 +677,28 @@ def create_mcp_server(event_bridge: Optional[EventBridge] = None) -> "MCPServer" channels = [] for plat, entries_list in directory.get("platforms", {}).items(): - if platform and plat.lower() != platform.lower(): + if not _platform_matches(platform, plat) or not isinstance(entries_list, list): continue - if isinstance(entries_list, list): - for ch in entries_list: - if isinstance(ch, dict): - chat_id = ch.get("id", ch.get("chat_id", "")) - channels.append({ - "target": f"{plat}:{chat_id}" if chat_id else plat, - "platform": plat, - "name": ch.get("name", ch.get("display_name", "")), - "chat_type": ch.get("type", ""), - }) - + for ch in entries_list: + if isinstance(ch, dict): + chat_id = ch.get("id", ch.get("chat_id", "")) + channels.append({ + "target": f"{plat}:{chat_id}" if chat_id else plat, "platform": plat, + "name": ch.get("name", ch.get("display_name", "")), "chat_type": ch.get("type", ""), + }) return json.dumps({"count": len(channels), "channels": channels}, indent=2) - # -- permissions_list_open --------------------------------------------- - - @mcp.tool() - def permissions_list_open() -> str: + def permissions_list_open(self) -> str: """List pending approval requests observed during this bridge session. Returns exec and plugin approval requests that the bridge has seen since it started. Approvals are live-session only — older approvals from before the bridge connected are not included. """ - approvals = bridge.list_pending_approvals() - return json.dumps({ - "count": len(approvals), - "approvals": approvals, - }, indent=2) + approvals = self.bridge.list_pending_approvals() + return json.dumps({"count": len(approvals), "approvals": approvals}, indent=2) - # -- permissions_respond ----------------------------------------------- - - @mcp.tool() - def permissions_respond( - id: str, - decision: str, - ) -> str: + def permissions_respond(self, id: str, decision: str) -> str: """Respond to a pending approval request. Args: @@ -1016,17 +710,35 @@ def create_mcp_server(event_bridge: Optional[EventBridge] = None) -> "MCPServer" "error": f"Invalid decision: {decision}. " f"Must be allow-once, allow-always, or deny" }) + return json.dumps(self.bridge.respond_to_approval(id, decision), indent=2) - result = bridge.respond_to_approval(id, decision) - return json.dumps(result, indent=2) +# Registration order == list_tools order (wire format). +_TOOL_NAMES = ( + "conversations_list", "conversation_get", "messages_read", "attachments_fetch", + "events_poll", "events_wait", "messages_send", "channels_list", + "permissions_list_open", "permissions_respond", +) + + +def create_mcp_server(event_bridge: Optional[EventBridge] = None) -> "MCPServer": + """Create and return the Hermes MCP server with all tools registered.""" + if not _MCP_SERVER_AVAILABLE: + raise ImportError( + "MCP server requires the 'mcp' package. " + f"Install with: {sys.executable} -m pip install 'mcp'" + ) + mcp = MCPServer("hermes", instructions=( + "Hermes Agent messaging bridge. Use these tools to interact with " + "conversations across Telegram, Discord, Slack, WhatsApp, Signal, " + "Matrix, and other connected platforms." + )) + handlers = _ToolHandlers(event_bridge or EventBridge()) + for name in _TOOL_NAMES: + mcp.tool()(getattr(handlers, name)) return mcp -# --------------------------------------------------------------------------- -# Entry point -# --------------------------------------------------------------------------- - def run_mcp_server(verbose: bool = False) -> None: """Start the Hermes MCP server on stdio.""" if not _MCP_SERVER_AVAILABLE: @@ -1037,14 +749,10 @@ def run_mcp_server(verbose: bool = False) -> None: ) sys.exit(1) - if verbose: - logging.basicConfig(level=logging.DEBUG, stream=sys.stderr) - else: - logging.basicConfig(level=logging.WARNING, stream=sys.stderr) + logging.basicConfig(level=logging.DEBUG if verbose else logging.WARNING, stream=sys.stderr) bridge = EventBridge() bridge.start() - server = create_mcp_server(event_bridge=bridge) import asyncio diff --git a/model_tools.py b/model_tools.py index f0df359105..345a64305e 100644 --- a/model_tools.py +++ b/model_tools.py @@ -1,10 +1,9 @@ -""" -Model Tools Module +"""Thin orchestration layer over the tool registry. -Thin orchestration layer over the tool registry: importing this module runs tool -discovery (each tools/*.py self-registers via tools.registry.register()), then -exposes get_tool_definitions() (schemas sent to the model, toolset-filtered) and -handle_function_call() (dispatch with hooks/middleware) plus registry wrappers. +Importing runs tool discovery (each tools/*.py self-registers via +tools.registry.register()); exposes get_tool_definitions() (toolset-filtered +schemas sent to the model) and handle_function_call() (dispatch with +hooks/middleware) plus registry pass-throughs. """ import os @@ -12,7 +11,7 @@ import json import re import asyncio from contextlib import contextmanager -from dataclasses import dataclass +from dataclasses import asdict, dataclass from contextvars import ContextVar import logging import threading @@ -61,7 +60,6 @@ _WARNED_DISABLED_BUNDLES: set = set() def _is_delegated_child_context() -> bool: try: from agent.delegation_context import is_delegated_child_context - return is_delegated_child_context() except Exception: return False @@ -72,19 +70,15 @@ def _is_dispatcher_owned_worker() -> bool: (delegate_task child, or a cron job fired in-process from a worker).""" try: from agent.delegation_context import is_dispatcher_owned_worker_context - return is_dispatcher_owned_worker_context() except Exception: return True -# ============================================================================= -# Async Bridging (single source of truth -- used by registry.dispatch too) -# ============================================================================= +# --- Async bridging (single source of truth; registry.dispatch uses it too) --- # Loops are persistent (never asyncio.run per call): cached httpx/AsyncOpenAI -# clients stay bound to a live loop, so their GC cleanup can't hit -# "Event loop is closed". Main thread shares one loop; worker threads -# (parallel tool execution) each own a thread-local loop to avoid contention. +# clients stay bound to a live loop, so their GC cleanup can't hit "Event loop +# is closed". Main thread shares one loop; worker threads own thread-local loops. _tool_loop = None # persistent loop for the main (CLI) thread _tool_loop_lock = threading.Lock() @@ -116,14 +110,11 @@ def _run_async(coro): loop = asyncio.get_running_loop() except RuntimeError: loop = None - if loop and loop.is_running(): - # Already inside an event loop: run in a fresh thread whose loop we - # hold a reference to, so on timeout we can cancel the task inside it - # (ThreadPoolExecutor.cancel() is a no-op on a running worker and - # would leak the thread on every 300 s timeout). + # Inside a running loop: run in a fresh thread whose loop we keep a + # reference to, so on timeout we can cancel the task inside it + # (ThreadPoolExecutor.cancel() is a no-op on a running worker). import concurrent.futures - worker_loop: Optional[asyncio.AbstractEventLoop] = None loop_ready = threading.Event() @@ -135,15 +126,12 @@ def _run_async(coro): asyncio.set_event_loop(worker_loop) return worker_loop.run_until_complete(coro) finally: - try: - # Drain tasks still pending after an external cancel. + try: # drain tasks still pending after an external cancel pending = asyncio.all_tasks(worker_loop) for t in pending: t.cancel() if pending: - worker_loop.run_until_complete( - asyncio.gather(*pending, return_exceptions=True) - ) + worker_loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True)) except Exception: pass worker_loop.close() @@ -151,7 +139,6 @@ def _run_async(coro): pool = concurrent.futures.ThreadPoolExecutor(max_workers=1) # Carry profile + approval/sudo context so get_hermes_home() resolves correctly. from tools.thread_context import propagate_context_to_thread - future = pool.submit(propagate_context_to_thread(_run_in_worker)) try: return future.result(timeout=300) @@ -165,36 +152,27 @@ def _run_async(coro): pass # loop already closed raise finally: - # wait=False: never block the caller on a stuck coroutine. - pool.shutdown(wait=False) + pool.shutdown(wait=False) # never block the caller on a stuck coroutine if threading.current_thread() is not threading.main_thread(): return _get_worker_loop().run_until_complete(coro) return _get_tool_loop().run_until_complete(coro) -# ============================================================================= -# Tool Discovery (importing each module triggers its registry.register calls) -# ============================================================================= - +# --- Tool discovery (importing each tools/*.py triggers registry.register) --- discover_builtin_tools() # MCP discovery is deliberately NOT run here: it blocks up to 120 s and the -# gateway lazy-imports this module inside its event loop. Each entry point -# (gateway/run.py, cli.py, tui_gateway, acp_adapter) runs it at its own startup. - -# Plugin tool discovery (user/project/pip plugins) -try: +# gateway lazy-imports this module inside its event loop; each entry point +# (gateway/run.py, cli.py, tui_gateway, acp_adapter) runs it at startup. +try: # plugin tool discovery (user/project/pip plugins) from hermes_cli.plugins import discover_plugins discover_plugins() except Exception as e: logger.debug("Plugin discovery failed: %s", e) -# ============================================================================= -# Backward-compat constants (built once after discovery) -# ============================================================================= - +# Backward-compat constants (built once after discovery) TOOL_TO_TOOLSET_MAP: Dict[str, str] = registry.get_tool_to_toolset_map() TOOLSET_REQUIREMENTS: Dict[str, dict] = registry.get_toolset_requirements() @@ -203,10 +181,7 @@ TOOLSET_REQUIREMENTS: Dict[str, dict] = registry.get_toolset_requirements() _last_resolved_tool_names: List[str] = [] -# ============================================================================= -# Legacy toolset name mapping (old _tools-suffixed names -> tool name lists) -# ============================================================================= - +# Legacy toolset names (old _tools-suffixed names -> tool name lists) _LEGACY_TOOLSET_MAP = { "web_tools": ["web_search", "web_extract"], "terminal_tools": ["terminal"], @@ -214,10 +189,8 @@ _LEGACY_TOOLSET_MAP = { "image_tools": ["image_generate"], "skills_tools": ["skills_list", "skill_view", "skill_manage"], "browser_tools": [ - "browser_navigate", "browser_snapshot", "browser_click", - "browser_type", "browser_scroll", "browser_back", - "browser_press", "browser_get_images", - "browser_vision", "browser_console" + "browser_navigate", "browser_snapshot", "browser_click", "browser_type", "browser_scroll", + "browser_back", "browser_press", "browser_get_images", "browser_vision", "browser_console", ], "cronjob_tools": ["cronjob_manage"], "file_tools": ["read_file", "write_file", "patch", "search_files"], @@ -225,10 +198,7 @@ _LEGACY_TOOLSET_MAP = { } -# ============================================================================= -# get_tool_definitions (the main schema provider) -# ============================================================================= - +# --- get_tool_definitions (the main schema provider) -------------------------- # Memo for get_tool_definitions(), active only with quiet_mode=True (the # non-quiet path prints). Hot callers (gateway runner, AIAgent.__init__) hit it # every turn; a miss costs ~7 ms of registry walk + check_fn probing. The key @@ -236,9 +206,7 @@ _LEGACY_TOOLSET_MAP = { # invalidation is transparent; check_fn drift is handled by registry.py's 30 s TTL. _tool_defs_cache: Dict[tuple, List[Dict[str, Any]]] = {} _tool_defs_cache_lock = threading.Lock() - -# FIFO cap: a long-lived gateway sees many toolset/config fingerprints; 8 -# covers the warm working set of platform/toolset combos it actually serves. +# FIFO cap: 8 covers a long-lived gateway's warm set of platform/toolset combos. _TOOL_DEFS_CACHE_MAX = 8 @@ -256,29 +224,27 @@ def get_tool_definitions( ) -> List[Dict[str, Any]]: """Tool definitions for model API calls, filtered by toolset. - Args: - enabled_toolsets: Only include tools from these toolsets (None = all). - disabled_toolsets: Toolsets subtracted after enabling. - quiet_mode: Suppress status prints (and enable memoization). - skip_tool_search_assembly: Return the pre-assembly list (raw schemas for - every enabled tool). Only the tool_search bridge should use this so - it reads the real catalog rather than the collapsed one. + enabled_toolsets None = all; disabled_toolsets are subtracted after enabling. + quiet_mode suppresses status prints and enables memoization. + skip_tool_search_assembly returns raw schemas for every enabled tool — only + the tool_search bridge should use it (it reads the real, uncollapsed catalog). """ - if not quiet_mode: + def compute(): return _compute_tool_definitions(enabled_toolsets, disabled_toolsets, quiet_mode, skip_tool_search_assembly=skip_tool_search_assembly) + if not quiet_mode: + return compute() + cache_key = _tool_defs_cache_key(enabled_toolsets, disabled_toolsets, skip_tool_search_assembly) with _tool_defs_cache_lock: cached = _tool_defs_cache.get(cache_key) if cache_key is not None else None if cached is None: - result = _compute_tool_definitions(enabled_toolsets, disabled_toolsets, quiet_mode, - skip_tool_search_assembly=skip_tool_search_assembly) + result = compute() if cache_key is None: return list(result) with _tool_defs_cache_lock: - # Another thread may have filled this key meanwhile; reuse it. - cached = _tool_defs_cache.get(cache_key) + cached = _tool_defs_cache.get(cache_key) # another thread may have filled it meanwhile if cached is None: if len(_tool_defs_cache) >= _TOOL_DEFS_CACHE_MAX: _tool_defs_cache.pop(next(iter(_tool_defs_cache))) @@ -286,23 +252,19 @@ def get_tool_definitions( else: global _last_resolved_tool_names _last_resolved_tool_names = [t["function"]["name"] for t in cached] - # Always a shallow copy: run_agent appends memory/LCM schemas to its list, and - # a shared list would accumulate duplicate tool names across agent inits - # (rejected with HTTP 400 by DeepSeek/Kimi/MiMo). + # Always a shallow copy: run_agent appends memory/LCM schemas to its list; a + # shared list would accumulate duplicate names (HTTP 400 from DeepSeek/Kimi/MiMo). return list(cached) def _tool_defs_cache_key( - enabled_toolsets: Optional[List[str]], - disabled_toolsets: Optional[List[str]], - skip_tool_search_assembly: bool, + enabled_toolsets: Optional[List[str]], disabled_toolsets: Optional[List[str]], skip_tool_search_assembly: bool, ) -> Optional[tuple]: """Memo key for get_tool_definitions, or None when caching must be bypassed. - Covers every argument plus everything that changes the result without an - argument changing: registry generation, config.yaml mtime/size (dynamic - schemas: execute_code mode, discord allowlist), kanban context, profile - scope. check_fn results are TTL-cached inside registry.get_definitions. + Covers every argument plus everything that changes the result without one: + registry generation, config.yaml mtime/size (dynamic schemas), kanban + context, profile scope. check_fn results are TTL-cached in the registry. """ profile_scope = check_fn_cache_scope() if profile_scope == CHECK_FN_CACHE_BYPASS: @@ -317,13 +279,9 @@ def _tool_defs_cache_key( registry.current_scope_key(), frozenset(enabled_toolsets) if enabled_toolsets is not None else None, frozenset(disabled_toolsets) if disabled_toolsets else None, - registry._generation, - cfg_fp, - bool(os.environ.get("HERMES_KANBAN_TASK")), - bool(skip_tool_search_assembly), - _is_delegated_child_context(), - _is_dispatcher_owned_worker(), - profile_scope, + registry._generation, cfg_fp, + bool(os.environ.get("HERMES_KANBAN_TASK")), bool(skip_tool_search_assembly), + _is_delegated_child_context(), _is_dispatcher_owned_worker(), profile_scope, ) @@ -335,20 +293,16 @@ def _apply_toolset_selection(tools: set, names: List[str], quiet_mode: bool, *, if validate_toolset(name): label = f"{verb} toolset" if disable and (name.startswith("hermes-") or (get_toolset(name) or {}).get("posture")): - # Platform bundles and posture toolsets re-list the core tools - # without owning them; subtracting the whole set would empty - # the tool list. Remove only the non-core delta. + # Bundles/postures re-list the core tools without owning them; + # subtracting the whole set would empty the list — remove only the non-core delta. resolved = sorted(bundle_non_core_tools(name)) if not quiet_mode and name.startswith("hermes-") and name not in _WARNED_DISABLED_BUNDLES: _WARNED_DISABLED_BUNDLES.add(name) logger.info( - "agent.disabled_toolsets contains platform-bundle " - "name '%s'; core tools are preserved and only its " - "platform-specific tools (%s) are removed. Bundle " - "names usually belong in `toolsets:`, not " - "`disabled_toolsets` (#33924).", - name, - ", ".join(resolved) if resolved else "none", + "agent.disabled_toolsets contains platform-bundle name '%s'; core tools are " + "preserved and only its platform-specific tools (%s) are removed. Bundle names " + "usually belong in `toolsets:`, not `disabled_toolsets` (#33924).", + name, ", ".join(resolved) if resolved else "none", ) else: resolved = resolve_toolset(name) @@ -395,19 +349,21 @@ def _select_tool_names( # --- Dynamic schema rewrites ------------------------------------------------- -# Each rewriter receives the tool definition and the set of tool names that -# passed check_fn filtering, and returns the (possibly replaced) definition or -# None to drop the tool. Cross-references must use that set (not the requested -# names) so the model is never told about a tool that isn't in the list. +# Each rewriter gets (tool definition, set of tool names that passed check_fn) +# and returns the (possibly replaced) definition, or None to drop the tool. +# Cross-references must use that set so the model never hears of an absent tool. _BROWSER_NAVIGATE_WEB_HINT = " For simple information retrieval, prefer web_search or web_extract (faster, cheaper)." +def _fn_def(schema: Dict[str, Any]) -> Dict[str, Any]: + return {"type": "function", "function": schema} + + def _rewrite_execute_code(td: Dict[str, Any], available: set) -> Optional[Dict[str, Any]]: """List only sandbox tools that are actually available.""" from tools.code_execution_tool import SANDBOX_ALLOWED_TOOLS, build_execute_code_schema, _get_execution_mode - schema = build_execute_code_schema(SANDBOX_ALLOWED_TOOLS & available, mode=_get_execution_mode()) - return {"type": "function", "function": schema} + return _fn_def(build_execute_code_schema(SANDBOX_ALLOWED_TOOLS & available, mode=_get_execution_mode())) def _discord_rewriter(schema_fn_name: str): @@ -418,7 +374,7 @@ def _discord_rewriter(schema_fn_name: str): dynamic = getattr(_dt, schema_fn_name)() except Exception: dynamic = None - return None if dynamic is None else {"type": "function", "function": dynamic} + return None if dynamic is None else _fn_def(dynamic) return _rewrite @@ -427,14 +383,13 @@ def _rewrite_browser_navigate(td: Dict[str, Any], available: set) -> Optional[Di if {"web_search", "web_extract"} & available: return td desc = td["function"].get("description", "").replace(_BROWSER_NAVIGATE_WEB_HINT, "") - return {"type": "function", "function": {**td["function"], "description": desc}} + return _fn_def({**td["function"], "description": desc}) def _rewrite_browser_exec(td: Dict[str, Any], available: set) -> Optional[Dict[str, Any]]: - """browser_exec runs arbitrary host Python; a session without the terminal - surface must not regain host execution through the browser toolset. This is - a session-level gate rather than a check_fn: check_fns are TTL-cached - process-wide while one gateway serves sessions with different toolsets.""" + """browser_exec runs arbitrary host Python: a session without the terminal surface + must not regain host execution via the browser toolset. Session-level gate rather + than a check_fn because check_fns are TTL-cached process-wide across sessions.""" return td if "terminal" in available else None @@ -448,29 +403,20 @@ def _rewrite_delegate_task(td: Dict[str, Any], available: set) -> Optional[Dict[ return td fn = td.get("function", {}) desc = fn.get("description", "") - full_offvariant = "delegate_task, clarify, memory, or cronjob" - full_onvariant = "clarify, memory, or cronjob" - if full_offvariant in desc: - full, keep_self = full_offvariant, True - elif full_onvariant in desc: - full, keep_self = full_onvariant, False + for full, self_named in (("delegate_task, clarify, memory, or cronjob", True), ("clarify, memory, or cronjob", False)): + if full in desc: + break else: return td - names = (["delegate_task"] if keep_self else []) + blocked_present if blocked_present: - if len(names) == 1: - replacement = names[0] - elif len(names) == 2: - replacement = f"{names[0]} or {names[1]}" - else: - replacement = ", ".join(names[:-1]) + ", or " + names[-1] + names = (["delegate_task"] if self_named else []) + blocked_present + replacement = " or ".join(names) if len(names) <= 2 else ", ".join(names[:-1]) + ", or " + names[-1] desc = desc.replace(full, replacement) else: # Both variants end at the following newline. start = desc.find("- Children cannot call " + full) if start != -1: - end = desc.index("\n", start) + 1 - desc = desc[:start] + desc[end:] + desc = desc[:start] + desc[desc.index("\n", start) + 1:] return {**td, "function": {**fn, "description": desc}} @@ -498,6 +444,15 @@ def _apply_dynamic_schemas(tool_defs: List[Dict[str, Any]]) -> List[Dict[str, An return out +_TOOL_SEARCH_LISTING_FORMS = { + "full": "catalog listing embedded", + "names": "names-only listing embedded", + "mixed": "listing embedded (oversized servers summarized)", + "groups": "server summary embedded (search-only discovery)", + "none": "no listing (search-only)", +} + + def _compute_tool_definitions( enabled_toolsets: Optional[List[str]] = None, disabled_toolsets: Optional[List[str]] = None, @@ -508,17 +463,15 @@ def _compute_tool_definitions( tools_to_include = _select_tool_names(enabled_toolsets, disabled_toolsets, quiet_mode) # Registry returns only tools whose check_fn passes. filtered_tools = _apply_dynamic_schemas(registry.get_definitions(tools_to_include, quiet=quiet_mode)) + global _last_resolved_tool_names + _last_resolved_tool_names = [t["function"]["name"] for t in filtered_tools] if not quiet_mode: if filtered_tools: - tool_names = [t["function"]["name"] for t in filtered_tools] - print(f"🛠️ Final tool selection ({len(filtered_tools)} tools): {', '.join(tool_names)}") + print(f"🛠️ Final tool selection ({len(filtered_tools)} tools): {', '.join(_last_resolved_tool_names)}") else: print("🛠️ No tools selected (all filtered out or unavailable)") - global _last_resolved_tool_names - _last_resolved_tool_names = [t["function"]["name"] for t in filtered_tools] - # Normalize schema shapes llama.cpp's grammar converter rejects (bare # "type": "object", string-valued nodes from malformed MCP servers). try: @@ -535,21 +488,13 @@ def _compute_tool_definitions( from tools.tool_search import assemble_tool_defs, load_config as _load_ts_config ts_cfg = _load_ts_config() if not skip_tool_search_assembly and ts_cfg.enabled != "off": - assembly = assemble_tool_defs( - filtered_tools, - context_length=_resolve_active_context_length(), - config=ts_cfg, - ) + assembly = assemble_tool_defs(filtered_tools, context_length=_resolve_active_context_length(), config=ts_cfg) if assembly.activated and not quiet_mode: - _forms = {"full": "catalog listing embedded", - "names": "names-only listing embedded", - "mixed": "listing embedded (oversized servers summarized)", - "groups": "server summary embedded (search-only discovery)", - "none": "no listing (search-only)"} print( f"🔎 Tool Search (tier {assembly.tier}): {assembly.deferred_count} " f"MCP/plugin tools deferred (~{assembly.deferred_tokens} tokens) behind " - f"tool_search/describe/call — {_forms.get(assembly.listing_form, assembly.listing_form)}." + f"tool_search/describe/call — " + f"{_TOOL_SEARCH_LISTING_FORMS.get(assembly.listing_form, assembly.listing_form)}." ) filtered_tools = assembly.tool_defs except Exception as e: # pragma: no cover — never break tool loading @@ -573,11 +518,10 @@ def _active_model_config() -> Tuple[str, Dict[str, Any]]: def _resolve_active_context_length() -> int: """Active model's context length for the tool-search gate (0 if unresolvable). - Order: explicit `model.context_length` in config.yaml; provider-aware - resolution (Codex OAuth enforces a smaller window than the direct API for - the same slug); the on-disk metadata cache (a slightly stale window is fine - for picking a disclosure tier and avoids a ~200 ms /models probe per CLI - startup); then the full live resolver. + Order: explicit `model.context_length`; provider-aware resolution (Codex OAuth + enforces a smaller window than the direct API for the same slug); the on-disk + metadata cache (slightly stale is fine for picking a tier and avoids a ~200 ms + /models probe per CLI startup); then the full live resolver. """ try: model_id, model_cfg = _active_model_config() @@ -598,11 +542,8 @@ def _resolve_active_context_length() -> int: base_url = str(rt.get("base_url") or base_url or "").strip() api_key = str(rt.get("api_key") or "").strip() except Exception as rt_exc: - logger.debug( - "Runtime credential resolution failed for tool-search " - "context gate (provider=%s): %s — using config values only", - provider, rt_exc, - ) + logger.debug("Runtime credential resolution failed for tool-search " + "context gate (provider=%s): %s — using config values only", provider, rt_exc) if config_ctx is None and base_url: try: cached_ctx = get_cached_context_length(model_id, base_url) @@ -611,11 +552,8 @@ def _resolve_active_context_length() -> int: except Exception: pass return int(get_model_context_length( - model_id, - base_url=base_url, - api_key=api_key, - config_context_length=config_ctx, - provider=provider, + model_id, base_url=base_url, api_key=api_key, + config_context_length=config_ctx, provider=provider, ) or 0) except Exception as e: logger.debug("Could not resolve active context length: %s", e) @@ -626,36 +564,28 @@ def _resolve_active_context_length() -> int: # handle_function_call (the main dispatcher) # ============================================================================= -# Tools the agent loop (run_agent.py) intercepts because they need agent-level -# state. The registry still holds their schemas; dispatch returns a stub error. +# Intercepted by the agent loop (need agent-level state); dispatch returns a stub error. _AGENT_LOOP_TOOLS = {"todo_list", "memory", "session_search", "delegate_task"} -# Legacy tool-name aliases (2026-08 renames), accepted at every dispatch seam so -# old sessions and saved prompts keep working; schemas advertise only new names. +# Legacy tool-name aliases accepted at every dispatch seam (old sessions/saved +# prompts keep working); schemas advertise only new names. _LEGACY_TOOL_ALIASES = { - "todo": "todo_list", - "cronjob": "cronjob_manage", - "process": "process_manage", - "tour": "gui_tour", - "tip": "show_tip", + "todo": "todo_list", "cronjob": "cronjob_manage", "process": "process_manage", + "tour": "gui_tour", "tip": "show_tip", } _READ_SEARCH_TOOLS = {"read_file", "search_files"} -# ========================================================================= -# Tool error sanitization -# ========================================================================= -# Defense-in-depth: json.dumps already prevents framing escape, but the model -# still reads the text, so strip role tags / CDATA / code fences from exception -# messages and cap length. The cap is shared with tools/registry.py so text never -# passes two different caps with two different markers. -_TOOL_ERROR_ROLE_TAG_RE = re.compile( - r'', - re.IGNORECASE, +# --- Tool error sanitization -------------------------------------------------- +# Defense-in-depth: strip role tags / CDATA / code fences from exception text the +# model will read, and cap length (cap shared with tools/registry.py so text never +# passes two different caps with two different markers). +_TOOL_ERROR_STRIP_RES = ( + re.compile(r'', re.IGNORECASE), + re.compile(r'^\s*```(?:json|xml|html|markdown)?\s*', re.MULTILINE), + re.compile(r'\s*```\s*$', re.MULTILINE), + re.compile(r'', re.DOTALL), ) -_TOOL_ERROR_FENCE_OPEN_RE = re.compile(r'^\s*```(?:json|xml|html|markdown)?\s*', re.MULTILINE) -_TOOL_ERROR_FENCE_CLOSE_RE = re.compile(r'\s*```\s*$', re.MULTILINE) -_TOOL_ERROR_CDATA_RE = re.compile(r'', re.DOTALL) from tools.registry import _MAX_TOOL_ERROR_CHARS as _TOOL_ERROR_MAX_LEN @@ -663,10 +593,9 @@ def _sanitize_tool_error(error_msg: str) -> str: """Strip structural framing tokens from a tool error before the model sees it.""" if not error_msg: return "[TOOL_ERROR] " - sanitized = _TOOL_ERROR_ROLE_TAG_RE.sub("", error_msg) - sanitized = _TOOL_ERROR_FENCE_OPEN_RE.sub("", sanitized) - sanitized = _TOOL_ERROR_FENCE_CLOSE_RE.sub("", sanitized) - sanitized = _TOOL_ERROR_CDATA_RE.sub("", sanitized) + sanitized = error_msg + for pattern in _TOOL_ERROR_STRIP_RES: + sanitized = pattern.sub("", sanitized) if len(sanitized) > _TOOL_ERROR_MAX_LEN: sanitized = sanitized[:_TOOL_ERROR_MAX_LEN - 3] + "..." return f"[TOOL_ERROR] {sanitized}" @@ -682,20 +611,11 @@ class _CallIds: api_request_id: Optional[str] = None def hook_kwargs(self) -> Dict[str, str]: - """The same fields with None normalized to "" (hook/middleware wire contract).""" - return { - "task_id": self.task_id or "", - "session_id": self.session_id or "", - "tool_call_id": self.tool_call_id or "", - "turn_id": self.turn_id or "", - "api_request_id": self.api_request_id or "", - } + """Same fields with None -> "" (hook/middleware wire contract).""" + return {k: v or "" for k, v in asdict(self).items()} -def _tool_result_observer_fields( - tool_name: str, - result: Any, -) -> tuple[str, Optional[str], Optional[str]]: +def _tool_result_observer_fields(tool_name: str, result: Any) -> tuple[str, Optional[str], Optional[str]]: """Derive (status, error_type, error_message) from a tool result for observer hooks.""" try: parsed_result = json.loads(result) if isinstance(result, str) else result @@ -705,7 +625,6 @@ def _tool_result_observer_fields( pass try: from agent.display import _detect_tool_failure - failed, suffix = _detect_tool_failure(tool_name, result) if failed: return "error", "tool_error", suffix.strip().strip("[]") or None @@ -715,27 +634,15 @@ def _tool_result_observer_fields( def _emit_post_tool_call_hook( - *, - function_name: str, - function_args: Dict[str, Any], - result: Any, - task_id: Optional[str] = None, - session_id: Optional[str] = None, - tool_call_id: Optional[str] = None, - turn_id: Optional[str] = None, - api_request_id: Optional[str] = None, - duration_ms: int = 0, - status: Optional[str] = None, - error_type: Optional[str] = None, - error_message: Optional[str] = None, + *, function_name: str, function_args: Dict[str, Any], result: Any, + task_id: Optional[str] = None, session_id: Optional[str] = None, tool_call_id: Optional[str] = None, + turn_id: Optional[str] = None, api_request_id: Optional[str] = None, + duration_ms: int = 0, status: Optional[str] = None, + error_type: Optional[str] = None, error_message: Optional[str] = None, middleware_trace: Optional[List[Dict[str, Any]]] = None, ) -> None: - """Emit the ``post_tool_call`` observer hook. - - Gated on has_hook so the no-listener path costs one dict lookup; when - ``status`` is None the ok/error fields are derived from the result only - after that gate. - """ + """Emit the ``post_tool_call`` observer hook; gated on has_hook, and ok/error + fields are derived from the result only past that gate when status is None.""" if _post_tool_call_hook_suppressed.get(): return try: @@ -743,20 +650,11 @@ def _emit_post_tool_call_hook( if not has_hook("post_tool_call"): return if status is None: - status, error_type, error_message = _tool_result_observer_fields( - function_name, - result, - ) + status, error_type, error_message = _tool_result_observer_fields(function_name, result) invoke_hook( - "post_tool_call", - tool_name=function_name, - args=function_args, - result=result, + "post_tool_call", tool_name=function_name, args=function_args, result=result, **_CallIds(task_id, session_id, tool_call_id, turn_id, api_request_id).hook_kwargs(), - duration_ms=duration_ms, - status=status, - error_type=error_type, - error_message=error_message, + duration_ms=duration_ms, status=status, error_type=error_type, error_message=error_message, middleware_trace=list(middleware_trace or []), ) except Exception as _hook_err: @@ -764,17 +662,14 @@ def _emit_post_tool_call_hook( def _dispatch_bridge_tool( - function_name: str, - function_args: Dict[str, Any], - enabled_toolsets: Optional[List[str]], - disabled_toolsets: Optional[List[str]], + function_name: str, function_args: Dict[str, Any], + enabled_toolsets: Optional[List[str]], disabled_toolsets: Optional[List[str]], ): """Handle a Tool Search bridge call (tool_search / tool_describe / tool_call). - Returns None when *function_name* is not a bridge tool. Otherwise returns - ``(result, None)`` for a finished catalog read or error, or - ``(None, (underlying_name, underlying_args))`` when a validated tool_call - should be re-dispatched as the real tool. + None when *function_name* is not a bridge tool; ``(result, None)`` for a + finished catalog read or error; ``(None, (name, args))`` when a validated + tool_call should be re-dispatched as the real tool. """ try: from tools import tool_search as ts @@ -782,13 +677,11 @@ def _dispatch_bridge_tool( return None if not ts.is_bridge_tool(function_name): return None - # Read the un-collapsed catalog, scoped to the session's toolsets so a - # restricted session (subagent, kanban worker) cannot see or invoke the - # whole process registry through the bridge. + # Un-collapsed catalog scoped to the session's toolsets, so a restricted + # session (subagent, kanban worker) can't reach the whole registry via the bridge. try: current_defs = get_tool_definitions( - enabled_toolsets=enabled_toolsets, - disabled_toolsets=disabled_toolsets, + enabled_toolsets=enabled_toolsets, disabled_toolsets=disabled_toolsets, quiet_mode=True, skip_tool_search_assembly=True, ) or [] except Exception: @@ -816,21 +709,30 @@ def _dispatch_bridge_tool( return None, (underlying_name, underlying_args) +def _apply_request_middleware( + function_name: str, function_args: Dict[str, Any], ids: _CallIds, trace: List[Dict[str, Any]], +) -> Tuple[Dict[str, Any], Dict[str, Any], List[Dict[str, Any]]]: + """tool_request middleware: returns (args, original_args, trace); fail-open.""" + try: + from hermes_cli.middleware import apply_tool_request_middleware + mw = apply_tool_request_middleware(function_name, function_args, **ids.hook_kwargs()) + return mw.payload, mw.original_payload, mw.trace + except Exception as _mw_err: + logger.debug("tool_request middleware error: %s", _mw_err) + return function_args, dict(function_args), trace + + def _pre_dispatch_guards( - function_name: str, - function_args: Dict[str, Any], - skip_pre_tool_call_hook: bool, - ids: _CallIds, - middleware_trace: List[Dict[str, Any]], + function_name: str, function_args: Dict[str, Any], skip_pre_tool_call_hook: bool, + ids: _CallIds, middleware_trace: List[Dict[str, Any]], ) -> Tuple[Dict[str, Any], Optional[Tuple[Any, str, Optional[str]]]]: """Plugin pre_tool_call hook, then ACP edit approval. - Returns ``(args, None)`` to proceed (args possibly modified by a plugin), or - ``(args, (result, error_type, error_message))`` when the call is blocked. + ``(args, None)`` to proceed (args possibly plugin-modified), or + ``(args, (result, error_type, error_message))`` when blocked. """ - # pre_tool_call fires exactly once per execution: _dispatch_pre_tool_call_hooks - # returns the block message (block/approve) and modified args (modify) from a - # single invoke_hook pass. skip=True means the caller already fired it. + # pre_tool_call fires exactly once per execution: one invoke_hook pass yields + # both the block message and modified args. skip=True: caller already fired it. if not skip_pre_tool_call_hook: block_message: Optional[str] = None try: @@ -849,101 +751,78 @@ def _pre_dispatch_guards( # via ContextVar only for ACP sessions, so CLI/gateway paths are unaffected. try: from acp_adapter.edit_approval import maybe_require_edit_approval - edit_block_message = maybe_require_edit_approval(function_name, function_args) if edit_block_message is not None: return function_args, (edit_block_message, "edit_approval_denied", None) except Exception as _edit_approval_err: logger.debug("ACP edit approval guard error: %s", _edit_approval_err) if function_name in {"write_file", "patch"}: - return function_args, ( - tool_error("Edit approval denied: approval guard failed"), "edit_approval_error", None, - ) + return function_args, (tool_error("Edit approval denied: approval guard failed"), "edit_approval_error", None) return function_args, None +@contextmanager +def _approval_observability(ids: _CallIds): + """Bind the approval observability context (turn/tool_call/session ids) for the block.""" + try: + from tools.approval import reset_current_observability_context, set_current_observability_context + tokens = set_current_observability_context( + turn_id=ids.turn_id or "", tool_call_id=ids.tool_call_id or "", session_id=ids.session_id or "", + ) + except Exception: + yield + return + try: + yield + finally: + try: + reset_current_observability_context(tokens) + except Exception: + pass + + def _execute_tool( - function_name: str, - function_args: Dict[str, Any], - original_args: Dict[str, Any], - ids: _CallIds, - *, - user_task: Optional[str], - enabled_tools: Optional[List[str]], - skip_tool_execution_middleware: bool, + function_name: str, function_args: Dict[str, Any], original_args: Dict[str, Any], ids: _CallIds, + *, user_task: Optional[str], enabled_tools: Optional[List[str]], skip_tool_execution_middleware: bool, ) -> Any: """Run the registry handler (through tool-execution middleware unless skipped) with the approval observability context bound for the duration.""" - approval_tokens = None - reset_obs = None - try: - from tools.approval import ( - reset_current_observability_context as reset_obs, - set_current_observability_context, - ) - approval_tokens = set_current_observability_context( - turn_id=ids.turn_id or "", - tool_call_id=ids.tool_call_id or "", - session_id=ids.session_id or "", - ) - except Exception: - reset_obs = None - try: - dispatch_kwargs: Dict[str, Any] = {"task_id": ids.task_id, "session_id": ids.session_id} - if function_name == "execute_code": - # Prefer the caller's list so subagents can't overwrite the - # parent's tool set via the process-global. - dispatch_kwargs["enabled_tools"] = ( - enabled_tools if enabled_tools is not None else _last_resolved_tool_names - ) - else: - dispatch_kwargs["user_task"] = user_task + dispatch_kwargs: Dict[str, Any] = {"task_id": ids.task_id, "session_id": ids.session_id} + if function_name == "execute_code": + # Prefer the caller's list so subagents can't overwrite the parent's + # tool set via the process-global. + dispatch_kwargs["enabled_tools"] = enabled_tools if enabled_tools is not None else _last_resolved_tool_names + else: + dispatch_kwargs["user_task"] = user_task - def _dispatch(next_args: Dict[str, Any]) -> Any: - return registry.dispatch(function_name, next_args, **dispatch_kwargs) + def _dispatch(next_args: Dict[str, Any]) -> Any: + return registry.dispatch(function_name, next_args, **dispatch_kwargs) + with _approval_observability(ids): if skip_tool_execution_middleware: return _dispatch(function_args) from hermes_cli.middleware import run_tool_execution_middleware - return run_tool_execution_middleware( function_name, function_args, _dispatch, original_args=original_args, **ids.hook_kwargs(), ) - finally: - if approval_tokens is not None and reset_obs is not None: - try: - reset_obs(approval_tokens) - except Exception: - pass def _apply_transform_tool_result_hook( - function_name: str, - function_args: Dict[str, Any], - result: Any, - duration_ms: int, - ids: _CallIds, + function_name: str, function_args: Dict[str, Any], result: Any, duration_ms: int, ids: _CallIds, ) -> Any: """transform_tool_result: plugins may replace the final result string. - Runs after post_tool_call (observational) and before the result enters - context. Fail-open; first valid string return wins; non-strings ignored. - Gated on has_hook so the no-listener path skips result-field derivation. + Runs after post_tool_call and before the result enters context. Fail-open; + first string return wins. Gated on has_hook so the no-listener path is cheap. """ try: from hermes_cli.lifecycle import has_hook, invoke_hook if has_hook("transform_tool_result"): status, error_type, error_message = _tool_result_observer_fields(function_name, result) hook_results = invoke_hook( - "transform_tool_result", - tool_name=function_name, - args=function_args, - result=result, - **ids.hook_kwargs(), - duration_ms=duration_ms, - status=status, - error_type=error_type, - error_message=error_message, + "transform_tool_result", tool_name=function_name, args=function_args, result=result, + **ids.hook_kwargs(), duration_ms=duration_ms, + status=status, error_type=error_type, error_message=error_message, ) for hook_result in hook_results: if isinstance(hook_result, str): @@ -953,6 +832,10 @@ def _apply_transform_tool_result_hook( return result +def _elapsed_ms(start: float) -> int: + return int((time.monotonic() - start) * 1000) + + def handle_function_call( function_name: str, function_args: Dict[str, Any], @@ -972,32 +855,24 @@ def handle_function_call( ) -> str: """Route a tool call through hooks/middleware to the registry; returns a JSON string. - Args: - task_id: Terminal/browser session isolation key. - user_task: The user's original task (browser_snapshot context). - enabled_tools: Session tool names; execute_code uses them to pick sandbox - tools (falls back to the process-global ``_last_resolved_tool_names``). - skip_pre_tool_call_hook: Caller already fired pre_tool_call (single-fire contract). - enabled_toolsets / disabled_toolsets: The session's toolset selection, - used to scope the Tool Search bridge catalog so tool_search / - tool_describe / tool_call only see tools this session was granted. - None = no restriction, matching get_tool_definitions semantics. + task_id isolates terminal/browser sessions; user_task feeds browser_snapshot. + enabled_tools picks execute_code's sandbox tools (default: the process-global + ``_last_resolved_tool_names``). skip_pre_tool_call_hook: caller already fired + it (single-fire contract). enabled/disabled_toolsets scope the Tool Search + bridge catalog to this session's grant (None = unrestricted). """ function_args = coerce_tool_args(function_name, function_args) if not isinstance(function_args, dict): function_args = {} - _tool_middleware_trace = list(tool_request_middleware_trace or []) + trace = list(tool_request_middleware_trace or []) function_name = _LEGACY_TOOL_ALIASES.get(function_name, function_name) ids = _CallIds(task_id, session_id, tool_call_id, turn_id, api_request_id) - _dispatch_start = time.monotonic() + start = time.monotonic() def _emit(result: Any, **extra: Any) -> Any: """Emit post_tool_call with this call's identity fields; returns *result*.""" - _emit_post_tool_call_hook( - function_name=function_name, function_args=function_args, result=result, - task_id=task_id, session_id=session_id, tool_call_id=tool_call_id, turn_id=turn_id, - api_request_id=api_request_id, middleware_trace=list(_tool_middleware_trace), **extra, - ) + _emit_post_tool_call_hook(function_name=function_name, function_args=function_args, result=result, + **asdict(ids), middleware_trace=list(trace), **extra) return result # Tool Search bridge: tool_search / tool_describe are catalog reads handled @@ -1007,45 +882,26 @@ def handle_function_call( if bridged is not None: result, underlying = bridged if underlying is None: - return _emit(result, duration_ms=int((time.monotonic() - _dispatch_start) * 1000)) - underlying_name, underlying_args = underlying + return _emit(result, duration_ms=_elapsed_ms(start)) return handle_function_call( - function_name=underlying_name, - function_args=underlying_args, - task_id=task_id, - tool_call_id=tool_call_id, - session_id=session_id, - turn_id=turn_id, - api_request_id=api_request_id, - user_task=user_task, - enabled_tools=enabled_tools, - skip_pre_tool_call_hook=skip_pre_tool_call_hook, + *underlying, task_id=task_id, tool_call_id=tool_call_id, session_id=session_id, + turn_id=turn_id, api_request_id=api_request_id, user_task=user_task, + enabled_tools=enabled_tools, skip_pre_tool_call_hook=skip_pre_tool_call_hook, skip_tool_request_middleware=skip_tool_request_middleware, skip_tool_execution_middleware=skip_tool_execution_middleware, - tool_request_middleware_trace=list(_tool_middleware_trace), - enabled_toolsets=enabled_toolsets, - disabled_toolsets=disabled_toolsets, + tool_request_middleware_trace=list(trace), + enabled_toolsets=enabled_toolsets, disabled_toolsets=disabled_toolsets, ) - _tool_original_args = dict(function_args) + original_args = dict(function_args) if not skip_tool_request_middleware: - try: - from hermes_cli.middleware import apply_tool_request_middleware - - _tool_request_mw = apply_tool_request_middleware(function_name, function_args, **ids.hook_kwargs()) - function_args = _tool_request_mw.payload - _tool_original_args = _tool_request_mw.original_payload - _tool_middleware_trace = _tool_request_mw.trace - except Exception as _mw_err: - logger.debug("tool_request middleware error: %s", _mw_err) + function_args, original_args, trace = _apply_request_middleware(function_name, function_args, ids, trace) try: if function_name in _AGENT_LOOP_TOOLS: return tool_error(f"{function_name} must be handled by the agent loop") - function_args, blocked = _pre_dispatch_guards( - function_name, function_args, skip_pre_tool_call_hook, ids, _tool_middleware_trace, - ) + function_args, blocked = _pre_dispatch_guards(function_name, function_args, skip_pre_tool_call_hook, ids, trace) if blocked is not None: result, error_type, error_message = blocked return _emit(result, status="blocked", error_type=error_type, error_message=error_message) @@ -1059,54 +915,45 @@ def handle_function_call( pass # file_tools may not be loaded yet # duration_ms (monotonic) is exposed to post_tool_call / transform_tool_result. - _dispatch_start = time.monotonic() + start = time.monotonic() result = _execute_tool( - function_name, function_args, _tool_original_args, ids, + function_name, function_args, original_args, ids, user_task=user_task, enabled_tools=enabled_tools, skip_tool_execution_middleware=skip_tool_execution_middleware, ) - duration_ms = int((time.monotonic() - _dispatch_start) * 1000) - + duration_ms = _elapsed_ms(start) _emit(result, duration_ms=duration_ms) - return _apply_transform_tool_result_hook(function_name, function_args, result, duration_ms, ids) except Exception as e: error_msg = f"Error executing {function_name}: {str(e)}" logger.exception(error_msg) - return _emit( - tool_error(_sanitize_tool_error(error_msg)), - duration_ms=int((time.monotonic() - _dispatch_start) * 1000), - status="error", - error_type=type(e).__name__, - error_message=str(e), - ) + return _emit(tool_error(_sanitize_tool_error(error_msg)), duration_ms=_elapsed_ms(start), + status="error", error_type=type(e).__name__, error_message=str(e)) # ============================================================================= -# Backward-compat wrapper functions +# Backward-compat wrapper functions (registry pass-throughs) # ============================================================================= def get_all_tool_names() -> List[str]: - """Return all registered tool names.""" return registry.get_all_tool_names() def get_toolset_for_tool(tool_name: str) -> Optional[str]: - """Return the toolset a tool belongs to.""" return registry.get_toolset_for_tool(tool_name) def get_available_toolsets() -> Dict[str, dict]: - """Return toolset availability info for UI display.""" + """Toolset availability info for UI display.""" return registry.get_available_toolsets() def check_toolset_requirements() -> Dict[str, bool]: - """Return {toolset: available_bool} for every registered toolset.""" + """{toolset: available_bool} for every registered toolset.""" return registry.check_toolset_requirements() def check_tool_availability(quiet: bool = False) -> Tuple[List[str], List[dict]]: - """Return (available_toolsets, unavailable_info).""" + """(available_toolsets, unavailable_info).""" return registry.check_tool_availability(quiet=quiet) diff --git a/toolset_distributions.py b/toolset_distributions.py index fa643d312e..c20372524d 100644 --- a/toolset_distributions.py +++ b/toolset_distributions.py @@ -1,22 +1,8 @@ #!/usr/bin/env python3 -""" -Toolset Distributions Module +"""Toolset distributions for batch data-generation runs. -This module defines distributions of toolsets for data generation runs. -Each distribution specifies which toolsets should be used and their probability -of being selected for any given prompt during the batch processing. - -A distribution is a dictionary mapping toolset names to their selection probability (%). -Probabilities should sum to 100, but the system will normalize if they don't. - -Usage: - from toolset_distributions import get_distribution, list_distributions - - # Get a specific distribution - dist = get_distribution("image_gen") - - # List all available distributions - all_dists = list_distributions() +A distribution maps toolset names to the % chance each is enabled for a prompt +(sampled independently, so several toolsets can be active at once). """ from typing import Dict, List, Optional @@ -24,335 +10,101 @@ import random from toolsets import validate_toolset -# Distribution definitions -# Each key is a distribution name, and the value is a dict of toolset_name: probability_percentage +def _dist(description: str, **toolsets: int) -> Dict[str, object]: + return {"description": description, "toolsets": toolsets} + + DISTRIBUTIONS = { - # Default: All tools available 100% of the time - "default": { - "description": "All available tools, all the time", - "toolsets": { - "web": 100, - "vision": 100, - "image_gen": 100, - "terminal": 100, - "file": 100, - "browser": 100 - } - }, - - # Image generation focused distribution - "image_gen": { - "description": "Heavy focus on image generation with vision and web support", - "toolsets": { - "image_gen": 90, # 80% chance of image generation tools - "vision": 90, # 60% chance of vision tools - "web": 55, # 40% chance of web tools - "terminal": 45 - } - }, - - # Research-focused distribution - "research": { - "description": "Web research with vision analysis and reasoning", - "toolsets": { - "web": 90, # 90% chance of web tools - "browser": 70, # 70% chance of browser tools for deep research - "vision": 50, # 50% chance of vision tools - "terminal": 10 # 10% chance of terminal tools - } - }, - - # Scientific problem solving focused distribution - "science": { - "description": "Scientific research with web, terminal, file, and browser capabilities", - "toolsets": { - "web": 94, # 94% chance of web tools - "terminal": 94, # 94% chance of terminal tools - "file": 94, # 94% chance of file tools - "vision": 65, # 65% chance of vision tools - "browser": 50, # 50% chance of browser for accessing papers/databases - "image_gen": 15 # 15% chance of image generation tools - } - }, - - # Development-focused distribution - "development": { - "description": "Terminal, file tools, and reasoning with occasional web lookup", - "toolsets": { - "terminal": 80, # 80% chance of terminal tools - "file": 80, # 80% chance of file tools (read, write, patch, search) - "web": 30, # 30% chance of web tools - "vision": 10 # 10% chance of vision tools - } - }, - - # Safe mode (no terminal) - "safe": { - "description": "All tools except terminal for safety", - "toolsets": { - "web": 80, - "browser": 70, # Browser is safe (no local filesystem access) - "vision": 60, - "image_gen": 60 - } - }, - - # Balanced distribution - "balanced": { - "description": "Equal probability of all toolsets", - "toolsets": { - "web": 50, - "vision": 50, - "image_gen": 50, - "terminal": 50, - "file": 50, - "browser": 50 - } - }, - - # Minimal (web only) - "minimal": { - "description": "Only web tools for basic research", - "toolsets": { - "web": 100 - } - }, - - # Terminal only - "terminal_only": { - "description": "Terminal and file tools for code execution tasks", - "toolsets": { - "terminal": 100, - "file": 100 - } - }, - - # Terminal + web (common for coding tasks that need docs) - "terminal_web": { - "description": "Terminal and file tools with web search for documentation lookup", - "toolsets": { - "terminal": 100, - "file": 100, - "web": 100 - } - }, - - # Creative (vision + image generation) - "creative": { - "description": "Image generation and vision analysis focus", - "toolsets": { - "image_gen": 90, - "vision": 90, - "web": 30 - } - }, - - # Reasoning heavy - "reasoning": { - "description": "Heavy research/reasoning distribution with minimal other tools", - "toolsets": { - "web": 90, - "file": 60, - "terminal": 20 - } - }, - - - # Browser-based web interaction - "browser_use": { - "description": "Full browser-based web interaction with search, vision, and page control", - "toolsets": { - "browser": 100, # All browser tools always available - "web": 80, # Web search for finding URLs and quick lookups - "vision": 70 # Vision analysis for images found on pages - } - }, - - # Browser only (no other tools) - "browser_only": { - "description": "Only browser automation tools for pure web interaction tasks", - "toolsets": { - "browser": 100 - } - }, - - # Browser-focused tasks distribution (for browser-use-tasks.jsonl) - "browser_tasks": { - "description": "Browser-focused distribution (browser toolset includes web_search for finding URLs since Google blocks direct browser searches)", - "toolsets": { - "browser": 97, # 97% - browser tools (includes web_search) almost always available - "vision": 12, # 12% - vision analysis occasionally - "terminal": 15 # 15% - terminal occasionally for local operations - } - }, - - # Terminal-focused tasks distribution (for nous-terminal-tasks.jsonl) - "terminal_tasks": { - "description": "Terminal-focused distribution with high terminal/file availability, occasional other tools", - "toolsets": { - "terminal": 97, # 97% - terminal almost always available - "file": 97, # 97% - file tools almost always available - "web": 97, # 15% - web search/scrape for documentation - "browser": 75, # 10% - browser occasionally for web interaction - "vision": 50, # 8% - vision analysis rarely - "image_gen": 10 # 3% - image generation very rarely - } - }, - - # Mixed browser+terminal tasks distribution (for mixed-browser-terminal-tasks.jsonl) - "mixed_tasks": { - "description": "Mixed distribution with high browser, terminal, and file availability for complex tasks", - "toolsets": { - "browser": 92, # 92% - browser tools highly available - "terminal": 92, # 92% - terminal highly available - "file": 92, # 92% - file tools highly available - "web": 35, # 35% - web search/scrape fairly common - "vision": 15, # 15% - vision analysis occasionally - "image_gen": 15 # 15% - image generation occasionally - } - } + "default": _dist("All available tools, all the time", + web=100, vision=100, image_gen=100, terminal=100, file=100, browser=100), + "image_gen": _dist("Heavy focus on image generation with vision and web support", + image_gen=90, vision=90, web=55, terminal=45), + "research": _dist("Web research with vision analysis and reasoning", + web=90, browser=70, vision=50, terminal=10), + "science": _dist("Scientific research with web, terminal, file, and browser capabilities", + web=94, terminal=94, file=94, vision=65, browser=50, image_gen=15), + "development": _dist("Terminal, file tools, and reasoning with occasional web lookup", + terminal=80, file=80, web=30, vision=10), + "safe": _dist("All tools except terminal for safety", + web=80, browser=70, vision=60, image_gen=60), + "balanced": _dist("Equal probability of all toolsets", + web=50, vision=50, image_gen=50, terminal=50, file=50, browser=50), + "minimal": _dist("Only web tools for basic research", web=100), + "terminal_only": _dist("Terminal and file tools for code execution tasks", terminal=100, file=100), + "terminal_web": _dist("Terminal and file tools with web search for documentation lookup", + terminal=100, file=100, web=100), + "creative": _dist("Image generation and vision analysis focus", image_gen=90, vision=90, web=30), + "reasoning": _dist("Heavy research/reasoning distribution with minimal other tools", + web=90, file=60, terminal=20), + "browser_use": _dist("Full browser-based web interaction with search, vision, and page control", + browser=100, web=80, vision=70), + "browser_only": _dist("Only browser automation tools for pure web interaction tasks", browser=100), + # browser-use-tasks.jsonl: the browser toolset includes web_search since Google blocks direct browser searches + "browser_tasks": _dist( + "Browser-focused distribution (browser toolset includes web_search for finding URLs since Google blocks direct browser searches)", + browser=97, vision=12, terminal=15, + ), + # nous-terminal-tasks.jsonl + "terminal_tasks": _dist( + "Terminal-focused distribution with high terminal/file availability, occasional other tools", + terminal=97, file=97, web=97, browser=75, vision=50, image_gen=10, + ), + # mixed-browser-terminal-tasks.jsonl + "mixed_tasks": _dist( + "Mixed distribution with high browser, terminal, and file availability for complex tasks", + browser=92, terminal=92, file=92, web=35, vision=15, image_gen=15, + ), } def get_distribution(name: str) -> Optional[Dict[str, any]]: - """ - Get a toolset distribution by name. - - Args: - name (str): Name of the distribution - - Returns: - Dict: Distribution definition with description and toolsets - None: If distribution not found - """ + """Distribution definition (description + toolsets), or None if unknown.""" return DISTRIBUTIONS.get(name) def list_distributions() -> Dict[str, Dict]: - """ - List all available distributions. - - Returns: - Dict: All distribution definitions - """ return DISTRIBUTIONS.copy() def sample_toolsets_from_distribution(distribution_name: str) -> List[str]: - """ - Sample toolsets based on a distribution's probabilities. - - Each toolset in the distribution has a % chance of being included. - This allows multiple toolsets to be active simultaneously. - - Args: - distribution_name (str): Name of the distribution to sample from - - Returns: - List[str]: List of sampled toolset names - - Raises: - ValueError: If distribution name is not found + """Sample toolset names, each included independently with its % probability. + + Falls back to the highest-probability toolset when nothing was rolled. + Raises ValueError for an unknown distribution. """ dist = get_distribution(distribution_name) if not dist: raise ValueError(f"Unknown distribution: {distribution_name}") - - # Sample each toolset independently based on its probability + selected_toolsets = [] - for toolset_name, probability in dist["toolsets"].items(): - # Validate toolset exists if not validate_toolset(toolset_name): print(f"⚠️ Warning: Toolset '{toolset_name}' in distribution '{distribution_name}' is not valid") continue - - # Roll the dice - if random value is less than probability, include this toolset if random.random() * 100 < probability: selected_toolsets.append(toolset_name) - - # If no toolsets were selected (can happen with low probabilities), - # ensure at least one toolset is selected by picking the highest probability one + if not selected_toolsets and dist["toolsets"]: - # Find toolset with highest probability highest_prob_toolset = max(dist["toolsets"].items(), key=lambda x: x[1])[0] if validate_toolset(highest_prob_toolset): selected_toolsets.append(highest_prob_toolset) - + return selected_toolsets def validate_distribution(distribution_name: str) -> bool: - """ - Check if a distribution name is valid. - - Args: - distribution_name (str): Distribution name to validate - - Returns: - bool: True if valid, False otherwise - """ return distribution_name in DISTRIBUTIONS def print_distribution_info(distribution_name: str) -> None: - """ - Print detailed information about a distribution. - - Args: - distribution_name (str): Distribution name - """ + """Print a distribution's description and toolset probabilities (highest first).""" dist = get_distribution(distribution_name) if not dist: print(f"❌ Unknown distribution: {distribution_name}") return - + print(f"\n📊 Distribution: {distribution_name}") print(f" Description: {dist['description']}") print(" Toolsets:") for toolset, prob in sorted(dist["toolsets"].items(), key=lambda x: x[1], reverse=True): print(f" • {toolset:15} : {prob:3}% chance") - - -if __name__ == "__main__": - """ - Demo and testing of the distributions system - """ - print("📊 Toolset Distributions Demo") - print("=" * 60) - - # List all distributions - print("\n📋 Available Distributions:") - print("-" * 40) - for name, dist in list_distributions().items(): - print(f"\n {name}:") - print(f" {dist['description']}") - toolset_list = ", ".join([f"{ts}({p}%)" for ts, p in dist["toolsets"].items()]) - print(f" Toolsets: {toolset_list}") - - # Demo sampling - print("\n\n🎲 Sampling Examples:") - print("-" * 40) - - test_distributions = ["image_gen", "research", "balanced", "default"] - - for dist_name in test_distributions: - print(f"\n{dist_name}:") - # Sample 5 times to show variability - samples = [] - for _ in range(5): - sampled = sample_toolsets_from_distribution(dist_name) - samples.append(sorted(sampled)) - - print(f" Sample 1: {samples[0]}") - print(f" Sample 2: {samples[1]}") - print(f" Sample 3: {samples[2]}") - print(f" Sample 4: {samples[3]}") - print(f" Sample 5: {samples[4]}") - - # Show detailed info - print("\n\n📊 Detailed Distribution Info:") - print("-" * 40) - print_distribution_info("image_gen") - print_distribution_info("research") - diff --git a/toolsets.py b/toolsets.py index eda4f8f5fc..a36469db95 100644 --- a/toolsets.py +++ b/toolsets.py @@ -3,45 +3,29 @@ from typing import Dict, List, Any, Set, Optional, Tuple -# Shared tool list for CLI and all messaging platform toolsets. -# Edit this once to update all platforms simultaneously. +# Shared tool list for CLI and all messaging platform toolsets (edit once, all +# platforms follow). Desktop GUI affordances are deliberately NOT here: they live +# in `desktop_ui`/`project`, enabled per desktop-sourced session by the GUI gateway +# (tui_gateway/server.py::_load_enabled_toolsets). HA, kanban and computer_use +# entries are further gated by their tools' check_fns. _HERMES_CORE_TOOLS = [ - # Web "web_search", "web_extract", - # Terminal + process management "terminal", "process_manage", - # Desktop GUI affordances (read_terminal, open_preview, project_*) are NOT - # here: they live in `desktop_ui`/`project`, enabled only by the GUI gateway - # per desktop-sourced session (tui_gateway/server.py::_load_enabled_toolsets). - # File manipulation "read_file", "write_file", "patch", "search_files", - # Vision + image generation "vision_analyze", "image_generate", - # Skills "skills_list", "skill_view", "skill_manage", - # Browser automation "browser_navigate", "browser_snapshot", "browser_click", "browser_type", "browser_scroll", "browser_back", "browser_press", "browser_get_images", "browser_vision", "browser_console", "browser_cdp", "browser_dialog", - # replaces other tools when browser.backend is "browser-use" - "browser_exec", - # Text-to-speech + "browser_exec", # replaces the other browser tools when browser.backend is "browser-use" "text_to_speech", - # Planning & memory "todo_list", "memory", - # Session history search "session_search", - # Clarifying questions "clarify", - # Code execution + delegation "execute_code", "delegate_task", - # Cronjob management "cronjob_manage", - # Home Assistant smart home control (gated on HASS_TOKEN via check_fn) "ha_list_entities", "ha_get_state", "ha_list_services", "ha_call_service", - # Kanban coordination — check_fn in tools/kanban_tools.py admits these only - # for kanban workers (HERMES_KANBAN_TASK) or profiles enabling `kanban`. "kanban_show", "kanban_list", "kanban_complete", "kanban_block", "kanban_request_review", "kanban_request_changes", @@ -49,17 +33,17 @@ _HERMES_CORE_TOOLS = [ "kanban_comment", "kanban_create", "kanban_link", "kanban_unblock", "kanban_attach", "kanban_attach_url", "kanban_attachments", - # Computer use (macOS, gated on cua-driver being installed via check_fn) "computer_use", ] # Webhook payloads are untrusted third-party content: no file/system execution. -_HERMES_WEBHOOK_SAFE_TOOLS = [ - "web_search", - "web_extract", - "vision_analyze", - "clarify", +_HERMES_WEBHOOK_SAFE_TOOLS = ["web_search", "web_extract", "vision_analyze", "clarify"] +_HA_TOOLS = ["ha_list_entities", "ha_get_state", "ha_list_services", "ha_call_service"] +_FEISHU_TOOLS = [ + "feishu_doc_read", "feishu_drive_list_comments", "feishu_drive_list_comment_replies", + "feishu_drive_reply_comment", "feishu_drive_add_comment", ] +_YUANBAO_TOOLS = ["yb_query_group_info", "yb_query_group_members", "yb_send_dm", "yb_search_sticker", "yb_send_sticker"] def _ts(description, tools=(), includes=(), **extra): @@ -79,11 +63,7 @@ def _core_without(*excluded, kanban=True): # Coding posture: everything you reach for while pairing on code; drops messaging, # tts, image_gen, home-assistant, cron, kanban and computer-use. -_CODING_TOOLS = _core_without( - "image_generate", "text_to_speech", "cronjob_manage", "computer_use", - "ha_list_entities", "ha_get_state", "ha_list_services", "ha_call_service", - kanban=False, -) +_CODING_TOOLS = _core_without("image_generate", "text_to_speech", "cronjob_manage", "computer_use", *_HA_TOOLS, kanban=False) # Core toolset definitions: individual tools or references to other toolsets. TOOLSETS = { @@ -129,12 +109,7 @@ TOOLSETS = { "browser": _ts( "Browser automation for web interaction (navigate, click, type, scroll, " "iframes, hold-click) with web search for finding URLs", - [ - "browser_navigate", "browser_snapshot", "browser_click", "browser_type", - "browser_scroll", "browser_back", "browser_press", "browser_get_images", - "browser_vision", "browser_console", "browser_cdp", "browser_dialog", - "browser_exec", "web_search", - ], + [t for t in _HERMES_CORE_TOOLS if t.startswith("browser_")] + ["web_search"], ), "cronjob": _ts( "Cronjob management tool - create, list, update, pause, resume, remove, and " @@ -178,10 +153,7 @@ TOOLSETS = { ["execute_code"], ), "delegation": _ts("Spawn subagents with isolated context for complex subtasks", ["delegate_task"]), - "homeassistant": _ts( - "Home Assistant smart home control and monitoring", - ["ha_list_entities", "ha_get_state", "ha_list_services", "ha_call_service"], - ), + "homeassistant": _ts("Home Assistant smart home control and monitoring", _HA_TOOLS), "kanban": _ts( "Kanban multi-agent coordination — only active when the agent is spawned by " "the kanban dispatcher (HERMES_KANBAN_TASK env set). The dispatcher runs " @@ -190,12 +162,7 @@ TOOLSETS = { "first-class review (request_review — not a block), return review changes, " "block for human input, heartbeat during long ops, comment on threads, attach " "files, and (for orchestrators) list, unblock, and fan out tasks.", - [ - "kanban_show", "kanban_list", "kanban_complete", "kanban_block", - "kanban_request_review", "kanban_request_changes", "kanban_heartbeat", - "kanban_comment", "kanban_create", "kanban_link", "kanban_unblock", - "kanban_attach", "kanban_attach_url", "kanban_attachments", - ], + [t for t in _HERMES_CORE_TOOLS if t.startswith("kanban_")], ), "discord": _ts( "Discord read and participate tools (fetch messages, search members, create threads)", @@ -205,21 +172,9 @@ TOOLSETS = { "Discord server management (list channels/roles, pin messages, assign roles)", ["discord_admin"], ), - "yuanbao": _ts( - "Yuanbao platform tools - group info, member queries, DM, stickers", - [ - "yb_query_group_info", "yb_query_group_members", "yb_send_dm", "yb_search_sticker", - "yb_send_sticker", - ], - ), + "yuanbao": _ts("Yuanbao platform tools - group info, member queries, DM, stickers", _YUANBAO_TOOLS), "feishu_doc": _ts("Read Feishu/Lark document content", ["feishu_doc_read"]), - "feishu_drive": _ts( - "Feishu/Lark document comment operations (list, reply, add)", - [ - "feishu_drive_list_comments", "feishu_drive_list_comment_replies", - "feishu_drive_reply_comment", "feishu_drive_add_comment", - ], - ), + "feishu_drive": _ts("Feishu/Lark document comment operations (list, reply, add)", _FEISHU_TOOLS[1:]), "spotify": _ts( "Native Spotify playback, search, playlist, album, and library tools", [ @@ -286,14 +241,7 @@ TOOLSETS = { "hermes-mattermost": _bundle("Mattermost bot toolset - self-hosted team messaging (full access)"), "hermes-matrix": _bundle("Matrix bot toolset - decentralized encrypted messaging (full access)"), "hermes-dingtalk": _bundle("DingTalk bot toolset - enterprise messaging platform (full access)"), - "hermes-feishu": _bundle( - "Feishu/Lark bot toolset - enterprise messaging via Feishu/Lark (full access)", - [ - "feishu_doc_read", "feishu_drive_list_comments", - "feishu_drive_list_comment_replies", "feishu_drive_reply_comment", - "feishu_drive_add_comment", - ], - ), + "hermes-feishu": _bundle("Feishu/Lark bot toolset - enterprise messaging via Feishu/Lark (full access)", _FEISHU_TOOLS), "hermes-weixin": _bundle("Weixin bot toolset - personal WeChat messaging via iLink (full access)"), "hermes-qqbot": _bundle("QQBot toolset - QQ messaging via Official Bot API v2 (full access)"), "hermes-wecom": _bundle("WeCom bot toolset - enterprise WeChat messaging (full access)"), @@ -302,10 +250,7 @@ TOOLSETS = { ), "hermes-yuanbao": { "description": "Yuanbao Bot 元宝消息平台工具集 - 群信息、成员查询、私聊、贴纸表情", - "tools": _HERMES_CORE_TOOLS + [ - "yb_query_group_info", "yb_query_group_members", "yb_send_dm", - "yb_search_sticker", "yb_send_sticker", - ], + "tools": _HERMES_CORE_TOOLS + _YUANBAO_TOOLS, "module": "tools.yuanbao_tools", "includes": [], }, @@ -337,30 +282,33 @@ def _registry(): return None +def _registry_call(method: str, default): + """registry.() or *default* when the registry is unavailable or the call fails.""" + registry = _registry() + if registry is None: + return default + try: + return getattr(registry, method)() + except Exception: + return default + + def _registry_generation() -> Tuple[int, int]: reg = _registry() return (id(reg), getattr(reg, "_generation", 0)) if reg is not None else (0, 0) -def _static_copy(toolset: Dict[str, Any]) -> Dict[str, Any]: - return { - **toolset, - "tools": list(toolset.get("tools", [])), - "includes": list(toolset.get("includes", [])), - } - - def get_toolset(name: str, *, include_registry: bool = True) -> Optional[Dict[str, Any]]: - """Return a toolset definition, or None if unknown. + """Toolset definition, or None if unknown. - include_registry=True merges tools plugins/overlays registered into this - toolset and resolves registry-only (plugin/MCP) toolsets and aliases. - include_registry=False returns only the static TOOLSETS entry (copied), so - platform reverse-mapping (#49622) is unaffected by registry additions. + include_registry=True merges plugin/overlay tools registered into this toolset + and resolves registry-only (plugin/MCP) toolsets and aliases; False returns a + copy of the static TOOLSETS entry only, so platform reverse-mapping is + unaffected by registry additions. """ toolset = TOOLSETS.get(name) if not include_registry: - return _static_copy(toolset) if toolset else None + return {**toolset, "tools": list(toolset.get("tools", [])), "includes": list(toolset.get("includes", []))} if toolset else None registry = _registry() if registry is None: @@ -391,24 +339,24 @@ def get_toolset(name: str, *, include_registry: bool = True) -> Optional[Dict[st def bundle_non_core_tools(toolset_name: str) -> Set[str]: """A bundle's tools minus _HERMES_CORE_TOOLS (one level of includes). - Bundles are `_HERMES_CORE_TOOLS + extras`; disabling one must not strip the - core tools every other toolset shares. One `includes` pass suffices because - only hermes-gateway nests bundles. Unknown names: full resolution minus core. + Disabling a `core + extras` bundle must not strip the core tools every other + toolset shares. One `includes` pass suffices (only hermes-gateway nests + bundles). Unknown names: full resolution minus core. """ core = set(_HERMES_CORE_TOOLS) ts_def = get_toolset(toolset_name) if not (ts_def and "tools" in ts_def): return set(resolve_toolset(toolset_name)) - core - to_remove = set(ts_def["tools"]) - core + to_remove = set(ts_def["tools"]) for inc in ts_def.get("includes", []): inc_def = get_toolset(inc) if inc_def and "tools" in inc_def: - to_remove.update(set(inc_def["tools"]) - core) - return to_remove + to_remove.update(inc_def["tools"]) + return to_remove - core -# Memo keyed on (name, include_registry, id(registry), registry generation). -# Engages only at the public entry (visited is None); recursion is untouched. +# Memo keyed on (name, include_registry, id(registry), registry generation); +# engages only at the public entry (visited is None). _resolve_toolset_memo: Dict[Tuple[str, bool, int, int], List[str]] = {} @@ -422,23 +370,19 @@ def _plugin_platform_bundle(name: str) -> List[str]: from gateway.platform_registry import platform_registry if not platform_registry.is_registered(platform_name): return [] - tools = set(_HERMES_CORE_TOOLS) - registry = _registry() - if registry is not None: - try: - tools.update(e.name for e in registry.get_all_entries() if e.toolset == platform_name) - except Exception: - pass - return list(tools) except Exception: return [] + tools = set(_HERMES_CORE_TOOLS) + try: + tools.update(e.name for e in _registry_call("get_all_entries", ()) if e.toolset == platform_name) + except Exception: + pass + return list(tools) def resolve_toolset(name: str, visited: Set[str] = None, *, include_registry: bool = True) -> List[str]: """Recursively resolve a toolset (and its includes) to a sorted tool-name list. - - include_registry=False resolves the static TOOLSETS view only (#49622). - """ + include_registry=False resolves the static TOOLSETS view only.""" external_call = visited is None if external_call: memo_key = (name, include_registry, *_registry_generation()) @@ -454,8 +398,7 @@ def resolve_toolset(name: str, visited: Set[str] = None, *, include_registry: bo all_tools.update(resolve_toolset(toolset_name, visited.copy(), include_registry=include_registry)) return sorted(all_tools) - # Diamond include or cycle: return [] silently — the tools were (or will - # be) collected via another path, so this is not an error. + # Diamond include or cycle: [] silently — the tools are collected via another path. if name in visited: return [] visited.add(name) @@ -470,32 +413,19 @@ def resolve_toolset(name: str, visited: Set[str] = None, *, include_registry: bo result = sorted(tools) if external_call: - # Stale-generation entries are never hit again; bound the memo. - if len(_resolve_toolset_memo) >= 256: + if len(_resolve_toolset_memo) >= 256: # stale-generation entries are never hit again _resolve_toolset_memo.clear() - _resolve_toolset_memo[(name, include_registry, *_registry_generation())] = list(result) + _resolve_toolset_memo[memo_key] = list(result) return result def _get_plugin_toolset_names() -> Set[str]: """Registry toolset names absent from the static TOOLSETS dict.""" - registry = _registry() - if registry is None: - return set() - try: - return {n for n in registry.get_registered_toolset_names() if n not in TOOLSETS} - except Exception: - return set() + return {n for n in _registry_call("get_registered_toolset_names", ()) if n not in TOOLSETS} def _get_registry_toolset_aliases() -> Dict[str, str]: - registry = _registry() - if registry is None: - return {} - try: - return registry.get_registered_toolset_aliases() - except Exception: - return {} + return _registry_call("get_registered_toolset_aliases", {}) def _display_alias(ts_name: str, aliases: Dict[str, str]) -> Optional[str]: @@ -527,23 +457,13 @@ def get_toolset_names() -> List[str]: def validate_toolset(name: str) -> bool: - if name in {"all", "*"} or name in TOOLSETS: - return True - return name in _get_plugin_toolset_names() or name in _get_registry_toolset_aliases() + return (name in {"all", "*"} or name in TOOLSETS + or name in _get_plugin_toolset_names() or name in _get_registry_toolset_aliases()) -def create_custom_toolset( - name: str, - description: str, - tools: List[str] = None, - includes: List[str] = None -) -> None: +def create_custom_toolset(name: str, description: str, tools: List[str] = None, includes: List[str] = None) -> None: """Register a runtime toolset in TOOLSETS.""" - TOOLSETS[name] = { - "description": description, - "tools": tools or [], - "includes": includes or [] - } + TOOLSETS[name] = _ts(description, tools or [], includes or []) def get_toolset_info(name: str) -> Dict[str, Any]: @@ -553,11 +473,8 @@ def get_toolset_info(name: str) -> Dict[str, Any]: return None resolved_tools = resolve_toolset(name) return { - "name": name, - "description": toolset["description"], - "direct_tools": toolset["tools"], - "includes": toolset["includes"], - "resolved_tools": resolved_tools, - "tool_count": len(resolved_tools), - "is_composite": bool(toolset["includes"]) + "name": name, "description": toolset["description"], + "direct_tools": toolset["tools"], "includes": toolset["includes"], + "resolved_tools": resolved_tools, "tool_count": len(resolved_tools), + "is_composite": bool(toolset["includes"]), }