feat(mcp): add tool_search for on-demand MCP schema fetching

model_tools.py

@@ -294,11 +294,20 @@ def get_tool_definitions(
             cfg_fp = (cfg_stat.st_mtime_ns, cfg_stat.st_size)
         except (FileNotFoundError, OSError, ImportError):
             cfg_fp = None
+        # Include the deferred pool generation: tool_search promotions
+        # mutate which MCP tool schemas appear in filtered_tools, so a
+        # cache key blind to it would serve stale lists.
+        try:
+            from tools.deferred_pool import get_pool as _get_def_pool
+            deferred_gen = _get_def_pool().generation
+        except Exception:
+            deferred_gen = 0
         cache_key = (
             frozenset(enabled_toolsets) if enabled_toolsets is not None else None,
             frozenset(disabled_toolsets) if disabled_toolsets else None,
             registry._generation,
             cfg_fp,
+            deferred_gen,
         )
         cached = _tool_defs_cache.get(cache_key)
         if cached is not None:
@@ -449,6 +458,14 @@ def _compute_tool_definitions(
                     }
                     break
 
+    # MCP Tool Search: when enabled, strip MCP tool schemas out of self.tools
+    # and park them in the deferred pool. The model interacts with them via
+    # the `tool_search` tool (tools/mcp_tool_search.py). See issue #6839.
+    try:
+        filtered_tools = _apply_mcp_tool_search(filtered_tools, quiet_mode)
+    except Exception as e:  # pragma: no cover — defensive
+        logger.warning("MCP tool_search pass skipped: %s", e)
+
     if not quiet_mode:
         if filtered_tools:
             tool_names = [t["function"]["name"] for t in filtered_tools]
@@ -474,6 +491,132 @@ def _compute_tool_definitions(
     return filtered_tools
 
 
+# =============================================================================
+# MCP Tool Search support
+# =============================================================================
+
+def _get_tool_search_config() -> Tuple[bool, bool, int]:
+    """Return (enabled, always_active, threshold_chars). Defaults: off."""
+    try:
+        from hermes_cli.config import load_config
+        cfg = load_config() or {}
+        mcp_cfg = (cfg.get("mcp") or {}).get("tool_search") or {}
+        enabled = bool(mcp_cfg.get("enabled", False))
+        always_active = bool(mcp_cfg.get("always_active", False))
+        threshold = int(mcp_cfg.get("threshold_chars", 8000))
+        return enabled, always_active, threshold
+    except Exception:
+        return False, False, 8000
+
+
+def _estimate_schema_chars(tool_def: Dict[str, Any]) -> int:
+    """Cheap proxy for tokens: serialized JSON length."""
+    try:
+        return len(json.dumps(tool_def, ensure_ascii=False))
+    except Exception:
+        return 0
+
+
+def _apply_mcp_tool_search(
+    filtered_tools: List[Dict[str, Any]],
+    quiet_mode: bool,
+) -> List[Dict[str, Any]]:
+    """Strip MCP tool schemas and park them in the deferred pool.
+
+    When ``mcp.tool_search.enabled: true`` and either ``always_active`` or
+    total MCP-tool schema size exceeds ``threshold_chars``, every tool whose
+    toolset starts with ``mcp-`` is removed from the returned list and
+    inserted into the deferred pool with a 1-line summary. The ``tool_search``
+    tool's check_fn (``len(pool) > 0``) then unhides it on the next pass.
+
+    Tools that the model has already "selected" via tool_search.* on a prior
+    turn (i.e. were removed from the pool) flow through unchanged.
+    """
+    enabled, always_active, threshold = _get_tool_search_config()
+    if not enabled:
+        return filtered_tools
+
+    from tools.deferred_pool import get_pool
+    pool = get_pool()
+    rget = registry.get_toolset_for_tool
+
+    promoted = pool.promoted_names()
+    mcp_tools: List[Tuple[int, Dict[str, Any], str]] = []
+    other_tools: List[Dict[str, Any]] = []
+    for td in filtered_tools:
+        name = td.get("function", {}).get("name", "")
+        toolset = rget(name) or ""
+        if toolset.startswith("mcp-"):
+            if name in promoted:
+                # Model has already fetched this schema via tool_search —
+                # keep it directly callable.
+                other_tools.append(td)
+            else:
+                mcp_tools.append((_estimate_schema_chars(td), td, toolset))
+        else:
+            other_tools.append(td)
+
+    if not mcp_tools:
+        return filtered_tools
+
+    total_chars = sum(c for c, _, _ in mcp_tools)
+    if not always_active and total_chars < threshold:
+        # Below threshold: hand back unchanged but make sure the pool is
+        # empty so tool_search's check_fn correctly hides it.
+        if len(pool) > 0:
+            pool.clear()
+        return filtered_tools
+
+    # Above threshold (or always_active): move every MCP tool into the pool,
+    # excluding any whose name has been re-promoted to other_tools via
+    # tool_search select. (Re-promoted tools won't appear in `mcp_tools`
+    # because they pass the toolset check but were removed from the pool —
+    # so they stay visible in self.tools.) We simply clear+repopulate to
+    # avoid stale entries when MCP servers refresh.
+    pool.clear()
+    for _, td, toolset in mcp_tools:
+        fn = td.get("function", {})
+        name = fn.get("name", "")
+        desc = fn.get("description", "") or ""
+        # 1-line summary: first sentence or first 120 chars
+        summary = desc.split("\n", 1)[0].strip()
+        if len(summary) > 120:
+            summary = summary[:117] + "..."
+        pool.put(
+            name=name,
+            schema={
+                "name": name,
+                "description": desc,
+                "parameters": fn.get("parameters", {}),
+            },
+            summary=summary,
+            toolset=toolset,
+        )
+
+    # Ensure tool_search itself is visible. Its check_fn gates on pool size,
+    # which was 0 at registry.get_definitions() time (before we filled the
+    # pool above), so it gets filtered out of `filtered_tools`. Re-inject
+    # from the registry directly — it's our handle to fetch the parked MCP
+    # schemas back, so it MUST be in the returned list.
+    if not any(t.get("function", {}).get("name") == "tool_search" for t in other_tools):
+        ts_entry = registry.get_entry("tool_search")
+        if ts_entry is not None:
+            other_tools.append({"type": "function", "function": ts_entry.schema})
+
+    logger.info(
+        "MCP tool_search active: parked %d MCP tool(s) (~%d chars saved); "
+        "%d promoted",
+        len(mcp_tools), total_chars, len(promoted),
+    )
+    if not quiet_mode:
+        print(
+            f"🔍 MCP tool_search active: parked {len(mcp_tools)} MCP tool(s) "
+            f"(~{total_chars} chars saved) — use `tool_search` to fetch"
+        )
+
+    return other_tools
+
+
 # =============================================================================
 # handle_function_call  (the main dispatcher)
 # =============================================================================

run_agent.py

@@ -6096,6 +6096,30 @@ def _build_system_prompt_parts(self, system_message: str = None) -> Dict[str, st
             from agent.prompt_builder import COMPUTER_USE_GUIDANCE
             stable_parts.append(COMPUTER_USE_GUIDANCE)
 
+        # MCP Tool Search: when active, surface the deferred-tool roster so
+        # the model knows which names exist behind ``tool_search``. Without
+        # this block the model can't discover MCP tools by name. See
+        # tools/mcp_tool_search.py + issue #6839.
+        if "tool_search" in self.valid_tool_names:
+            try:
+                from tools.deferred_pool import get_pool as _get_def_pool
+                _summaries = _get_def_pool().summaries()
+            except Exception:
+                _summaries = []
+            if _summaries:
+                # Group by toolset for readability (mcp-github, mcp-slack, ...).
+                _by_ts: Dict[str, List[Tuple[str, str]]] = {}
+                for _n, _ts, _s in _summaries:
+                    _by_ts.setdefault(_ts, []).append((_n, _s))
+                _lines = [
+                    "Deferred MCP tools (full schemas not loaded; use `tool_search` to fetch):",
+                ]
+                for _ts in sorted(_by_ts.keys()):
+                    _lines.append(f"  [{_ts}]")
+                    for _n, _s in sorted(_by_ts[_ts]):
+                        _lines.append(f"    - {_n}: {_s}" if _s else f"    - {_n}")
+                stable_parts.append("\n".join(_lines))
+
         nous_subscription_prompt = build_nous_subscription_prompt(self.valid_tool_names)
         if nous_subscription_prompt:
             stable_parts.append(nous_subscription_prompt)
@@ -11242,6 +11266,25 @@ def _run_tool(index, tool_call, function_name, function_args):
                 logger.info("tool %s failed (%.2fs): %s", function_name, duration, result[:200])
             else:
                 logger.info("tool %s completed (%.2fs, %d chars)", function_name, duration, len(result))
+                # MCP Tool Search: a successful tool_search promotion
+                # mutates the deferred pool, so rebuild self.tools/
+                # valid_tool_names so promoted MCP tools become callable on
+                # the next round. Invalidates the cached system prompt so
+                # the deferred-roster block re-renders. See issue #6839.
+                if function_name == "tool_search":
+                    try:
+                        self.tools = get_tool_definitions(
+                            enabled_toolsets=self.enabled_toolsets,
+                            disabled_toolsets=self.disabled_toolsets,
+                            quiet_mode=self.quiet_mode,
+                        )
+                        self.valid_tool_names = (
+                            {tool["function"]["name"] for tool in self.tools}
+                            if self.tools else set()
+                        )
+                        self._cached_system_prompt = None
+                    except Exception as _tsfx:
+                        logger.warning("tool_search post-promotion refresh failed: %s", _tsfx)
             results[index] = (function_name, function_args, result, duration, is_error, False)
             # Tear down worker-tid tracking.  Clear any interrupt bit we may
             # have set so the next task scheduled onto this recycled tid
@@ -11795,6 +11838,22 @@ def _execute_tool_calls_sequential(self, assistant_message, messages: list, effe
                 logger.warning("Tool %s returned error (%.2fs): %s", function_name, tool_duration, result_preview)
             else:
                 logger.info("tool %s completed (%.2fs, %d chars)", function_name, tool_duration, _result_len)
+                # MCP Tool Search: sequential dispatch path (see also the
+                # parallel path above). See issue #6839.
+                if function_name == "tool_search":
+                    try:
+                        self.tools = get_tool_definitions(
+                            enabled_toolsets=self.enabled_toolsets,
+                            disabled_toolsets=self.disabled_toolsets,
+                            quiet_mode=self.quiet_mode,
+                        )
+                        self.valid_tool_names = (
+                            {tool["function"]["name"] for tool in self.tools}
+                            if self.tools else set()
+                        )
+                        self._cached_system_prompt = None
+                    except Exception as _tsfx:
+                        logger.warning("tool_search post-promotion refresh failed: %s", _tsfx)
 
             # Track file-mutation outcome for the turn-end verifier.  See
             # the concurrent path for the rationale; both paths must feed

tools/deferred_pool.py

@@ -0,0 +1,117 @@
+"""Deferred tool pool for ToolSearch (MCP context-bloat mitigation).
+
+When ``mcp.tool_search.enabled: true``, MCP tool schemas are removed from
+the per-call ``self.tools`` array and parked here. The model sees only a
+compact list of (name, 1-line summary) in the system message and uses the
+``tool_search`` tool to fetch full schemas on demand. On a successful
+select, the schema is promoted back into ``self.tools`` / ``valid_tool_names``
+so subsequent calls dispatch normally.
+
+Thread-safe; designed for the long-lived Gateway process where MCP refresh
+notifications can mutate the pool while another agent thread is reading.
+"""
+from __future__ import annotations
+
+import threading
+from typing import Any, Dict, List, Optional, Set, Tuple
+
+
+class DeferredToolPool:
+    __slots__ = ("_entries", "_promoted", "_lock", "_generation")
+
+    def __init__(self) -> None:
+        self._entries: Dict[str, Dict[str, Any]] = {}
+        # Names the model has fetched via tool_search and should stay visible
+        # in self.tools across repopulation passes.
+        self._promoted: Set[str] = set()
+        self._lock = threading.RLock()
+        self._generation: int = 0
+
+    def put(self, name: str, schema: Dict[str, Any], summary: str, toolset: str) -> None:
+        with self._lock:
+            self._entries[name] = {
+                "name": name,
+                "schema": schema,
+                "summary": summary,
+                "toolset": toolset,
+            }
+            self._generation += 1
+
+    def remove(self, name: str) -> Optional[Dict[str, Any]]:
+        with self._lock:
+            entry = self._entries.pop(name, None)
+            if entry is not None:
+                self._promoted.add(name)
+                self._generation += 1
+            return entry
+
+    def is_promoted(self, name: str) -> bool:
+        with self._lock:
+            return name in self._promoted
+
+    def promoted_names(self) -> Set[str]:
+        with self._lock:
+            return set(self._promoted)
+
+    def reset_promotions(self) -> None:
+        """Clear the promoted set — call when MCP servers refresh and the
+        per-session promotion state should be discarded."""
+        with self._lock:
+            if self._promoted:
+                self._promoted.clear()
+                self._generation += 1
+
+    def remove_by_toolset(self, toolset: str) -> List[str]:
+        with self._lock:
+            removed = [n for n, e in self._entries.items() if e["toolset"] == toolset]
+            for n in removed:
+                self._entries.pop(n, None)
+            if removed:
+                self._generation += 1
+            return removed
+
+    def get(self, name: str) -> Optional[Dict[str, Any]]:
+        with self._lock:
+            return self._entries.get(name)
+
+    def names(self) -> List[str]:
+        with self._lock:
+            return list(self._entries.keys())
+
+    def items(self) -> List[Tuple[str, Dict[str, Any]]]:
+        with self._lock:
+            return list(self._entries.items())
+
+    def summaries(self) -> List[Tuple[str, str, str]]:
+        """Return (name, toolset, summary) for system-message rendering."""
+        with self._lock:
+            return [(n, e["toolset"], e["summary"]) for n, e in self._entries.items()]
+
+    def __len__(self) -> int:
+        with self._lock:
+            return len(self._entries)
+
+    def __contains__(self, name: str) -> bool:
+        with self._lock:
+            return name in self._entries
+
+    @property
+    def generation(self) -> int:
+        return self._generation
+
+    def clear(self) -> None:
+        """Clear pool entries only — preserves promotion history.
+
+        Use ``reset_promotions()`` separately to forget which tools the model
+        has already fetched (e.g. after an MCP server schema refresh).
+        """
+        with self._lock:
+            self._entries.clear()
+            self._generation += 1
+
+
+_pool = DeferredToolPool()
+
+
+def get_pool() -> DeferredToolPool:
+    return _pool