fix: structure MCP tool error envelopes

packages/nmp_common/src/nmp/common/mcp/README.md

@@ -41,8 +41,12 @@ Converts exceptions into standardized error responses with automatic logging.
 ```python
 {
     "success": False,
-    "error": "Connection refused to localhost:8080",
-    "error_type": "ConnectionError"
+    "error": {
+        "code": "ConnectionError",
+        "message": "Connection refused to localhost:8080",
+        "hint": "Check the MCP server logs for details, then retry after fixing the request or platform state.",
+        "retryable": True
+    }
 }
 ```
 
@@ -68,9 +72,9 @@ async def deploy_model(model_id: str) -> dict[str, Any]:
 **Why Use This Pattern**:
 
 - AI agents can reliably check `success` field
-- Consistent error structure across all tools
+- Consistent structured `error.code`, `error.message`, `error.hint`, and `error.retryable` across all tools
 - Automatic error logging with stack traces
-- Easy to add error codes, retry hints, or sanitization
+- Easy to add richer retry hints or sanitization
 - Success responses manually constructed with explicit fields
 
 ---
@@ -190,19 +194,14 @@ See `packages/nmp_common/src/nmp/common/sdk_factory.py` for SDK factory implemen
 def format_error_response(error: Exception) -> dict[str, Any]:
     logger.error(f"Error in MCP tool: {error}", exc_info=True)
 
-    # Map exception types to codes
-    error_codes = {
-        "ConnectionError": "PLATFORM_UNAVAILABLE",
-        "TimeoutError": "PLATFORM_TIMEOUT",
-        "HTTPStatusError": "API_ERROR",
-    }
-
     return {
         "success": False,
-        "error": str(error),
-        "error_type": type(error).__name__,
-        "error_code": error_codes.get(type(error).__name__, "UNKNOWN_ERROR"),
-        "retryable": isinstance(error, (ConnectionError, TimeoutError))
+        "error": {
+            "code": type(error).__name__,
+            "message": str(error),
+            "hint": "Check the MCP server logs for details, then retry after fixing the request or platform state.",
+            "retryable": isinstance(error, (ConnectionError, TimeoutError)),
+        }
     }
 ```
 

packages/nmp_common/src/nmp/common/mcp/error_handling.py

@@ -10,6 +10,8 @@
 
 logger = logging.getLogger(__name__)
 
+_DEFAULT_ERROR_HINT = "Check the MCP server logs for details, then retry after fixing the request or platform state."
+
 
 def format_error_response(error: Exception) -> dict[str, Any]:
     """
@@ -21,7 +23,7 @@ def format_error_response(error: Exception) -> dict[str, Any]:
         error: The exception to format
 
     Returns:
-        Dictionary with success=False, error message, and error type
+        Dictionary with success=False and a structured error object
 
     Example:
         >>> try:
@@ -30,8 +32,13 @@ def format_error_response(error: Exception) -> dict[str, Any]:
         ...     return format_error_response(e)
     """
     logger.error(f"Error in MCP tool: {error}", exc_info=True)
+    message = str(error) or type(error).__name__
     return {
         "success": False,
-        "error": str(error),
-        "error_type": type(error).__name__,
+        "error": {
+            "code": type(error).__name__,
+            "message": message,
+            "hint": _DEFAULT_ERROR_HINT,
+            "retryable": isinstance(error, (ConnectionError, TimeoutError)),
+        },
     }

packages/nmp_common/tests/mcp/test_error_handling.py

@@ -0,0 +1,43 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for shared MCP error response formatting."""
+
+from __future__ import annotations
+
+from nmp.common.mcp import format_error_response
+
+
+def test_format_error_response_returns_structured_error() -> None:
+    response = format_error_response(ValueError("bad input"))
+
+    assert response["success"] is False
+    assert response["error"] == {
+        "code": "ValueError",
+        "message": "bad input",
+        "hint": "Check the MCP server logs for details, then retry after fixing the request or platform state.",
+        "retryable": False,
+    }
+    assert "error_type" not in response
+
+
+def test_format_error_response_uses_exception_type_when_message_empty() -> None:
+    class EmptyMessageError(Exception):
+        pass
+
+    response = format_error_response(EmptyMessageError())
+
+    assert response["error"]["code"] == "EmptyMessageError"
+    assert response["error"]["message"] == "EmptyMessageError"
+
+
+def test_format_error_response_marks_connection_failures_retryable() -> None:
+    response = format_error_response(ConnectionError("connection reset"))
+
+    assert response["error"]["retryable"] is True
+
+
+def test_format_error_response_marks_timeouts_retryable() -> None:
+    response = format_error_response(TimeoutError("request timed out"))
+
+    assert response["error"]["retryable"] is True

services/core/entities/tests/integration/smoke_test_mcp.py

@@ -10,10 +10,11 @@
 from __future__ import annotations
 
 import os
-from typing import Generator
+from typing import Any, Generator
 
 import pytest
 from fastmcp import FastMCP
+from mcp.types import TextContent
 from nemo_platform import NeMoPlatform
 from nmp.common.sdk_factory import get_platform_sdk
 from nmp.core.entities.mcp.server import create_server
@@ -40,6 +41,12 @@ def mcp_server(nmp_base_url: str) -> Generator[FastMCP, None, None]:
     yield server
 
 
+def _text_content(tool_result: Any) -> str:
+    first_content = tool_result.content[0]
+    assert isinstance(first_content, TextContent)
+    return first_content.text
+
+
 class TestEntitiesMCPServerSmoke:
     """Smoke tests for entities MCP server basic functionality."""
 
@@ -78,7 +85,7 @@ async def test_list_workspaces_matches_sdk(self, mcp_server: FastMCP, nemo_sdk:
         # Get workspaces via MCP tool
         tool_result = await mcp_server.call_tool("list_workspaces", {})
 
-        mcp_result = json.loads(tool_result.content[0].text)
+        mcp_result = json.loads(_text_content(tool_result))
         assert isinstance(mcp_result, dict)  # Type narrowing for ty
         assert mcp_result["success"] is True
 
@@ -94,20 +101,40 @@ async def test_list_workspaces_matches_sdk(self, mcp_server: FastMCP, nemo_sdk:
         )
 
     @pytest.mark.asyncio
-    async def test_list_workspaces_error_handling(self) -> None:
+    async def test_list_workspaces_error_handling(self, monkeypatch: pytest.MonkeyPatch) -> None:
         """
         Verify list_workspaces handles connection errors gracefully.
 
-        Creates a server with invalid URL to test error handling.
+        Creates a server with a controlled failing SDK client to test error handling.
         """
         import json
 
-        bad_server = create_server("http://invalid-host:9999")
-        tool_result = await bad_server.call_tool("list_workspaces", {})
+        class FailingWorkspacesClient:
+            def list(self, *args: object, **kwargs: object) -> object:
+                raise RuntimeError("platform unavailable")
+
+        class FailingPlatformClient:
+            workspaces = FailingWorkspacesClient()
 
-        result = json.loads(tool_result.content[0].text)
+        def get_failing_platform_sdk(base_url: str | None = None) -> FailingPlatformClient:
+            _ = base_url
+            return FailingPlatformClient()
+
+        monkeypatch.setattr(
+            "nmp.core.entities.mcp.server.get_platform_sdk",
+            get_failing_platform_sdk,
+        )
+        bad_server = create_server("http://unused.example.com")
+        tool_result = await bad_server.call_tool("list_workspaces", {})
 
-        assert isinstance(result, dict)
-        assert result["success"] is False, "Should indicate failure"
-        assert "error" in result, "Should contain error message"
-        assert "error_type" in result, "Should contain error type"
+        result = json.loads(_text_content(tool_result))
+
+        assert result == {
+            "success": False,
+            "error": {
+                "code": "RuntimeError",
+                "message": "platform unavailable",
+                "hint": "Check the MCP server logs for details, then retry after fixing the request or platform state.",
+                "retryable": False,
+            },
+        }

services/core/mcp/tests/integration/smoke_test.py

@@ -10,16 +10,15 @@
 from __future__ import annotations
 
 import os
-from typing import TYPE_CHECKING, Generator
+from typing import Any, Generator
 
 import pytest
+from fastmcp import FastMCP
+from mcp.types import TextContent
+from nemo_platform import NeMoPlatform
 from nmp.common.sdk_factory import get_platform_sdk
 from nmp.core.mcp.server import create_server
 
-if TYPE_CHECKING:
-    from fastmcp import FastMCP
-    from nemo_platform import NeMoPlatform
-
 
 @pytest.fixture(scope="module")
 def nmp_base_url() -> str:
@@ -41,6 +40,12 @@ def mcp_server(nmp_base_url: str) -> Generator[FastMCP, None, None]:
     yield server
 
 
+def _text_content(tool_result: Any) -> str:
+    first_content = tool_result.content[0]
+    assert isinstance(first_content, TextContent)
+    return first_content.text
+
+
 class TestMCPServerSmoke:
     """Smoke tests for MCP server basic functionality."""
 
@@ -79,7 +84,7 @@ async def test_list_workspaces_matches_sdk(self, mcp_server: FastMCP, nemo_sdk:
         # Get workspaces via MCP tool
         tool_result = await mcp_server.call_tool("list_workspaces", {})
 
-        mcp_result = json.loads(tool_result.content[0].text)
+        mcp_result = json.loads(_text_content(tool_result))
         assert isinstance(mcp_result, dict)  # Type narrowing for ty
 
         # Verify MCP returns same workspace IDs
@@ -95,24 +100,43 @@ async def test_list_workspaces_matches_sdk(self, mcp_server: FastMCP, nemo_sdk:
         )
 
     @pytest.mark.asyncio
-    async def test_list_workspaces_error_handling(self, nmp_base_url: str) -> None:
+    async def test_list_workspaces_error_handling(self, monkeypatch: pytest.MonkeyPatch) -> None:
         """
         Verify tool handles connection errors gracefully.
 
-        Creates a server with invalid URL to test error handling.
+        Mounts a controlled failing tool to test error handling.
         """
         import json
 
-        # Create server with invalid URL
-        bad_server = create_server("http://invalid-host:9999")
+        import nmp.core.mcp.server as mcp_server_module
+        from nmp.common.mcp import format_error_response
+
+        def create_failing_entities_mcp(_base_url: str | None = None) -> FastMCP:
+            server = FastMCP("Failing Entities Service")
+
+            @server.tool(description="List workspaces in the NeMo platform")
+            async def list_workspaces() -> dict[str, object]:
+                try:
+                    raise RuntimeError("platform unavailable")
+                except Exception as e:
+                    return format_error_response(e)
+
+            return server
+
+        monkeypatch.setattr(mcp_server_module, "create_entities_mcp", create_failing_entities_mcp)
+        bad_server = mcp_server_module.create_server("http://unused.example.com")
 
         # Execute tool - should return error, not raise
         tool_result = await bad_server.call_tool("list_workspaces", {})
 
-        result = json.loads(tool_result.content[0].text)
-
-        # Verify error response structure
-        assert isinstance(result, dict)
-        assert result["success"] is False, "Should indicate failure"
-        assert "error" in result, "Should contain error message"
-        assert "error_type" in result, "Should contain error type"
+        result = json.loads(_text_content(tool_result))
+
+        assert result == {
+            "success": False,
+            "error": {
+                "code": "RuntimeError",
+                "message": "platform unavailable",
+                "hint": "Check the MCP server logs for details, then retry after fixing the request or platform state.",
+                "retryable": False,
+            },
+        }