feat: implement comprehensive MCP tool filtering with static and dynamic support

Author: devtalker

docs/mcp.md

@@ -43,16 +43,24 @@ agent=Agent(
 
 ## Tool filtering
 
-You can filter which tools are available to your Agent using server-level filtering:
+You can filter which tools are available to your Agent by configuring tool filters on MCP servers. The SDK supports both static and dynamic tool filtering.
+
+### Static tool filtering
+
+For simple allow/block lists, you can use static filtering:
 
 ```python
+from agents.mcp import create_static_tool_filter
+
 # Only expose specific tools from this server
 server = MCPServerStdio(
     params={
         "command": "npx",
         "args": ["-y", "@modelcontextprotocol/server-filesystem", samples_dir],
     },
-    allowed_tools=["read_file", "write_file"],  # Only these tools will be available
+    tool_filter=create_static_tool_filter(
+        allowed_tool_names=["read_file", "write_file"]
+    )
 )
 
 # Exclude specific tools from this server
@@ -61,10 +69,65 @@ server = MCPServerStdio(
         "command": "npx", 
         "args": ["-y", "@modelcontextprotocol/server-filesystem", samples_dir],
     },
-    excluded_tools=["delete_file"],  # This tool will be filtered out
+    tool_filter=create_static_tool_filter(
+        blocked_tool_names=["delete_file"]
+    )
+)
+
+```
+
+**When both `allowed_tool_names` and `blocked_tool_names` are configured, the processing order is:**
+1. First apply `allowed_tool_names` (allowlist) - only keep the specified tools
+2. Then apply `blocked_tool_names` (blocklist) - exclude specified tools from the remaining tools
+
+For example, if you configure `allowed_tool_names=["read_file", "write_file", "delete_file"]` and `blocked_tool_names=["delete_file"]`, only `read_file` and `write_file` tools will be available.
+
+### Dynamic tool filtering
+
+For more complex filtering logic, you can use dynamic filters with functions:
+
+```python
+from agents.mcp import ToolFilterContext
+
+# Simple synchronous filter
+def custom_filter(context: ToolFilterContext, tool) -> bool:
+    """Example of a custom tool filter."""
+    # Filter logic based on tool name patterns
+    return tool.name.startswith("allowed_prefix")
+
+# Context-aware filter
+def context_aware_filter(context: ToolFilterContext, tool) -> bool:
+    """Filter tools based on context information."""
+    # Access agent information
+    agent_name = context.agent.name
+
+    # Access server information  
+    server_name = context.server_name
+
+    # Implement your custom filtering logic here
+    return some_filtering_logic(agent_name, server_name, tool)
+
+# Asynchronous filter
+async def async_filter(context: ToolFilterContext, tool) -> bool:
+    """Example of an asynchronous filter."""
+    # Perform async operations if needed
+    result = await some_async_check(context, tool)
+    return result
+
+server = MCPServerStdio(
+    params={
+        "command": "npx",
+        "args": ["-y", "@modelcontextprotocol/server-filesystem", samples_dir],
+    },
+    tool_filter=custom_filter  # or context_aware_filter or async_filter
 )
 ```
 
+The `ToolFilterContext` provides access to:
+- `run_context`: The current run context
+- `agent`: The agent requesting the tools 
+- `server_name`: The name of the MCP server
+
 ## Caching
 
 Every time an Agent runs, it calls `list_tools()` on the MCP server. This can be a latency hit, especially if the server is a remote server. To automatically cache the list of tools, you can pass `cache_tools_list=True` to [`MCPServerStdio`][agents.mcp.server.MCPServerStdio], [`MCPServerSse`][agents.mcp.server.MCPServerSse], and [`MCPServerStreamableHttp`][agents.mcp.server.MCPServerStreamableHttp]. You should only do this if you're certain the tool list will not change.

src/agents/agent.py

@@ -256,14 +256,18 @@ async def get_prompt(
         """Get the prompt for the agent."""
         return await PromptUtil.to_model_input(self.prompt, run_context, self)
 
-    async def get_mcp_tools(self) -> list[Tool]:
+    async def get_mcp_tools(
+        self, run_context: RunContextWrapper[TContext] | None = None
+    ) -> list[Tool]:
         """Fetches the available tools from the MCP servers."""
         convert_schemas_to_strict = self.mcp_config.get("convert_schemas_to_strict", False)
-        return await MCPUtil.get_all_function_tools(self.mcp_servers, convert_schemas_to_strict)
+        return await MCPUtil.get_all_function_tools(
+            self.mcp_servers, convert_schemas_to_strict, run_context, self
+        )
 
     async def get_all_tools(self, run_context: RunContextWrapper[Any]) -> list[Tool]:
         """All agent tools, including MCP tools and function tools."""
-        mcp_tools = await self.get_mcp_tools()
+        mcp_tools = await self.get_mcp_tools(run_context)
 
         async def _check_tool_enabled(tool: Tool) -> bool:
             if not isinstance(tool, FunctionTool):

src/agents/mcp/server.py

@@ -2,10 +2,11 @@
 
 import abc
 import asyncio
+import inspect
 from contextlib import AbstractAsyncContextManager, AsyncExitStack
 from datetime import timedelta
 from pathlib import Path
-from typing import Any, Literal
+from typing import TYPE_CHECKING, Any, Literal, cast
 
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
 from mcp import ClientSession, StdioServerParameters, Tool as MCPTool, stdio_client
@@ -17,7 +18,11 @@
 
 from ..exceptions import UserError
 from ..logger import logger
-from .util import ToolFilter, ToolFilterStatic
+from ..run_context import RunContextWrapper
+from .util import ToolFilter, ToolFilterCallable, ToolFilterContext, ToolFilterStatic
+
+if TYPE_CHECKING:
+    from ..agent import Agent
 
 
 class MCPServer(abc.ABC):
@@ -45,7 +50,11 @@ async def cleanup(self):
         pass
 
     @abc.abstractmethod
-    async def list_tools(self) -> list[MCPTool]:
+    async def list_tools(
+        self,
+        run_context: RunContextWrapper[Any] | None = None,
+        agent: Agent[Any] | None = None,
+    ) -> list[MCPTool]:
         """List the tools available on the server."""
         pass
 
@@ -90,38 +99,106 @@ def __init__(
 
         self.tool_filter = tool_filter
 
-    def _apply_tool_filter(self, tools: list[MCPTool]) -> list[MCPTool]:
+    async def _apply_tool_filter(
+        self,
+        tools: list[MCPTool],
+        run_context: RunContextWrapper[Any] | None,
+        agent: Agent[Any] | None,
+    ) -> list[MCPTool]:
         """Apply the tool filter to the list of tools."""
         if self.tool_filter is None:
             return tools
 
         # Handle static tool filter
         if isinstance(self.tool_filter, dict):
-            static_filter: ToolFilterStatic = self.tool_filter
-            filtered_tools = tools
+            return self._apply_static_tool_filter(tools, self.tool_filter)
 
-            # Apply allowed_tool_names filter (whitelist)
-            if "allowed_tool_names" in static_filter:
-                allowed_names = static_filter["allowed_tool_names"]
-                filtered_tools = [t for t in filtered_tools if t.name in allowed_names]
+        # Handle callable tool filter (dynamic filter)
+        else:
+            return await self._apply_dynamic_tool_filter(tools, run_context, agent)
 
-            # Apply blocked_tool_names filter (blacklist)
-            if "blocked_tool_names" in static_filter:
-                blocked_names = static_filter["blocked_tool_names"]
-                filtered_tools = [t for t in filtered_tools if t.name not in blocked_names]
+    def _apply_static_tool_filter(
+        self,
+        tools: list[MCPTool],
+        static_filter: ToolFilterStatic
+    ) -> list[MCPTool]:
+        """Apply static tool filtering based on allowlist and blocklist."""
+        filtered_tools = tools
 
-            return filtered_tools
+        # Apply allowed_tool_names filter (whitelist)
+        if "allowed_tool_names" in static_filter:
+            allowed_names = static_filter["allowed_tool_names"]
+            filtered_tools = [t for t in filtered_tools if t.name in allowed_names]
 
-        # Handle callable tool filter
-        # For now, we can't support callable filters because we don't have access to
-        # run context and agent in the current list_tools signature.
-        # This could be enhanced in the future by modifying the call chain.
-        else:
-            raise NotImplementedError(
-                "Callable tool filters are not yet supported. Please use ToolFilterStatic "
-                "with 'allowed_tool_names' and/or 'blocked_tool_names' for now."
+        # Apply blocked_tool_names filter (blacklist)
+        if "blocked_tool_names" in static_filter:
+            blocked_names = static_filter["blocked_tool_names"]
+            filtered_tools = [t for t in filtered_tools if t.name not in blocked_names]
+
+        return filtered_tools
+
+    async def _apply_dynamic_tool_filter(
+        self,
+        tools: list[MCPTool],
+        run_context: RunContextWrapper[Any] | None,
+        agent: Agent[Any] | None,
+    ) -> list[MCPTool]:
+        """Apply dynamic tool filtering using a callable filter function."""
+
+        # Ensure we have a callable filter and cast to help mypy
+        if not callable(self.tool_filter):
+            raise ValueError("Tool filter must be callable for dynamic filtering")
+        tool_filter_func = cast(ToolFilterCallable, self.tool_filter)
+
+        # Create filter context - it may be None if run_context or agent is None
+        filter_context = None
+        if run_context is not None and agent is not None:
+            filter_context = ToolFilterContext(
+                run_context=run_context,
+                agent=agent,
+                server_name=self.name,
             )
 
+        filtered_tools = []
+        for tool in tools:
+            try:
+                # Try to call the filter function
+                if filter_context is not None:
+                    # We have full context, call with context
+                    result = tool_filter_func(filter_context, tool)
+                else:
+                    # Try to call without context first to see if it works
+                    try:
+                        # Some filters might not need context parameters at all
+                        result = tool_filter_func(None, tool)
+                    except (TypeError, AttributeError) as e:
+                        # If the filter tries to access context attributes, raise a helpful error
+                        raise UserError(
+                            "Dynamic tool filters require both run_context and agent when the "
+                            "filter function accesses context information. This typically happens "
+                            "when calling list_tools() directly without these parameters. Either "
+                            "provide both parameters or use a static tool filter instead."
+                        ) from e
+
+                if inspect.isawaitable(result):
+                    should_include = await result
+                else:
+                    should_include = result
+
+                if should_include:
+                    filtered_tools.append(tool)
+            except UserError:
+                # Re-raise UserError as-is (this includes our context requirement error)
+                raise
+            except Exception as e:
+                logger.error(
+                    f"Error applying tool filter to tool '{tool.name}' on server '{self.name}': {e}"
+                )
+                # On error, exclude the tool for safety
+                continue
+
+        return filtered_tools
+
     @abc.abstractmethod
     def create_streams(
         self,
@@ -172,7 +249,11 @@ async def connect(self):
             await self.cleanup()
             raise
 
-    async def list_tools(self) -> list[MCPTool]:
+    async def list_tools(
+        self,
+        run_context: RunContextWrapper[Any] | None = None,
+        agent: Agent[Any] | None = None,
+    ) -> list[MCPTool]:
         """List the tools available on the server."""
         if not self.session:
             raise UserError("Server not initialized. Make sure you call `connect()` first.")
@@ -190,7 +271,7 @@ async def list_tools(self) -> list[MCPTool]:
         # Filter tools based on tool_filter
         filtered_tools = tools
         if self.tool_filter is not None:
-            filtered_tools = self._apply_tool_filter(filtered_tools)
+            filtered_tools = await self._apply_tool_filter(filtered_tools, run_context, agent)
         return filtered_tools
 
     async def call_tool(self, tool_name: str, arguments: dict[str, Any] | None) -> CallToolResult:

src/agents/mcp/util.py

@@ -2,6 +2,7 @@
 import json
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Any, Callable, Union
+
 from typing_extensions import NotRequired, TypedDict
 
 from agents.strict_schema import ensure_strict_json_schema
@@ -35,11 +36,12 @@ class ToolFilterContext:
     """The name of the MCP server."""
 
 
-ToolFilterCallable = Callable[["ToolFilterContext", "MCPTool"], MaybeAwaitable[bool]]
+ToolFilterCallable = Callable[["ToolFilterContext | None", "MCPTool"], MaybeAwaitable[bool]]
 """A function that determines whether a tool should be available.
 
 Args:
     context: The context information including run context, agent, and server name.
+             Can be None if run_context or agent is not available.
     tool: The MCP tool to filter.
 
 Returns:
@@ -51,10 +53,12 @@ class ToolFilterStatic(TypedDict):
     """Static tool filter configuration using allowlists and blocklists."""
 
     allowed_tool_names: NotRequired[list[str]]
-    """Optional list of tool names to allow (whitelist). If set, only these tools will be available."""
+    """Optional list of tool names to allow (whitelist).
+    If set, only these tools will be available."""
 
     blocked_tool_names: NotRequired[list[str]]
-    """Optional list of tool names to exclude (blacklist). If set, these tools will be filtered out."""
+    """Optional list of tool names to exclude (blacklist).
+    If set, these tools will be filtered out."""
 
 
 ToolFilter = Union[ToolFilterCallable, ToolFilterStatic, None]
@@ -93,13 +97,19 @@ class MCPUtil:
 
     @classmethod
     async def get_all_function_tools(
-        cls, servers: list["MCPServer"], convert_schemas_to_strict: bool
+        cls,
+        servers: list["MCPServer"],
+        convert_schemas_to_strict: bool,
+        run_context: RunContextWrapper[Any] | None = None,
+        agent: "Agent[Any] | None" = None,
     ) -> list[Tool]:
         """Get all function tools from a list of MCP servers."""
         tools = []
         tool_names: set[str] = set()
         for server in servers:
-            server_tools = await cls.get_function_tools(server, convert_schemas_to_strict)
+            server_tools = await cls.get_function_tools(
+                server, convert_schemas_to_strict, run_context, agent
+            )
             server_tool_names = {tool.name for tool in server_tools}
             if len(server_tool_names & tool_names) > 0:
                 raise UserError(
@@ -113,12 +123,16 @@ async def get_all_function_tools(
 
     @classmethod
     async def get_function_tools(
-        cls, server: "MCPServer", convert_schemas_to_strict: bool
+        cls,
+        server: "MCPServer",
+        convert_schemas_to_strict: bool,
+        run_context: RunContextWrapper[Any] | None = None,
+        agent: "Agent[Any] | None" = None,
     ) -> list[Tool]:
         """Get all function tools from a single MCP server."""
 
         with mcp_tools_span(server=server.name) as span:
-            tools = await server.list_tools()
+            tools = await server.list_tools(run_context, agent)
             span.span_data.result = [tool.name for tool in tools]
 
         return [cls.to_function_tool(tool, server, convert_schemas_to_strict) for tool in tools]

src/agents/tool.py

@@ -26,8 +26,8 @@
 from .util._types import MaybeAwaitable
 
 if TYPE_CHECKING:
+
     from .agent import Agent
-    from mcp.types import Tool as MCPTool
 
 ToolParams = ParamSpec("ToolParams")