-
Notifications
You must be signed in to change notification settings - Fork 40
Description
Overview
I'd like to propose adding tool filtering capability to MCPAdapt to enable restricting which tools an agent can access by name.
Use Case
In multi-agent systems (e.g., CrewAI), we often want different agents to have access to different subsets of tools based on their responsibilities. Currently, mcpadapt provides all tools to all agents, which can lead to:
- Security concerns (agents having access to tools they shouldn't use)
- Cluttered agent prompts with irrelevant tool descriptions
- Lack of precise access control at the agent level
Proposed Implementation
After reviewing the code in mcpadapt/src/mcpadapt/core.py
and crewai_adapter.py
, I propose adding the filtering functionality directly to the MCPAdapt
class:
class MCPAdapt:
def __init__(
self,
serverparams: StdioServerParameters
| dict[str, Any]
| list[StdioServerParameters | dict[str, Any]],
adapter: ToolAdapter,
allowed_tools: Optional[List[str]] = None, # New parameter
connect_timeout: int = 30,
):
"""
Manage the MCP server / client lifecycle and expose tools adapted with the adapter.
Args:
serverparams: MCP server parameters (stdio or sse). Can be a list for multiple MCPs.
adapter: Adapter to convert MCP tools call into agentic framework tools.
allowed_tools: Optional list of tool names to allow (all tools if None).
connect_timeout: Connection timeout in seconds to the mcp server.
"""
# Existing initialization code
self.allowed_tools = allowed_tools
# ...
def _filter_tools(self, tools: list[mcp.types.Tool]) -> list[mcp.types.Tool]:
"""Filter tools based on the allowed list."""
if not self.allowed_tools:
# No filter specified, return all tools
return tools
# Return only tools that match the allowed list
filtered = [tool for tool in tools if tool.name in self.allowed_tools]
print(f"Filtered {len(tools)} tools down to {len(filtered)}")
return filtered
def tools(self) -> list[Any]:
"""Returns the filtered tools adapted to the desired Agent framework."""
if not self.sessions:
raise RuntimeError("Session not initialized")
def _sync_call_tool(
session, name: str, arguments: dict | None = None
) -> mcp.types.CallToolResult:
return asyncio.run_coroutine_threadsafe(
session.call_tool(name, arguments), self.loop
).result()
# Apply filtering before adaptation
all_tools = []
for session, tool_list in zip(self.sessions, self.mcp_tools):
filtered_tools = self._filter_tools(tool_list)
for tool in filtered_tools:
all_tools.append(self.adapter.adapt(partial(_sync_call_tool, session, tool.name), tool))
return all_tools
Similarly, the atools
method would be updated to apply filtering.
Integration Example for CrewAI
The change would be backward compatible and allow for simple usage:
# Current usage (unchanged, gets all tools)
with MCPAdapt(server_params, CrewAIAdapter()) as tools:
agent = Agent(..., tools=tools)
# New usage with filtering
with MCPAdapt(
server_params,
CrewAIAdapter(),
allowed_tools=["tool1", "tool2"]
) as tools:
agent = Agent(..., tools=tools)
For crewai-tools Package
To utilize this in crewai-tools' MCPServerAdapter:
class MCPServerAdapter:
"""Manages the lifecycle of an MCP server and make its tools available to CrewAI."""
def __init__(
self,
serverparams: StdioServerParameters | dict[str, Any],
allowed_tools: Optional[List[str]] = None,
):
"""Initialize the MCP Server
Args:
serverparams: The parameters for the MCP server
allowed_tools: Optional list of tool names to allow
"""
self._adapter = MCPAdapt(
serverparams,
CrewAIAdapter(),
allowed_tools=allowed_tools
)
# ...
Benefits
- Minimal Changes: Builds on the existing architecture without changing the core interfaces
- Backward Compatibility: All existing code continues to work
- Security Improvement: Agents only see tools they're authorized to use
- Clean API: Simple parameter addition without complexity
Would you be interested in this feature addition? It would significantly enhance the security and usability of mcpadapt in multi-agent systems.