Enhance MCP tool resolution and related events (#4580)

* feat: enhance MCP tool resolution

* feat: emit event when MCP configuration fails

* feat: emit event when MCP tool execution has failed

* style: resolve linter issues

* refactor: use clear and natural mcp tool name resolution

* test: fix broken tests

* fix: resolve MCP connection leaks, slug validation, duplicate connections, and httpx exception handling

---------

Co-authored-by: Greyson LaLonde <greyson.r.lalonde@gmail.com>
Co-authored-by: Greyson LaLonde <greyson@crewai.com>

Author: lucasgomide

lib/crewai/src/crewai/agent/core.py

@@ -4,7 +4,8 @@
 from collections.abc import Callable
 from copy import copy as shallow_copy
 from hashlib import md5
-from typing import Any, Literal
+import re
+from typing import Any, Final, Literal
 import uuid
 
 from pydantic import (
@@ -36,6 +37,11 @@
 from crewai.utilities.string_utils import interpolate_only
 
 
+_SLUG_RE: Final[re.Pattern[str]] = re.compile(
+    r"^(?:crewai-amp:)?[a-zA-Z0-9][a-zA-Z0-9_-]*(?:#\w+)?$"
+)
+
+
 PlatformApp = Literal[
     "asana",
     "box",
@@ -197,7 +203,7 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
     )
     mcps: list[str | MCPServerConfig] | None = Field(
         default=None,
-        description="List of MCP server references. Supports 'https://server.com/path' for external servers and 'crewai-amp:mcp-name' for AMP marketplace. Use '#tool_name' suffix for specific tools.",
+        description="List of MCP server references. Supports 'https://server.com/path' for external servers and bare slugs like 'notion' for connected MCP integrations. Use '#tool_name' suffix for specific tools.",
     )
     memory: Any = Field(
         default=None,
@@ -276,14 +282,16 @@ def validate_mcps(
         validated_mcps: list[str | MCPServerConfig] = []
         for mcp in mcps:
             if isinstance(mcp, str):
-                if mcp.startswith(("https://", "crewai-amp:")):
+                if mcp.startswith("https://"):
+                    validated_mcps.append(mcp)
+                elif _SLUG_RE.match(mcp):
                     validated_mcps.append(mcp)
                 else:
                     raise ValueError(
-                        f"Invalid MCP reference: {mcp}. "
-                        "String references must start with 'https://' or 'crewai-amp:'"
+                        f"Invalid MCP reference: {mcp!r}. "
+                        "String references must be an 'https://' URL or a valid "
+                        "slug (e.g. 'notion', 'notion#search', 'crewai-amp:notion')."
                     )
-
             elif isinstance(mcp, (MCPServerConfig)):
                 validated_mcps.append(mcp)
             else:

lib/crewai/src/crewai/agents/agent_builder/base_agent.py

@@ -190,6 +190,15 @@ def mark_trace_batch_as_failed(
             timeout=30,
         )
 
+    def get_mcp_configs(self, slugs: list[str]) -> httpx.Response:
+        """Get MCP server configurations for the given slugs."""
+        return self._make_request(
+            "GET",
+            f"{self.INTEGRATIONS_RESOURCE}/mcp_configs",
+            params={"slugs": ",".join(slugs)},
+            timeout=30,
+        )
+
     def get_triggers(self) -> httpx.Response:
         """Get all available triggers from integrations."""
         return self._make_request("GET", f"{self.INTEGRATIONS_RESOURCE}/apps")

lib/crewai/src/crewai/cli/plus_api.py

@@ -63,6 +63,7 @@
     AgentLogsStartedEvent,
 )
 from crewai.events.types.mcp_events import (
+    MCPConfigFetchFailedEvent,
     MCPConnectionCompletedEvent,
     MCPConnectionFailedEvent,
     MCPConnectionStartedEvent,
@@ -165,6 +166,7 @@
     "LiteAgentExecutionCompletedEvent",
     "LiteAgentExecutionErrorEvent",
     "LiteAgentExecutionStartedEvent",
+    "MCPConfigFetchFailedEvent",
     "MCPConnectionCompletedEvent",
     "MCPConnectionFailedEvent",
     "MCPConnectionStartedEvent",

lib/crewai/src/crewai/events/__init__.py

@@ -68,6 +68,7 @@
     AgentLogsStartedEvent,
 )
 from crewai.events.types.mcp_events import (
+    MCPConfigFetchFailedEvent,
     MCPConnectionCompletedEvent,
     MCPConnectionFailedEvent,
     MCPConnectionStartedEvent,
@@ -665,6 +666,16 @@ def on_mcp_connection_failed(_: Any, event: MCPConnectionFailedEvent) -> None:
                 event.error_type,
             )
 
+        @crewai_event_bus.on(MCPConfigFetchFailedEvent)
+        def on_mcp_config_fetch_failed(
+            _: Any, event: MCPConfigFetchFailedEvent
+        ) -> None:
+            self.formatter.handle_mcp_config_fetch_failed(
+                event.slug,
+                event.error,
+                event.error_type,
+            )
+
         @crewai_event_bus.on(MCPToolExecutionStartedEvent)
         def on_mcp_tool_execution_started(
             _: Any, event: MCPToolExecutionStartedEvent

lib/crewai/src/crewai/events/event_listener.py

@@ -67,6 +67,7 @@
     LLMGuardrailStartedEvent,
 )
 from crewai.events.types.mcp_events import (
+    MCPConfigFetchFailedEvent,
     MCPConnectionCompletedEvent,
     MCPConnectionFailedEvent,
     MCPConnectionStartedEvent,
@@ -181,4 +182,5 @@
     | MCPToolExecutionStartedEvent
     | MCPToolExecutionCompletedEvent
     | MCPToolExecutionFailedEvent
+    | MCPConfigFetchFailedEvent
 )

lib/crewai/src/crewai/events/event_types.py

@@ -83,3 +83,16 @@ class MCPToolExecutionFailedEvent(MCPEvent):
     error_type: str | None = None  # "timeout", "validation", "server_error", etc.
     started_at: datetime | None = None
     failed_at: datetime | None = None
+
+
+class MCPConfigFetchFailedEvent(BaseEvent):
+    """Event emitted when fetching an AMP MCP server config fails.
+
+    This covers cases where the slug is not connected, the API call
+    failed, or native MCP resolution failed after config was fetched.
+    """
+
+    type: str = "mcp_config_fetch_failed"
+    slug: str
+    error: str
+    error_type: str | None = None  # "not_connected", "api_error", "connection_failed"

lib/crewai/src/crewai/events/types/mcp_events.py

@@ -1512,6 +1512,34 @@ def handle_mcp_connection_failed(
         self.print(panel)
         self.print()
 
+    def handle_mcp_config_fetch_failed(
+        self,
+        slug: str,
+        error: str = "",
+        error_type: str | None = None,
+    ) -> None:
+        """Handle MCP config fetch failed event (AMP resolution failures)."""
+        if not self.verbose:
+            return
+
+        content = Text()
+        content.append("MCP Config Fetch Failed\n\n", style="red bold")
+        content.append("Server: ", style="white")
+        content.append(f"{slug}\n", style="red")
+
+        if error_type:
+            content.append("Error Type: ", style="white")
+            content.append(f"{error_type}\n", style="red")
+
+        if error:
+            content.append("\nError: ", style="white bold")
+            error_preview = error[:500] + "..." if len(error) > 500 else error
+            content.append(f"{error_preview}\n", style="red")
+
+        panel = self.create_panel(content, "❌ MCP Config Failed", "red")
+        self.print(panel)
+        self.print()
+
     def handle_mcp_tool_execution_started(
         self,
         server_name: str,

lib/crewai/src/crewai/events/utils/console_formatter.py

@@ -18,6 +18,7 @@
     create_dynamic_tool_filter,
     create_static_tool_filter,
 )
+from crewai.mcp.tool_resolver import MCPToolResolver
 from crewai.mcp.transports.base import BaseTransport, TransportType
 
 
@@ -28,6 +29,7 @@
     "MCPServerHTTP",
     "MCPServerSSE",
     "MCPServerStdio",
+    "MCPToolResolver",
     "StaticToolFilter",
     "ToolFilter",
     "ToolFilterContext",

lib/crewai/src/crewai/mcp/__init__.py

@@ -6,7 +6,7 @@
 from datetime import datetime
 import logging
 import time
-from typing import Any
+from typing import Any, NamedTuple
 
 from typing_extensions import Self
 
@@ -34,6 +34,13 @@
 from crewai.utilities.string_utils import sanitize_tool_name
 
 
+class _MCPToolResult(NamedTuple):
+    """Internal result from an MCP tool call, carrying the ``isError`` flag."""
+
+    content: str
+    is_error: bool
+
+
 # MCP Connection timeout constants (in seconds)
 MCP_CONNECTION_TIMEOUT = 30  # Increased for slow servers
 MCP_TOOL_EXECUTION_TIMEOUT = 30
@@ -420,6 +427,7 @@ async def _list_tools_impl(self) -> list[dict[str, Any]]:
         return [
             {
                 "name": sanitize_tool_name(tool.name),
+                "original_name": tool.name,
                 "description": getattr(tool, "description", ""),
                 "inputSchema": getattr(tool, "inputSchema", {}),
             }
@@ -461,29 +469,46 @@ async def call_tool(
         )
 
         try:
-            result = await self._retry_operation(
+            tool_result: _MCPToolResult = await self._retry_operation(
                 lambda: self._call_tool_impl(tool_name, cleaned_arguments),
                 timeout=self.execution_timeout,
             )
 
-            completed_at = datetime.now()
-            execution_duration_ms = (completed_at - started_at).total_seconds() * 1000
-            crewai_event_bus.emit(
-                self,
-                MCPToolExecutionCompletedEvent(
-                    server_name=server_name,
-                    server_url=server_url,
-                    transport_type=transport_type,
-                    tool_name=tool_name,
-                    tool_args=cleaned_arguments,
-                    result=result,
-                    started_at=started_at,
-                    completed_at=completed_at,
-                    execution_duration_ms=execution_duration_ms,
-                ),
-            )
+            finished_at = datetime.now()
+            execution_duration_ms = (finished_at - started_at).total_seconds() * 1000
+
+            if tool_result.is_error:
+                crewai_event_bus.emit(
+                    self,
+                    MCPToolExecutionFailedEvent(
+                        server_name=server_name,
+                        server_url=server_url,
+                        transport_type=transport_type,
+                        tool_name=tool_name,
+                        tool_args=cleaned_arguments,
+                        error=tool_result.content,
+                        error_type="tool_error",
+                        started_at=started_at,
+                        failed_at=finished_at,
+                    ),
+                )
+            else:
+                crewai_event_bus.emit(
+                    self,
+                    MCPToolExecutionCompletedEvent(
+                        server_name=server_name,
+                        server_url=server_url,
+                        transport_type=transport_type,
+                        tool_name=tool_name,
+                        tool_args=cleaned_arguments,
+                        result=tool_result.content,
+                        started_at=started_at,
+                        completed_at=finished_at,
+                        execution_duration_ms=execution_duration_ms,
+                    ),
+                )
 
-            return result
+            return tool_result.content
         except Exception as e:
             failed_at = datetime.now()
             error_type = (
@@ -564,23 +589,27 @@ def _clean_tool_arguments(self, arguments: dict[str, Any]) -> dict[str, Any]:
 
         return cleaned
 
-    async def _call_tool_impl(self, tool_name: str, arguments: dict[str, Any]) -> Any:
+    async def _call_tool_impl(
+        self, tool_name: str, arguments: dict[str, Any]
+    ) -> _MCPToolResult:
         """Internal implementation of call_tool."""
         result = await asyncio.wait_for(
             self.session.call_tool(tool_name, arguments),
             timeout=self.execution_timeout,
         )
 
+        is_error = getattr(result, "isError", False) or False
+
         # Extract result content
         if hasattr(result, "content") and result.content:
             if isinstance(result.content, list) and len(result.content) > 0:
                 content_item = result.content[0]
                 if hasattr(content_item, "text"):
-                    return str(content_item.text)
-                return str(content_item)
-            return str(result.content)
+                    return _MCPToolResult(str(content_item.text), is_error)
+                return _MCPToolResult(str(content_item), is_error)
+            return _MCPToolResult(str(result.content), is_error)
 
-        return str(result)
+        return _MCPToolResult(str(result), is_error)
 
     async def list_prompts(self) -> list[dict[str, Any]]:
         """List available prompts from MCP server.