feat(catalog): implement meta tool filtering in catalog processing

- Introduced a new function `is_meta_tool` to identify and filter out meta tools from various catalog processing functions.
- Updated `_normalize_tool`, `_normalize_tools`, `record_tool_attempt_feedback`, and `filter_tools_for_tier_tracking` to skip processing for tools identified as meta tools.
- Enhanced `build_catalog_from_tools` and `_catalog_entries_from_dicts` to prevent meta tools from being included in the catalog.
- Improved overall catalog management by ensuring that meta tools do not interfere with standard tool operations.

Author: qdrddr

src/cyt/cyt_mcp/catalog.py

@@ -192,9 +192,13 @@ def _normalize_catalog_scope(raw: object) -> str | None:
 
 
 def _normalize_tool(tool: dict[str, Any]) -> dict[str, Any] | None:
+    from cyt_mcp.search import is_meta_tool
+
     name = str(tool.get("name") or "").strip()
     if not name:
         return None
+    if is_meta_tool(name):
+        return None
     schema = tool.get("input_schema")
     if not isinstance(schema, dict):
         schema = tool.get("inputSchema")

src/cyt/hook/catalog_registry.py

@@ -111,10 +111,16 @@ def _registry_key(
 def _normalize_tools(raw: object) -> list[dict[str, Any]]:
     if not isinstance(raw, list):
         return []
+    from cyt_mcp.search import is_meta_tool
+
     tools: list[dict[str, Any]] = []
     for item in raw:
-        if isinstance(item, dict):
-            tools.append(copy.deepcopy(item))
+        if not isinstance(item, dict):
+            continue
+        name = str(item.get("name") or "").strip()
+        if name and is_meta_tool(name):
+            continue
+        tools.append(copy.deepcopy(item))
     return tools
 
 

src/cyt/tiers/adapters/tools.py

@@ -58,9 +58,14 @@ def filter_tools_for_tier_tracking(
     effective = resolve_effective_permissions(config=config)
     deny_entries = effective.mcp.deny
     tracked: list[dict[str, Any]] = []
+    from cyt_mcp.search import is_meta_tool
+
     for tool in tools:
         if not isinstance(tool, dict):
             continue
+        catalog_name = str(tool.get("name") or "").strip()
+        if catalog_name and is_meta_tool(catalog_name):
+            continue
         stamped = stamp_tool_catalog_source(tool)
         if resolve_tool_catalog_source(stamped) not in allowed:
             continue

src/cyt/tiers/feedback.py

@@ -66,11 +66,17 @@ def record_tool_attempt_feedback(
             tool_tracked_for_config,
         )
 
+        from cyt_mcp.search import is_meta_tool
+
+        if is_meta_tool(tool_name):
+            return
         canonical_name = resolve_canonical_tool_name_for_tiers(
             tool_name,
             config=cfg,
             catalog=catalog,
         )
+        if is_meta_tool(canonical_name):
+            return
         tool = stamp_tool_catalog_source(
             {
                 "name": canonical_name,

src/cyt_mcp/catalog_build.py

@@ -12,7 +12,7 @@
 
 from cyt_mcp.config import AggregatorConfig
 from cyt_mcp.runtime_cache import RuntimeToolCache
-from cyt_mcp.search import MCP_WIRE_SEARCH_TOOL_NAME, refresh_search_tool_schema
+from cyt_mcp.search import MCP_WIRE_SEARCH_TOOL_NAME, is_meta_tool, refresh_search_tool_schema
 from cyt_mcp.tool_identity import enrich_tool_identity, tool_name_allowed_for_servers
 
 JsonScalar = str | int | float | bool | None
@@ -86,7 +86,7 @@ def build_catalog_from_tools(
     for tool in tools:
         mcp_tool = tool.to_mcp_tool()
         name = str(mcp_tool.name)
-        if name == MCP_WIRE_SEARCH_TOOL_NAME:
+        if is_meta_tool(name):
             continue
         if server_keys and not tool_name_allowed_for_servers(name, server_keys):
             continue
@@ -244,7 +244,7 @@ def _catalog_entries_from_dicts(
     search_index: dict[str, dict[str, Any]] = {}
     for raw in tools:
         name = str(raw.get("name") or "").strip()
-        if not name or name == MCP_WIRE_SEARCH_TOOL_NAME:
+        if not name or is_meta_tool(name):
             continue
         if server_keys and not tool_name_allowed_for_servers(name, server_keys):
             continue

src/cyt_mcp/search.py

@@ -29,6 +29,16 @@
 _EXCLUDED_LOOKUP_NAMES = frozenset({SEARCH_TOOL_NAME, MCP_WIRE_SEARCH_TOOL_NAME})
 
 
+def is_meta_tool(tool_name: str) -> bool:
+    """True for cyt-mcp frontend meta tools that must not enter hook/tier catalogs."""
+    normalized = str(tool_name or "").strip()
+    if not normalized:
+        return True
+    if normalized in _EXCLUDED_LOOKUP_NAMES:
+        return True
+    return normalized.endswith(("_get-tool-definitions", "__get-tool-definitions"))
+
+
 def search_tool_description(*, agent: str | None) -> str:
     description = _SEARCH_TOOL_BASE_DESCRIPTION
     if agent == "cursor":

src/cyt_mcp/tool_use_feedback_middleware.py

@@ -14,31 +14,13 @@
 from cyt_mcp.config import AggregatorConfig, load_known_mcp_server_keys
 from cyt_mcp.config_holder import ConfigHolder
 from cyt_mcp.runtime_cache import RuntimeToolCache
-from cyt_mcp.search import MCP_WIRE_SEARCH_TOOL_NAME, SEARCH_TOOL_NAME
+from cyt_mcp.search import is_meta_tool
 from cyt_mcp.session_context import get_current_session_runtime
 from cyt_mcp.tier_feedback_push import schedule_tool_use_feedback
 from cyt_mcp.tool_identity import canonical_backend_identity, resolve_backend_identity
 
 logger = logging.getLogger(__name__)
 
-_META_TOOL_NAMES = frozenset(
-    {
-        SEARCH_TOOL_NAME,
-        MCP_WIRE_SEARCH_TOOL_NAME,
-        "cyt-mcp_get-tool-definitions",
-    },
-)
-
-
-def _is_meta_tool(tool_name: str) -> bool:
-    normalized = tool_name.strip()
-    if not normalized:
-        return True
-    if normalized in _META_TOOL_NAMES:
-        return True
-    return normalized.endswith(("_get-tool-definitions", "__get-tool-definitions"))
-
-
 def _tool_call_succeeded(result: ToolResult) -> bool:
     if getattr(result, "is_error", False):
         return False
@@ -105,7 +87,7 @@ async def on_call_tool(
         arguments = getattr(params, "arguments", None)
         args = dict(arguments) if isinstance(arguments, dict) else {}
 
-        if _is_meta_tool(tool_name):
+        if is_meta_tool(tool_name):
             return await call_next(context)
 
         success = False