diff --git a/letta/services/mcp/base_client.py b/letta/services/mcp/base_client.py
index 31b25e29..c53b6f99 100644
--- a/letta/services/mcp/base_client.py
+++ b/letta/services/mcp/base_client.py
@@ -11,6 +11,31 @@ from letta.log import get_logger
 
 logger = get_logger(__name__)
 
+EXPECTED_MCP_TOOL_ERRORS = (
+    "McpError",
+    "ToolError",
+    "HTTPStatusError",
+    "ConnectError",
+    "ConnectTimeout",
+    "ReadTimeout",
+    "ReadError",
+    "RemoteProtocolError",
+    "LocalProtocolError",
+    "ConnectionError",
+    "SSLError",
+    "MaxRetryError",
+    "ProtocolError",
+    "BrokenResourceError",
+)
+
+
+def _log_mcp_tool_error(log: "get_logger", tool_name: str, exc: Exception) -> None:
+    exc_name = type(exc).__name__
+    if exc_name in EXPECTED_MCP_TOOL_ERRORS:
+        log.info(f"MCP tool '{tool_name}' execution failed ({exc_name}): {exc}")
+    else:
+        log.warning(f"MCP tool '{tool_name}' execution failed with unexpected error ({exc_name}): {exc}", exc_info=True)
+
 
 # TODO: Get rid of Async prefix on this class name once we deprecate old sync code
 class AsyncBaseMCPClient:
@@ -81,24 +106,11 @@ class AsyncBaseMCPClient:
         try:
             result = await self.session.call_tool(tool_name, tool_args)
         except Exception as e:
-            # ToolError is raised by fastmcp for input validation errors (e.g., missing required properties)
-            # McpError is raised for other MCP-related errors
-            # Both are expected user-facing issues from external MCP servers
-            # Log at debug level to avoid triggering production alerts for expected failures
-
-            # Handle ExceptionGroup wrapping (Python 3.11+ async TaskGroup can wrap exceptions)
             exception_to_check = e
-            if hasattr(e, "exceptions") and e.exceptions:
-                # If it's an ExceptionGroup with a single wrapped exception, unwrap it
-                if len(e.exceptions) == 1:
-                    exception_to_check = e.exceptions[0]
-
-            if exception_to_check.__class__.__name__ in ("McpError", "ToolError"):
-                logger.debug(f"MCP tool '{tool_name}' execution failed: {str(exception_to_check)}")
-                # Return error message with failure status instead of raising to avoid Datadog alerts
-                return str(e), False
-            # Re-raise unexpected errors
-            raise
+            if hasattr(e, "exceptions") and e.exceptions and len(e.exceptions) == 1:
+                exception_to_check = e.exceptions[0]
+            _log_mcp_tool_error(logger, tool_name, exception_to_check)
+            return str(exception_to_check), False
 
         parsed_content = []
         for content_piece in result.content:
diff --git a/letta/services/mcp/fastmcp_client.py b/letta/services/mcp/fastmcp_client.py
index 01190e56..2a2b05c2 100644
--- a/letta/services/mcp/fastmcp_client.py
+++ b/letta/services/mcp/fastmcp_client.py
@@ -19,6 +19,7 @@ from mcp import Tool as MCPTool
 from letta.errors import LettaMCPConnectionError
 from letta.functions.mcp_client.types import SSEServerConfig, StreamableHTTPServerConfig
 from letta.log import get_logger
+from letta.services.mcp.base_client import _log_mcp_tool_error
 from letta.services.mcp.server_side_oauth import ServerSideOAuth
 
 logger = get_logger(__name__)
@@ -142,21 +143,11 @@ class AsyncFastMCPSSEClient:
         try:
             result = await self.client.call_tool(tool_name, tool_args)
         except Exception as e:
-            # ToolError is raised by fastmcp for input validation errors (e.g., missing required properties)
-            # McpError is raised for other MCP-related errors
-            # Both are expected user-facing issues from external MCP servers
-            # Log at debug level to avoid triggering production alerts for expected failures
-
-            # Handle ExceptionGroup wrapping (Python 3.11+ async TaskGroup can wrap exceptions)
             exception_to_check = e
-            if hasattr(e, "exceptions") and e.exceptions:
-                # If it's an ExceptionGroup with a single wrapped exception, unwrap it
-                if len(e.exceptions) == 1:
-                    exception_to_check = e.exceptions[0]
-
-            if exception_to_check.__class__.__name__ in ("McpError", "ToolError"):
-                logger.debug(f"MCP tool '{tool_name}' execution failed: {str(exception_to_check)}")
-            raise
+            if hasattr(e, "exceptions") and e.exceptions and len(e.exceptions) == 1:
+                exception_to_check = e.exceptions[0]
+            _log_mcp_tool_error(logger, tool_name, exception_to_check)
+            return str(exception_to_check), False
 
         # Parse content from result
         parsed_content = []
@@ -309,21 +300,11 @@ class AsyncFastMCPStreamableHTTPClient:
         try:
             result = await self.client.call_tool(tool_name, tool_args)
         except Exception as e:
-            # ToolError is raised by fastmcp for input validation errors (e.g., missing required properties)
-            # McpError is raised for other MCP-related errors
-            # Both are expected user-facing issues from external MCP servers
-            # Log at debug level to avoid triggering production alerts for expected failures
-
-            # Handle ExceptionGroup wrapping (Python 3.11+ async TaskGroup can wrap exceptions)
             exception_to_check = e
-            if hasattr(e, "exceptions") and e.exceptions:
-                # If it's an ExceptionGroup with a single wrapped exception, unwrap it
-                if len(e.exceptions) == 1:
-                    exception_to_check = e.exceptions[0]
-
-            if exception_to_check.__class__.__name__ in ("McpError", "ToolError"):
-                logger.debug(f"MCP tool '{tool_name}' execution failed: {str(exception_to_check)}")
-            raise
+            if hasattr(e, "exceptions") and e.exceptions and len(e.exceptions) == 1:
+                exception_to_check = e.exceptions[0]
+            _log_mcp_tool_error(logger, tool_name, exception_to_check)
+            return str(exception_to_check), False
 
         # Parse content from result
         parsed_content = []