From 3849eb6e9f318bdeee2c089c9e4e8b2c47d057b6 Mon Sep 17 00:00:00 2001 From: keaixiaozhu Date: Sat, 4 Jul 2026 07:54:54 +0800 Subject: [PATCH 1/3] feat(mcp): add SemanticToolFilter for embedding-based tool filtering Add SemanticToolFilter, a dynamic MCP tool filter that includes a tool only when its embedding (name + description) is semantically similar to a query derived from the requesting agent's role/goal/backstory, or from run_context[run_context_key] when populated. It implements the existing per-tool ToolFilter protocol (2-arg __call__), so it plugs into tool_resolver with no resolver changes. Reuses embedders from crewai.rag.embeddings (build_embedder) and computes cosine similarity in pure Python (no new dependency). Embeddings are cached per text. Fails open: when no query can be derived or the embedder raises, the tool is included so a misconfigured embedder never silently strips all tools. Also adds a create_semantic_tool_filter factory and exports both from crewai.mcp. --- lib/crewai/src/crewai/mcp/__init__.py | 4 + lib/crewai/src/crewai/mcp/filters.py | 177 +++++++++++++++++++++++++- 2 files changed, 179 insertions(+), 2 deletions(-) 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..9a8275c04f 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,172 @@ 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=False)) + 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) + except Exception: + return True + + return _cosine_similarity(query_vec, tool_vec) >= 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, + ) From 31f8876c7fcac4a000a18f533547748a64cb02bc Mon Sep 17 00:00:00 2001 From: keaixiaozhu Date: Sat, 4 Jul 2026 07:54:54 +0800 Subject: [PATCH 2/3] test(mcp): cover SemanticToolFilter Add tests for SemanticToolFilter: relevant-vs-irrelevant streaming by agent profile, fail-open when no query is available, fail-open when the embedder raises, run_context query taking precedence over the agent profile, threshold boundary, per-text embedding caching, the factory, and empty tool name/description failing open. Uses a deterministic bag-of-words embedder so no model or network is required. --- .../tests/mcp/test_semantic_tool_filter.py | 162 ++++++++++++++++++ 1 file changed, 162 insertions(+) create mode 100644 lib/crewai/tests/mcp/test_semantic_tool_filter.py 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..f57337b210 --- /dev/null +++ b/lib/crewai/tests/mcp/test_semantic_tool_filter.py @@ -0,0 +1,162 @@ +"""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_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 From 155e901f96d71690d87e2b55f96d3d887b6c8121 Mon Sep 17 00:00:00 2001 From: keaixiaozhu Date: Sat, 4 Jul 2026 16:04:13 +0800 Subject: [PATCH 3/3] fix(mcp): fail open on empty/mismatched embeddings in SemanticToolFilter Address CodeRabbit review: treat empty embedding vectors as a failure (return True, fail open) instead of letting _cosine_similarity return 0.0 and silently exclude the tool; move the cosine call inside the try/except so it is covered by the fail-open path; and use zip(strict=True) so mismatched embedding dimensions raise explicitly and are caught by the same fallback. Add tests for empty-vector and mismatched-length fail-open. --- lib/crewai/src/crewai/mcp/filters.py | 9 ++++-- .../tests/mcp/test_semantic_tool_filter.py | 32 +++++++++++++++++++ 2 files changed, 39 insertions(+), 2 deletions(-) diff --git a/lib/crewai/src/crewai/mcp/filters.py b/lib/crewai/src/crewai/mcp/filters.py index 9a8275c04f..549b58ba65 100644 --- a/lib/crewai/src/crewai/mcp/filters.py +++ b/lib/crewai/src/crewai/mcp/filters.py @@ -179,7 +179,7 @@ def _cosine_similarity(a: list[float], b: list[float]) -> float: """ if not a or not b: return 0.0 - dot = sum(x * y for x, y in zip(a, b, strict=False)) + 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: @@ -261,10 +261,15 @@ def __call__(self, context: ToolFilterContext, tool: dict[str, Any]) -> bool: 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 _cosine_similarity(query_vec, tool_vec) >= self.threshold + return similarity >= self.threshold def _query_text(self, context: ToolFilterContext) -> str | None: """Derive the query string from run_context or the agent profile.""" diff --git a/lib/crewai/tests/mcp/test_semantic_tool_filter.py b/lib/crewai/tests/mcp/test_semantic_tool_filter.py index f57337b210..b1aa22ccee 100644 --- a/lib/crewai/tests/mcp/test_semantic_tool_filter.py +++ b/lib/crewai/tests/mcp/test_semantic_tool_filter.py @@ -91,6 +91,38 @@ def test_fail_open_when_embedder_raises(self) -> 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()