diff --git a/lib/crewai/src/crewai/mcp/__init__.py b/lib/crewai/src/crewai/mcp/__init__.py index ee057af14b..cc3f83d0b7 100644 --- a/lib/crewai/src/crewai/mcp/__init__.py +++ b/lib/crewai/src/crewai/mcp/__init__.py @@ -20,10 +20,12 @@ MCPServerStdio, ) from crewai.mcp.filters import ( + SemanticToolFilter, StaticToolFilter, ToolFilter, ToolFilterContext, create_dynamic_tool_filter, + create_semantic_tool_filter, create_static_tool_filter, ) @@ -59,10 +61,12 @@ def __getattr__(name: str) -> Any: "MCPServerSSE", "MCPServerStdio", "MCPToolResolver", + "SemanticToolFilter", "StaticToolFilter", "ToolFilter", "ToolFilterContext", "TransportType", "create_dynamic_tool_filter", + "create_semantic_tool_filter", "create_static_tool_filter", ] diff --git a/lib/crewai/src/crewai/mcp/filters.py b/lib/crewai/src/crewai/mcp/filters.py index ec8bb2e171..549b58ba65 100644 --- a/lib/crewai/src/crewai/mcp/filters.py +++ b/lib/crewai/src/crewai/mcp/filters.py @@ -1,17 +1,21 @@ """Tool filtering support for MCP servers. This module provides utilities for filtering tools from MCP servers, -including static allow/block lists and dynamic context-aware filtering. +including static allow/block lists, dynamic context-aware filtering, and +semantic (embedding-based) filtering. """ +from __future__ import annotations + from collections.abc import Callable +import math from typing import TYPE_CHECKING, Any from pydantic import BaseModel, Field if TYPE_CHECKING: - pass + from crewai.rag.core.base_embeddings_callable import EmbeddingFunction class ToolFilterContext(BaseModel): @@ -161,3 +165,177 @@ async def context_aware_filter( ``` """ return filter_func + + +def _cosine_similarity(a: list[float], b: list[float]) -> float: + """Cosine similarity between two vectors (pure Python, no numpy). + + Args: + a: First vector. + b: Second vector. + + Returns: + Cosine similarity in [-1, 1]; 0.0 if either vector is empty or zero. + """ + if not a or not b: + return 0.0 + dot = sum(x * y for x, y in zip(a, b, strict=True)) + na = math.sqrt(sum(x * x for x in a)) + nb = math.sqrt(sum(y * y for y in b)) + if na == 0.0 or nb == 0.0: + return 0.0 + return dot / (na * nb) + + +class SemanticToolFilter: + """Dynamic MCP tool filter that includes tools semantically relevant to the agent. + + Embeds a query derived from the requesting agent's role/goal/backstory + (or from ``run_context[run_context_key]`` when populated) and each tool's + ``name`` + ``description``, including the tool only when cosine similarity + between the two embeddings is at least ``threshold``. + + Fits the existing per-tool ``ToolFilter`` protocol and is applied to MCP + server tools at resolution time (see ``tool_resolver``). It is + threshold-based (not top-K) because the protocol decides each tool + independently. + + Fails open: if no query can be derived, or the embedder raises, the tool is + included — a misconfigured embedder never silently strips all tools. + + Example: + ```python + from crewai.rag.embeddings.factory import build_embedder + from crewai.mcp.filters import SemanticToolFilter + + embedder = build_embedder({"provider": "openai", "config": {...}}) + mcp_server = MCPServerStdio( + command="npx", + args=["-y", "@modelcontextprotocol/server-filesystem"], + tool_filter=SemanticToolFilter(embedder=embedder, threshold=0.35), + ) + ``` + """ + + def __init__( + self, + embedder: EmbeddingFunction, + threshold: float = 0.3, + run_context_key: str = "query", + ) -> None: + """Initialize semantic tool filter. + + Args: + embedder: Embedding function used to embed the query and tool texts. + Build one with ``crewai.rag.embeddings.factory.build_embedder``. + threshold: Minimum cosine similarity (0.0-1.0) for a tool to be + included. + run_context_key: Key in ``ToolFilterContext.run_context`` holding an + explicit query string. When absent, the agent's + role/goal/backstory is used as the query. + """ + self._embedder = embedder + self.threshold = threshold + self._run_context_key = run_context_key + self._cache: dict[str, list[float]] = {} + + def __call__(self, context: ToolFilterContext, tool: dict[str, Any]) -> bool: + """Include ``tool`` if it is semantically relevant to the query. + + Args: + context: Tool filter context (agent, server, run context). + tool: Tool definition dict with at least ``name``/``description``. + + Returns: + True if the tool should be included, False otherwise. Fails open + (returns True) when no query is available or embedding fails. + """ + query = self._query_text(context) + if not query: + return True + + tool_text = self._tool_text(tool) + if not tool_text: + return True + + try: + query_vec = self._embed(query) + tool_vec = self._embed(tool_text) + if not query_vec or not tool_vec: + # Empty embedding (e.g. embedder returned nothing useful) + # is treated as a failure and fails open. + return True + similarity = _cosine_similarity(query_vec, tool_vec) + except Exception: + return True + + return similarity >= self.threshold + + def _query_text(self, context: ToolFilterContext) -> str | None: + """Derive the query string from run_context or the agent profile.""" + run_context = context.run_context or {} + query = run_context.get(self._run_context_key) + if isinstance(query, str) and query.strip(): + return query + + agent = context.agent + if agent is None: + return None + + parts = [ + getattr(agent, "role", ""), + getattr(agent, "goal", ""), + getattr(agent, "backstory", ""), + ] + parts = [p for p in parts if isinstance(p, str) and p.strip()] + return " ".join(parts) if parts else None + + @staticmethod + def _tool_text(tool: dict[str, Any]) -> str: + """Build the text representation of a tool for embedding.""" + name = tool.get("name", "") or "" + description = tool.get("description", "") or "" + if not isinstance(name, str): + name = str(name) + if not isinstance(description, str): + description = str(description) + name = name.strip() + description = description.strip() + if not name and not description: + return "" + if name and description: + return f"{name}: {description}" + return name or description + + def _embed(self, text: str) -> list[float]: + """Embed ``text`` using the embedder, with per-text caching.""" + if text in self._cache: + return self._cache[text] + result = self._embedder(input=[text]) + vec = list(result[0]) if result else [] + self._cache[text] = vec + return vec + + +def create_semantic_tool_filter( + embedder: EmbeddingFunction, + threshold: float = 0.3, + run_context_key: str = "query", +) -> SemanticToolFilter: + """Create a semantic tool filter. + + Convenience wrapper around :class:`SemanticToolFilter`. + + Args: + embedder: Embedding function (build one with ``build_embedder``). + threshold: Minimum cosine similarity for inclusion. + run_context_key: Optional run_context key holding an explicit query. + + Returns: + A ``SemanticToolFilter`` usable as ``tool_filter`` on an MCP server. + """ + return SemanticToolFilter( + embedder=embedder, + threshold=threshold, + run_context_key=run_context_key, + ) diff --git a/lib/crewai/tests/mcp/test_semantic_tool_filter.py b/lib/crewai/tests/mcp/test_semantic_tool_filter.py new file mode 100644 index 0000000000..b1aa22ccee --- /dev/null +++ b/lib/crewai/tests/mcp/test_semantic_tool_filter.py @@ -0,0 +1,194 @@ +"""Tests for the SemanticToolFilter (embedding-based MCP tool filtering).""" + +from __future__ import annotations + +import re +from typing import Any +from unittest.mock import MagicMock + +import pytest + +from crewai.mcp import ( + SemanticToolFilter, + ToolFilterContext, + create_semantic_tool_filter, +) + + +VOCAB = ["search", "web", "file", "read", "code", "execute", "database", "query"] + + +def _tokens(text: str) -> set[str]: + return set(re.findall(r"\w+", text.lower())) + + +class BagOfWordsEmbedder: + """Deterministic multi-hot embedder over a fixed vocabulary. + + Cosine similarity of two vectors equals the fraction of shared vocab words + scaled by each vector's magnitude — realistic enough to exercise the filter + without any model or network. Tracks call count to verify caching. + """ + + def __init__(self) -> None: + self.calls = 0 + + def __call__(self, input: list[str]) -> list[list[float]]: + self.calls += 1 + return [[1.0 if word in _tokens(text) else 0.0 for word in VOCAB] for text in input] + + +class RaisingEmbedder: + """Embedder that always raises, to exercise fail-open behavior.""" + + def __call__(self, input: list[str]) -> list[list[float]]: + raise RuntimeError("embedder unavailable") + + +def _agent(role: str = "", goal: str = "", backstory: str = "") -> Any: + agent = MagicMock() + agent.role = role + agent.goal = goal + agent.backstory = backstory + return agent + + +class TestSemanticToolFilter: + """Tests for SemanticToolFilter.""" + + def test_includes_relevant_excludes_irrelevant(self) -> None: + """Tools semantically close to the agent profile are included.""" + embedder = BagOfWordsEmbedder() + flt = SemanticToolFilter(embedder=embedder, threshold=0.3) + ctx = ToolFilterContext( + agent=_agent(role="Researcher", goal="search the web"), + server_name="srv", + run_context=None, + ) + + web_search = {"name": "web_search", "description": "search the web"} + read_file = {"name": "read_file", "description": "read a file from disk"} + + assert flt(ctx, web_search) is True + assert flt(ctx, read_file) is False + + def test_fail_open_when_no_query(self) -> None: + """With no agent and no run_context query, all tools are included.""" + embedder = BagOfWordsEmbedder() + flt = SemanticToolFilter(embedder=embedder, threshold=0.3) + ctx = ToolFilterContext(agent=None, server_name="srv", run_context=None) + + assert flt(ctx, {"name": "any", "description": "anything"}) is True + assert embedder.calls == 0 + + def test_fail_open_when_embedder_raises(self) -> None: + """Embedder failures never strip tools.""" + flt = SemanticToolFilter(embedder=RaisingEmbedder(), threshold=0.3) + ctx = ToolFilterContext( + agent=_agent(role="Researcher", goal="search the web"), + server_name="srv", + run_context=None, + ) + assert flt(ctx, {"name": "web_search", "description": "search the web"}) is True + + def test_fail_open_when_embedder_returns_empty_vector(self) -> None: + """An embedder returning an empty vector fails open (tool included).""" + + class EmptyEmbedder: + def __call__(self, input: list[str]) -> list[list[float]]: + return [[] for _ in input] + + flt = SemanticToolFilter(embedder=EmptyEmbedder(), threshold=0.3) + ctx = ToolFilterContext( + agent=_agent(role="Researcher", goal="search the web"), + server_name="srv", + run_context=None, + ) + assert flt(ctx, {"name": "web_search", "description": "search the web"}) is True + + def test_fail_open_when_embeddings_differ_in_length(self) -> None: + """Mismatched embedding lengths fail open (strict zip raises, caught).""" + + class MismatchedEmbedder: + def __call__(self, input: list[str]) -> list[list[float]]: + # Query (agent profile) contains "Researcher"; tool text does not. + # Yield different lengths so the strict zip raises. + return [[1.0, 0.0, 0.0] if "Researcher" in t else [1.0, 0.0] for t in input] + + flt = SemanticToolFilter(embedder=MismatchedEmbedder(), threshold=0.3) + ctx = ToolFilterContext( + agent=_agent(role="Researcher", goal="search the web"), + server_name="srv", + run_context=None, + ) + assert flt(ctx, {"name": "web_search", "description": "search the web"}) is True + + def test_run_context_query_takes_precedence(self) -> None: + """An explicit run_context query overrides the agent profile.""" + embedder = BagOfWordsEmbedder() + flt = SemanticToolFilter(embedder=embedder, threshold=0.3) + ctx = ToolFilterContext( + agent=_agent(role="Coder", goal="write code"), + server_name="srv", + run_context={"query": "search the web"}, + ) + # Relevant to the run_context query, not the coder profile. + assert flt(ctx, {"name": "web_search", "description": "search the web"}) is True + # Relevant to the coder profile but not the query. + assert flt(ctx, {"name": "run_code", "description": "execute code"}) is False + + def test_threshold_boundary(self) -> None: + """threshold controls the include/exclude cutoff.""" + # query "search the web" -> {search, web} (magnitude sqrt(2)) + # tool "search" -> {search} -> cosine = 1 / (sqrt(2) * 1) ~= 0.707 + embedder = BagOfWordsEmbedder() + ctx = ToolFilterContext( + agent=_agent(role="R", goal="search the web"), + server_name="srv", + run_context=None, + ) + flt_low = SemanticToolFilter(embedder=BagOfWordsEmbedder(), threshold=0.5) + flt_high = SemanticToolFilter(embedder=BagOfWordsEmbedder(), threshold=0.8) + tool = {"name": "search_only", "description": "search"} + + assert flt_low(ctx, tool) is True # 0.707 >= 0.5 + assert flt_high(ctx, tool) is False # 0.707 < 0.8 + + def test_embeddings_are_cached(self) -> None: + """The query is embedded once even when filtering many tools.""" + embedder = BagOfWordsEmbedder() + flt = SemanticToolFilter(embedder=embedder, threshold=0.3) + ctx = ToolFilterContext( + agent=_agent(role="Researcher", goal="search the web"), + server_name="srv", + run_context=None, + ) + flt(ctx, {"name": "web_search", "description": "search the web"}) + flt(ctx, {"name": "read_file", "description": "read a file"}) + + # 1 query + 2 distinct tool texts = 3 embedder calls (no re-embed of query). + assert embedder.calls == 3 + + def test_factory_returns_working_filter(self) -> None: + """create_semantic_tool_filter returns a usable SemanticToolFilter.""" + flt = create_semantic_tool_filter( + embedder=BagOfWordsEmbedder(), threshold=0.3 + ) + assert isinstance(flt, SemanticToolFilter) + ctx = ToolFilterContext( + agent=_agent(role="Researcher", goal="search the web"), + server_name="srv", + run_context=None, + ) + assert flt(ctx, {"name": "web_search", "description": "search the web"}) is True + assert flt(ctx, {"name": "read_file", "description": "read a file"}) is False + + def test_empty_tool_description_fail_open(self) -> None: + """A tool with no name/description is included (nothing to embed).""" + flt = SemanticToolFilter(embedder=BagOfWordsEmbedder(), threshold=0.3) + ctx = ToolFilterContext( + agent=_agent(role="Researcher", goal="search the web"), + server_name="srv", + run_context=None, + ) + assert flt(ctx, {"name": "", "description": ""}) is True