feat: tag resolved MCP tools with their AMP slug

lib/crewai/src/crewai/mcp/tool_resolver.py

@@ -152,6 +152,8 @@ class MCPToolResolver:
 
             try:
                 tools, clients = self._resolve_native(mcp_server_config)
+                for tool in tools:
+                    tool._amp_slug = slug
                 resolved_cache[slug] = (tools, clients)
                 all_clients.extend(clients)
             except Exception as e:

lib/crewai/src/crewai/tools/mcp_native_tool.py

@@ -59,6 +59,11 @@ class MCPNativeTool(BaseTool):
         self._client_factory = client_factory
         self._original_tool_name = original_tool_name or tool_name
         self._server_name = server_name
+        # Set by MCPToolResolver._resolve_amp when this tool is produced for
+        # an AMP slug; remains None for direct config / external URL refs.
+        # Consumed downstream by enterprise tooling to recover the canonical
+        # tool_id (e.g. "crewai_oauth:<slug>|mcp").
+        self._amp_slug: str | None = None
 
     @property
     def original_tool_name(self) -> str:

lib/crewai/src/crewai/tools/mcp_tool_wrapper.py

@@ -54,6 +54,11 @@ class MCPToolWrapper(BaseTool):
         self._mcp_server_params = mcp_server_params
         self._original_tool_name = tool_name
         self._server_name = server_name
+        # Set by MCPToolResolver._resolve_amp when this wrapper is produced for
+        # an AMP slug; remains None for direct config / external URL refs.
+        # Consumed downstream by enterprise tooling to recover the canonical
+        # tool_id (e.g. "crewai_oauth:<slug>|mcp").
+        self._amp_slug: str | None = None
 
     @property
     def mcp_server_params(self) -> dict[str, Any]:

lib/crewai/tests/mcp/test_tool_resolver_amp_slug.py

@@ -0,0 +1,135 @@
+"""Tests for the _amp_slug attribute set by MCPToolResolver._resolve_amp.
+
+The slug is private metadata used downstream by enterprise tooling to recover
+the canonical tool_id (e.g. ``crewai_oauth:<slug>|mcp``) for ACP rule
+evaluation.
+"""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from crewai.agent.core import Agent
+from crewai.mcp.config import MCPServerHTTP
+from crewai.mcp.tool_resolver import MCPToolResolver
+from crewai.tools.mcp_native_tool import MCPNativeTool
+from crewai.tools.mcp_tool_wrapper import MCPToolWrapper
+
+
+@pytest.fixture
+def agent():
+    return Agent(
+        role="Test Agent",
+        goal="Test goal",
+        backstory="Test backstory",
+    )
+
+
+@pytest.fixture
+def resolver(agent):
+    return MCPToolResolver(agent=agent, logger=agent._logger)
+
+
+class TestAmpSlugDefaultsNone:
+    def test_native_tool_default_amp_slug_is_none(self):
+        tool = MCPNativeTool(
+            client_factory=lambda: None,
+            tool_name="search",
+            tool_schema={"description": "Search"},
+            server_name="notion",
+        )
+        assert tool._amp_slug is None
+
+    def test_wrapper_tool_default_amp_slug_is_none(self):
+        tool = MCPToolWrapper(
+            mcp_server_params={"url": "https://mcp.example.com"},
+            tool_name="search",
+            tool_schema={"description": "Search"},
+            server_name="notion",
+        )
+        assert tool._amp_slug is None
+
+
+class TestAmpSlugSetByResolveAmp:
+    @patch("crewai.mcp.tool_resolver.MCPToolResolver._resolve_native")
+    @patch("crewai.mcp.tool_resolver.MCPToolResolver._fetch_amp_mcp_configs")
+    def test_resolve_amp_tags_each_tool_with_its_slug(
+        self, mock_fetch_configs, mock_resolve_native, resolver
+    ):
+        mock_fetch_configs.return_value = {
+            "notion": {"url": "https://mcp.crewai.com/notion"},
+            "github": {"url": "https://mcp.crewai.com/github"},
+        }
+
+        notion_tool = MCPNativeTool(
+            client_factory=lambda: None,
+            tool_name="search",
+            tool_schema={"description": "search Notion"},
+            server_name="notion",
+        )
+        github_tool = MCPNativeTool(
+            client_factory=lambda: None,
+            tool_name="list_repos",
+            tool_schema={"description": "list github repos"},
+            server_name="github",
+        )
+
+        def fake_resolve_native(config):
+            url = config.url if hasattr(config, "url") else config["url"]
+            if "notion" in url:
+                return ([notion_tool], [MagicMock()])
+            return ([github_tool], [MagicMock()])
+
+        mock_resolve_native.side_effect = fake_resolve_native
+
+        tools, _ = resolver._resolve_amp(
+            [("notion", None), ("github", None)]
+        )
+
+        assert {tool._amp_slug for tool in tools} == {"notion", "github"}
+
+    @patch("crewai.mcp.tool_resolver.MCPToolResolver._fetch_amp_mcp_configs")
+    def test_resolve_amp_does_not_tag_when_config_missing(
+        self, mock_fetch_configs, resolver
+    ):
+        mock_fetch_configs.return_value = {}
+
+        tools, _ = resolver._resolve_amp([("unknown", None)])
+
+        assert tools == []
+
+
+class TestAmpSlugUntaggedForOtherPaths:
+    @patch("crewai.mcp.tool_resolver.MCPClient")
+    def test_resolve_external_does_not_set_amp_slug(self, mock_client_class, resolver):
+        mock_client = AsyncMock()
+        mock_client.list_tools = AsyncMock(
+            return_value=[{"name": "search", "description": "Search"}]
+        )
+        mock_client.connected = False
+        mock_client.connect = AsyncMock()
+        mock_client.disconnect = AsyncMock()
+        mock_client_class.return_value = mock_client
+
+        with patch.object(
+            resolver, "_get_mcp_tool_schemas", return_value={"search": {"description": "Search"}}
+        ):
+            tools = resolver._resolve_external("https://mcp.example.com/api")
+
+        assert len(tools) == 1
+        assert tools[0]._amp_slug is None
+
+    @patch("crewai.mcp.tool_resolver.MCPClient")
+    def test_resolve_native_does_not_set_amp_slug(self, mock_client_class, resolver):
+        mock_client = AsyncMock()
+        mock_client.list_tools = AsyncMock(
+            return_value=[{"name": "search", "description": "Search"}]
+        )
+        mock_client.connected = False
+        mock_client.connect = AsyncMock()
+        mock_client.disconnect = AsyncMock()
+        mock_client_class.return_value = mock_client
+
+        config = MCPServerHTTP(url="https://mcp.example.com/api")
+        tools, _ = resolver._resolve_native(config)
+
+        assert all(tool._amp_slug is None for tool in tools)