feat: add MCP tool filtering support

Author: devtalker

docs/mcp.md

@@ -41,6 +41,57 @@ agent=Agent(
 )
 ```
 
+## Tool filtering
+
+You can filter which tools are available to your Agent in two ways:
+
+### Server-level filtering
+
+Each MCP server instance can be configured with `allowed_tools` and `excluded_tools` parameters to control which tools it exposes:
+
+```python
+# 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
+)
+
+# Exclude specific tools from this server
+server = MCPServerStdio(
+    params={
+        "command": "npx", 
+        "args": ["-y", "@modelcontextprotocol/server-filesystem", samples_dir],
+    },
+    excluded_tools=["delete_file"],  # This tool will be filtered out
+)
+```
+
+### Agent-level filtering
+
+You can also filter tools at the Agent level using the `mcp_config` parameter. This allows you to control which tools are available across all MCP servers:
+
+```python
+agent = Agent(
+    name="Assistant",
+    instructions="Use the tools to achieve the task",
+    mcp_servers=[server1, server2, server3],
+    mcp_config={
+        "allowed_tools": {
+            "server1": ["read_file", "write_file"],  # Only these tools from server1
+            "server2": ["search"],                   # Only search tool from server2
+        },
+        "excluded_tools": {
+            "server3": ["dangerous_tool"],           # Exclude this tool from server3
+        }
+    }
+)
+```
+
+**Filtering priority**: Server-level filtering is applied first, then Agent-level filtering. This allows for fine-grained control where servers can limit their exposed tools, and Agents can further restrict which tools they use.
+
 ## 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

@@ -62,6 +62,10 @@ class MCPConfig(TypedDict):
     """If True, we will attempt to convert the MCP schemas to strict-mode schemas. This is a
     best-effort conversion, so some schemas may not be convertible. Defaults to False.
     """
+    allowed_tools: NotRequired[dict[str, list[str]]]
+    """Optional: server_name -> allowed tool names (whitelist)"""
+    excluded_tools: NotRequired[dict[str, list[str]]]
+    """Optional: server_name -> excluded tool names (blacklist)"""
 
 
 @dataclass
@@ -245,7 +249,14 @@ async def get_system_prompt(self, run_context: RunContextWrapper[TContext]) -> s
     async def get_mcp_tools(self) -> 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)
+        allowed_tools_map = self.mcp_config.get("allowed_tools", {})
+        excluded_tools_map = self.mcp_config.get("excluded_tools", {})
+        return await MCPUtil.get_all_function_tools(
+            self.mcp_servers,
+            convert_schemas_to_strict,
+            allowed_tools_map,
+            excluded_tools_map,
+        )
 
     async def get_all_tools(self, run_context: RunContextWrapper[Any]) -> list[Tool]:
         """All agent tools, including MCP tools and function tools."""

src/agents/mcp/server.py

@@ -57,7 +57,13 @@ async def call_tool(self, tool_name: str, arguments: dict[str, Any] | None) -> C
 class _MCPServerWithClientSession(MCPServer, abc.ABC):
     """Base class for MCP servers that use a `ClientSession` to communicate with the server."""
 
-    def __init__(self, cache_tools_list: bool, client_session_timeout_seconds: float | None):
+    def __init__(
+        self,
+        cache_tools_list: bool,
+        client_session_timeout_seconds: float | None,
+        allowed_tools: list[str] | None = None,
+        excluded_tools: list[str] | None = None,
+    ):
         """
         Args:
             cache_tools_list: Whether to cache the tools list. If `True`, the tools list will be
@@ -68,6 +74,10 @@ def __init__(self, cache_tools_list: bool, client_session_timeout_seconds: float
             (by avoiding a round-trip to the server every time).
 
             client_session_timeout_seconds: the read timeout passed to the MCP ClientSession.
+            allowed_tools: Optional list of tool names to allow (whitelist).
+                If set, only these tools will be available.
+            excluded_tools: Optional list of tool names to exclude (blacklist).
+                If set, these tools will be filtered out.
         """
         self.session: ClientSession | None = None
         self.exit_stack: AsyncExitStack = AsyncExitStack()
@@ -81,6 +91,9 @@ def __init__(self, cache_tools_list: bool, client_session_timeout_seconds: float
         self._cache_dirty = True
         self._tools_list: list[MCPTool] | None = None
 
+        self.allowed_tools = allowed_tools
+        self.excluded_tools = excluded_tools
+
     @abc.abstractmethod
     def create_streams(
         self,
@@ -138,14 +151,21 @@ async def list_tools(self) -> list[MCPTool]:
 
         # Return from cache if caching is enabled, we have tools, and the cache is not dirty
         if self.cache_tools_list and not self._cache_dirty and self._tools_list:
-            return self._tools_list
-
-        # Reset the cache dirty to False
-        self._cache_dirty = False
-
-        # Fetch the tools from the server
-        self._tools_list = (await self.session.list_tools()).tools
-        return self._tools_list
+            tools = self._tools_list
+        else:
+            # Reset the cache dirty to False
+            self._cache_dirty = False
+            # Fetch the tools from the server
+            self._tools_list = (await self.session.list_tools()).tools
+            tools = self._tools_list
+
+        # Filter tools based on allowed and excluded tools
+        filtered_tools = tools
+        if self.allowed_tools is not None:
+            filtered_tools = [t for t in filtered_tools if t.name in self.allowed_tools]
+        if self.excluded_tools is not None:
+            filtered_tools = [t for t in filtered_tools if t.name not in self.excluded_tools]
+        return filtered_tools
 
     async def call_tool(self, tool_name: str, arguments: dict[str, Any] | None) -> CallToolResult:
         """Invoke a tool on the server."""
@@ -206,6 +226,8 @@ def __init__(
         cache_tools_list: bool = False,
         name: str | None = None,
         client_session_timeout_seconds: float | None = 5,
+        allowed_tools: list[str] | None = None,
+        excluded_tools: list[str] | None = None,
     ):
         """Create a new MCP server based on the stdio transport.
 
@@ -223,8 +245,15 @@ def __init__(
             name: A readable name for the server. If not provided, we'll create one from the
                 command.
             client_session_timeout_seconds: the read timeout passed to the MCP ClientSession.
+            allowed_tools: Optional list of tool names to allow (whitelist).
+            excluded_tools: Optional list of tool names to exclude (blacklist).
         """
-        super().__init__(cache_tools_list, client_session_timeout_seconds)
+        super().__init__(
+            cache_tools_list,
+            client_session_timeout_seconds,
+            allowed_tools,
+            excluded_tools,
+        )
 
         self.params = StdioServerParameters(
             command=params["command"],
@@ -283,6 +312,8 @@ def __init__(
         cache_tools_list: bool = False,
         name: str | None = None,
         client_session_timeout_seconds: float | None = 5,
+        allowed_tools: list[str] | None = None,
+        excluded_tools: list[str] | None = None,
     ):
         """Create a new MCP server based on the HTTP with SSE transport.
 
@@ -302,8 +333,15 @@ def __init__(
                 URL.
 
             client_session_timeout_seconds: the read timeout passed to the MCP ClientSession.
+            allowed_tools: Optional list of tool names to allow (whitelist).
+            excluded_tools: Optional list of tool names to exclude (blacklist).
         """
-        super().__init__(cache_tools_list, client_session_timeout_seconds)
+        super().__init__(
+            cache_tools_list,
+            client_session_timeout_seconds,
+            allowed_tools,
+            excluded_tools,
+        )
 
         self.params = params
         self._name = name or f"sse: {self.params['url']}"
@@ -362,6 +400,8 @@ def __init__(
         cache_tools_list: bool = False,
         name: str | None = None,
         client_session_timeout_seconds: float | None = 5,
+        allowed_tools: list[str] | None = None,
+        excluded_tools: list[str] | None = None,
     ):
         """Create a new MCP server based on the Streamable HTTP transport.
 
@@ -382,8 +422,15 @@ def __init__(
                 URL.
 
             client_session_timeout_seconds: the read timeout passed to the MCP ClientSession.
+            allowed_tools: Optional list of tool names to allow (whitelist).
+            excluded_tools: Optional list of tool names to exclude (blacklist).
         """
-        super().__init__(cache_tools_list, client_session_timeout_seconds)
+        super().__init__(
+            cache_tools_list,
+            client_session_timeout_seconds,
+            allowed_tools,
+            excluded_tools,
+        )
 
         self.params = params
         self._name = name or f"streamable_http: {self.params['url']}"

src/agents/mcp/util.py

@@ -1,6 +1,6 @@
 import functools
 import json
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Optional
 
 from agents.strict_schema import ensure_strict_json_schema
 
@@ -22,13 +22,23 @@ 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,
+        allowed_tools_map: Optional[dict[str, list[str]]] = None,
+        excluded_tools_map: Optional[dict[str, list[str]]] = None,
     ) -> list[Tool]:
         """Get all function tools from a list of MCP servers."""
         tools = []
         tool_names: set[str] = set()
+        allowed_tools_map = allowed_tools_map or {}
+        excluded_tools_map = excluded_tools_map or {}
         for server in servers:
-            server_tools = await cls.get_function_tools(server, convert_schemas_to_strict)
+            allowed = allowed_tools_map.get(server.name)
+            excluded = excluded_tools_map.get(server.name)
+            server_tools = await cls.get_function_tools(
+                server, convert_schemas_to_strict, allowed, excluded
+            )
             server_tool_names = {tool.name for tool in server_tools}
             if len(server_tool_names & tool_names) > 0:
                 raise UserError(
@@ -42,15 +52,29 @@ 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,
+        allowed_tools: Optional[list[str]] = None,
+        excluded_tools: Optional[list[str]] = 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()
             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]
+        # Apply Agent-level filtering (additional filtering on top of server-level filtering)
+        filtered_tools = tools
+        if allowed_tools is not None:
+            filtered_tools = [t for t in filtered_tools if t.name in allowed_tools]
+        if excluded_tools is not None:
+            filtered_tools = [t for t in filtered_tools if t.name not in excluded_tools]
+
+        return [
+            cls.to_function_tool(tool, server, convert_schemas_to_strict)
+            for tool in filtered_tools
+        ]
 
     @classmethod
     def to_function_tool(