Dynamic OS MCP Tool Filtering for Agents

src/agents/art/config.py

@@ -0,0 +1,106 @@
+"""ART agent configuration.
+
+Controls which MCP tool categories are available to each specialized agent.
+All settings can be overridden via environment variables (``ART_`` prefix).
+
+**Category names** (see :data:`~agents.art.specialized_agents.TOOL_GROUPS`):
+
+- ``core_tools`` — general search and index operations (SearchIndexTool, etc.)
+- ``search_relevance`` — all Search Relevance Workbench tools (experiments,
+  judgment lists, search configurations, query sets)
+- ``experiment`` — experiment lifecycle only
+- ``judgment`` — judgment list management only
+- ``search_config`` — search configuration management only
+- ``query_set`` — query set management only
+
+Individual tool names (e.g. ``GetExperimentTool``) can also be mixed in.
+
+**Examples** (via environment variables)::
+
+    # Give the UBI agent access to data-distribution as well as core tools
+    ART_UBI_AGENT_TOOLS=core_tools,DataDistributionTool
+
+    # Give the evaluation agent every search-relevance tool
+    ART_EVALUATION_AGENT_TOOLS=core_tools,search_relevance
+"""
+
+from __future__ import annotations
+
+from pydantic import field_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class ARTAgentConfig(BaseSettings):
+    """Configuration for the ART specialized agents' MCP tool access.
+
+    Each field is a list of category names and/or individual tool names.
+    Set via a comma-separated environment variable, e.g.::
+
+        ART_HYPOTHESIS_AGENT_TOOLS=core_tools,experiment,search_config,query_set
+
+    Implementation note: the fields are typed ``list[str] | str`` so that
+    pydantic-settings treats failed JSON parsing as a non-fatal parse failure
+    and passes the raw string through to the ``_normalise_list`` validator,
+    which converts it to ``list[str]``.  At runtime the value is always a
+    ``list[str]``.
+    """
+
+    model_config = SettingsConfigDict(
+        env_prefix="ART_",
+        case_sensitive=False,
+        extra="ignore",
+    )
+
+    hypothesis_agent_tools: list[str] | str = [
+        "core_tools",
+        "experiment",
+        "search_config",
+        "query_set",
+        "somTool"
+    ]
+    """Tool categories for the hypothesis agent.
+
+    Defaults to search + experiment management + search configs + query sets.
+    Judgment lists are excluded — the hypothesis agent only does pairwise
+    sanity checks, not full offline evaluation.
+    """
+
+    evaluation_agent_tools: list[str] | str = [
+        "core_tools",
+        "search_relevance",
+    ]
+    """Tool categories for the evaluation agent.
+
+    Defaults to search + all Search Relevance Workbench tools, giving the
+    agent access to experiments, judgment lists, configs, and query sets.
+    """
+
+    ubi_agent_tools: list[str] | str = [
+        "core_tools",
+    ]
+    """Tool categories for the user-behavior-analysis agent.
+
+    Defaults to core search tools only — UBI analysis is read-only queries
+    against the ubi_queries and ubi_events indices.
+    """
+
+    @field_validator(
+        "hypothesis_agent_tools",
+        "evaluation_agent_tools",
+        "ubi_agent_tools",
+        mode="before",
+    )
+    @classmethod
+    def _normalise_list(cls, v: object) -> list[str]:
+        """Normalise to ``list[str]``, trimming whitespace and dropping empty items.
+
+        Handles both a raw comma-separated string (from an env var) and a
+        plain list (from direct instantiation or a JSON-formatted env var).
+        """
+        if isinstance(v, str):
+            items: list[str] = v.split(",")
+        elif isinstance(v, list):
+            items = [str(i) for i in v]
+        else:
+            return v  # type: ignore[return-value]
+        return [item.strip() for item in items if item.strip()]

src/agents/art/specialized_agents.py

@@ -13,6 +13,8 @@
 from strands import Agent
 from strands.models.bedrock import BedrockModel
 
+from agents.art.config import ARTAgentConfig
+from agents.tool_filter import TOOL_GROUPS, _select_tools
 from utils.logging_helpers import get_logger, log_info_event
 from utils.monitored_tool import monitored_tool
 
@@ -190,6 +192,9 @@
 # Global variable to store MCP tools (will be set during initialization)
 _opensearch_tools: list = []
 
+# Agent tool configuration — reads ART_* env vars once at import time.
+_art_config = ARTAgentConfig()
+
 
 def set_opensearch_tools(tools: list[Any]) -> None:
     """Set the OpenSearch MCP tools to be used by specialized agents."""
@@ -234,9 +239,9 @@ async def hypothesis_agent(query: str) -> str:
         )
 
         hypothesis_tools = [
-            # OpenSearch MCP tools
-            *_opensearch_tools,
-            # Experiment tools
+            # OpenSearch MCP tools — driven by ART_HYPOTHESIS_AGENT_TOOLS
+            *_select_tools(_opensearch_tools, _art_config.hypothesis_agent_tools),
+            # Experiment aggregation (local computation, not an MCP tool)
             aggregate_experiment_results,
         ]
 
@@ -291,9 +296,9 @@ async def evaluation_agent(query: str) -> str:
 
         # Combine OpenSearch MCP tools with evaluation-specific tools
         evaluation_tools = [
-            # OpenSearch MCP tools
-            *_opensearch_tools,
-            # Experiment tools
+            # OpenSearch MCP tools — driven by ART_EVALUATION_AGENT_TOOLS
+            *_select_tools(_opensearch_tools, _art_config.evaluation_agent_tools),
+            # Experiment aggregation (local computation, not an MCP tool)
             aggregate_experiment_results,
         ]
 
@@ -342,8 +347,8 @@ async def user_behavior_analysis_agent(query: str) -> str:
         )
 
         ubi_tools = [
-            # OpenSearch MCP tools
-            *_opensearch_tools,
+            # OpenSearch MCP tools — driven by ART_UBI_AGENT_TOOLS
+            *_select_tools(_opensearch_tools, _art_config.ubi_agent_tools),
         ]
 
         # Create specialized agent with UBI analytics focus

src/agents/default_config.py

@@ -0,0 +1,58 @@
+"""Fallback agent configuration.
+
+Controls which MCP tool categories are available to the fallback agent.
+The setting can be overridden via the ``FALLBACK_AGENT_TOOLS`` environment
+variable.
+
+By default the fallback agent has access to **all** tools exposed by the MCP
+server (``FALLBACK_AGENT_TOOLS`` is empty).  Set it to a comma-separated list
+of category names or individual tool names to restrict access::
+
+    # Give the fallback agent only core search tools
+    FALLBACK_AGENT_TOOLS=core_tools
+
+    # Give the fallback agent core tools plus one SRW tool
+    FALLBACK_AGENT_TOOLS=core_tools,GetExperimentTool
+
+See :data:`~agents.tool_filter.TOOL_GROUPS` for valid category names.
+"""
+
+from __future__ import annotations
+
+from pydantic import field_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class FallbackAgentConfig(BaseSettings):
+    """Configuration for the fallback agent's MCP tool access.
+
+    Set via a comma-separated environment variable, e.g.::
+
+        FALLBACK_AGENT_TOOLS=core_tools,search_relevance
+
+    An empty value (the default) means all MCP tools are available.
+    """
+
+    model_config = SettingsConfigDict(
+        env_prefix="FALLBACK_",
+        case_sensitive=False,
+        extra="ignore",
+    )
+
+    agent_tools: list[str] | str = []
+    """Tool filter for the fallback agent.
+
+    Defaults to an empty list, which passes all MCP tools through unchanged.
+    """
+
+    @field_validator("agent_tools", mode="before")
+    @classmethod
+    def _normalise_list(cls, v: object) -> list[str]:
+        """Normalise to ``list[str]``, trimming whitespace and dropping empty items."""
+        if isinstance(v, str):
+            items: list[str] = v.split(",")
+        elif isinstance(v, list):
+            items = [str(i) for i in v]
+        else:
+            return v  # type: ignore[return-value]
+        return [item.strip() for item in items if item.strip()]

src/agents/fallback_agent.py

@@ -12,6 +12,8 @@
 from strands import Agent
 from strands.tools.mcp import MCPClient
 
+from agents.default_config import FallbackAgentConfig
+from agents.tool_filter import _select_tools
 from server.constants import DEFAULT_MCP_SERVER_URL
 from utils.logging_helpers import get_logger, log_info_event
 
@@ -34,16 +36,24 @@
 - If you don't have the right tool for a request, explain what's available
 """
 
+# Agent tool configuration — reads FALLBACK_* env vars once at import time.
+_fallback_config = FallbackAgentConfig()
+
 
 def create_fallback_agent(
     opensearch_url: str, headers: dict[str, str] | None = None
 ) -> Agent:
-    """Create the fallback agent with all OpenSearch MCP tools.
+    """Create the fallback agent with OpenSearch MCP tools.
 
     Connects to the OpenSearch MCP server via Streamable HTTP transport.
     The server URL defaults to ``http://localhost:3001/mcp`` and can be
     overridden with the ``MCP_SERVER_URL`` environment variable.
 
+    The set of tools available to the agent is controlled by the
+    ``FALLBACK_AGENT_TOOLS`` environment variable (comma-separated category
+    names or individual tool names).  When the variable is unset or empty
+    all MCP tools are available.
+
     Args:
         opensearch_url: OpenSearch cluster URL (informational — the MCP
             server is assumed to already be configured for this cluster).
@@ -56,19 +66,23 @@ def create_fallback_agent(
     mcp_server_url = os.getenv("MCP_SERVER_URL", DEFAULT_MCP_SERVER_URL)
 
     mcp_client = MCPClient(lambda: streamablehttp_client(mcp_server_url, headers=headers))
+    mcp_client.start()
+
+    all_tools = list(mcp_client.list_tools_sync())
+    tools = _select_tools(all_tools, _fallback_config.agent_tools)
 
     agent = Agent(
         system_prompt=FALLBACK_SYSTEM_PROMPT,
-        tools=[mcp_client],
+        tools=tools,
     )
 
-    tool_count = len(agent.tool_registry.registry)
     log_info_event(
         logger,
-        f"Fallback agent initialized with {tool_count} MCP tools "
+        f"Fallback agent initialized with {len(tools)}/{len(all_tools)} MCP tools "
         f"(server={mcp_server_url}).",
         "fallback_agent.initialized",
-        tool_count=tool_count,
+        tool_count=len(tools),
+        total_tool_count=len(all_tools),
         mcp_server_url=mcp_server_url,
         opensearch_url=opensearch_url,
     )