Add log_level parameter to FastMCP errors (#4036)

Co-authored-by: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>

src/fastmcp/exceptions.py

@@ -1,5 +1,7 @@
 """Custom exceptions for FastMCP."""
 
+import logging
+
 from mcp import McpError  # noqa: F401
 
 
@@ -15,6 +17,10 @@ class FastMCPDeprecationWarning(DeprecationWarning):
 class FastMCPError(Exception):
     """Base error for FastMCP."""
 
+    def __init__(self, *args: object, log_level: int = logging.ERROR) -> None:
+        super().__init__(*args)
+        self.log_level = log_level
+
 
 class ValidationError(FastMCPError):
     """Error in validating parameters or return values."""

src/fastmcp/prompts/function_prompt.py

@@ -24,7 +24,7 @@ from pydantic.json_schema import SkipJsonSchema
 
 import fastmcp
 from fastmcp.decorators import resolve_task_config
-from fastmcp.exceptions import FastMCPDeprecationWarning, PromptError
+from fastmcp.exceptions import FastMCPDeprecationWarning, FastMCPError, PromptError
 from fastmcp.prompts.base import Prompt, PromptArgument, PromptResult
 from fastmcp.server.auth.authorization import AuthCheck
 from fastmcp.server.dependencies import (
@@ -361,6 +361,8 @@ class FunctionPrompt(Prompt):
                     result = await result
 
             return self.convert_result(result)
+        except FastMCPError:
+            raise
         except Exception as e:
             logger.exception(f"Error rendering prompt {self.name}")
             raise PromptError(f"Error rendering prompt {self.name!r}: {e}") from e

src/fastmcp/server/sampling/run.py

@@ -304,7 +304,11 @@ async def execute_tools(
             )
         except ToolError as e:
             # ToolError is the escape hatch - always pass message through
-            logger.exception(f"Error calling sampling tool '{tool_use.name}'")
+            logger.log(
+                e.log_level,
+                f"Error calling sampling tool '{tool_use.name}'",
+                exc_info=True,
+            )
             return ToolResultContent(
                 type="tool_result",
                 toolUseId=tool_use.id,

src/fastmcp/server/server.py

@@ -1264,8 +1264,10 @@ class FastMCP(
                     task_meta = replace(task_meta, fn_key=tool.key)
                 try:
                     return await tool._run(arguments or {}, task_meta=task_meta)
-                except FastMCPError:
-                    logger.exception(f"Error calling tool {name!r}")
+                except FastMCPError as e:
+                    logger.log(
+                        e.log_level, f"Error calling tool {name!r}", exc_info=True
+                    )
                     raise
                 except (ValidationError, PydanticValidationError):
                     logger.exception(f"Error validating tool {name!r}")
@@ -1394,7 +1396,14 @@ class FastMCP(
                         task_meta = replace(task_meta, fn_key=resource.key)
                     try:
                         return await resource._read(task_meta=task_meta)
-                    except (FastMCPError, McpError):
+                    except FastMCPError as e:
+                        logger.log(
+                            e.log_level,
+                            f"Error reading resource {uri!r}",
+                            exc_info=True,
+                        )
+                        raise
+                    except McpError:
                         logger.exception(f"Error reading resource {uri!r}")
                         raise
                     except Exception as e:
@@ -1433,7 +1442,12 @@ class FastMCP(
                     task_meta = replace(task_meta, fn_key=template.key)
                 try:
                     return await template._read(uri, params, task_meta=task_meta)
-                except (FastMCPError, McpError):
+                except FastMCPError as e:
+                    logger.log(
+                        e.log_level, f"Error reading resource {uri!r}", exc_info=True
+                    )
+                    raise
+                except McpError:
                     logger.exception(f"Error reading resource {uri!r}")
                     raise
                 except Exception as e:
@@ -1547,7 +1561,12 @@ class FastMCP(
                     task_meta = replace(task_meta, fn_key=prompt.key)
                 try:
                     return await prompt._render(arguments, task_meta=task_meta)
-                except (FastMCPError, McpError):
+                except FastMCPError as e:
+                    logger.log(
+                        e.log_level, f"Error rendering prompt {name!r}", exc_info=True
+                    )
+                    raise
+                except McpError:
                     logger.exception(f"Error rendering prompt {name!r}")
                     raise
                 except Exception as e:

tests/client/client/test_client.py

@@ -284,7 +284,7 @@ async def test_server_deserialization_error():
     client = Client(transport=FastMCPTransport(server))
 
     async with client:
-        with pytest.raises(McpError, match="Error rendering prompt"):
+        with pytest.raises(McpError, match="Could not convert argument"):
             await client.get_prompt(
                 "strict_typed_prompt",
                 {

tests/client/client/test_error_handling.py

@@ -1,14 +1,17 @@
 """Client error handling tests."""
 
+import logging
+
 import mcp.types
 import pytest
-from mcp.types import TextContent
+from mcp.types import TextContent, ToolUseContent
 from pydantic import AnyUrl
 
 from fastmcp.client import Client
 from fastmcp.client.mixins.tools import _parse_call_tool_result
 from fastmcp.client.transports import FastMCPTransport
-from fastmcp.exceptions import ResourceError, ToolError
+from fastmcp.exceptions import PromptError, ResourceError, ToolError
+from fastmcp.server.sampling.run import SamplingTool, execute_tools
 from fastmcp.server.server import FastMCP
 
 
@@ -265,3 +268,175 @@ class TestParseToolResultEdgeCases:
         assert parsed.is_error is True
         assert parsed.data is None
         assert parsed.structured_content == {"key": "value"}
+
+
+class TestLogLevel:
+    async def test_tool_error_with_custom_log_level(self, caplog):
+        """ToolError with custom log_level should log at specified level."""
+        mcp = FastMCP("TestServer")
+
+        @mcp.tool
+        def custom_level_tool():
+            raise ToolError("Missing required parameter", log_level=logging.WARNING)
+
+        async with Client(transport=FastMCPTransport(mcp)) as client:
+            with caplog.at_level(logging.WARNING):
+                result = await client.call_tool_mcp("custom_level_tool", {})
+
+        assert result.isError
+        assert isinstance(result.content[0], TextContent)
+        assert "Missing required parameter" in result.content[0].text
+        assert any(
+            "Error calling tool" in record.message and record.levelname == "WARNING"
+            for record in caplog.records
+        )
+        assert not any(
+            "Error calling tool" in record.message and record.levelname == "ERROR"
+            for record in caplog.records
+        )
+
+    async def test_regular_tool_error_logs_at_error(self, caplog):
+        """ToolError with default log_level logs at ERROR."""
+        mcp = FastMCP("TestServer")
+
+        @mcp.tool
+        def regular_error_tool():
+            raise ToolError("Something went wrong")
+
+        async with Client(transport=FastMCPTransport(mcp)) as client:
+            with caplog.at_level(logging.ERROR):
+                result = await client.call_tool_mcp("regular_error_tool", {})
+
+        assert result.isError
+        assert isinstance(result.content[0], TextContent)
+        assert "Something went wrong" in result.content[0].text
+        assert any(
+            "Error calling tool 'regular_error_tool'" in record.message
+            and record.levelname == "ERROR"
+            for record in caplog.records
+        )
+
+    async def test_resource_error_with_custom_log_level(self, caplog):
+        """ResourceError with custom log_level should log at specified level."""
+        mcp = FastMCP("TestServer")
+
+        @mcp.resource("test://custom")
+        def custom_level_resource():
+            raise ResourceError(
+                "Resource unavailable, try again later", log_level=logging.WARNING
+            )
+
+        async with Client(transport=FastMCPTransport(mcp)) as client:
+            with caplog.at_level(logging.WARNING):
+                with pytest.raises(Exception) as exc_info:
+                    await client.read_resource_mcp("test://custom")
+
+        assert "Resource unavailable, try again later" in str(exc_info.value)
+        assert any(
+            "Error reading resource" in record.message and record.levelname == "WARNING"
+            for record in caplog.records
+        )
+        assert not any(
+            "Error reading resource" in record.message and record.levelname == "ERROR"
+            for record in caplog.records
+        )
+
+    async def test_regular_resource_error_logs_at_error(self, caplog):
+        """ResourceError with default log_level logs at ERROR."""
+        mcp = FastMCP("TestServer")
+
+        @mcp.resource("test://regular")
+        def regular_resource():
+            raise ResourceError("Something went wrong")
+
+        async with Client(transport=FastMCPTransport(mcp)) as client:
+            with caplog.at_level(logging.ERROR):
+                with pytest.raises(Exception) as exc_info:
+                    await client.read_resource_mcp("test://regular")
+
+        assert "Something went wrong" in str(exc_info.value)
+        assert any(
+            "Error reading resource 'test://regular'" in record.message
+            and record.levelname == "ERROR"
+            for record in caplog.records
+        )
+
+    async def test_prompt_error_with_custom_log_level(self, caplog):
+        """PromptError with custom log_level should log at specified level."""
+        mcp = FastMCP("TestServer")
+
+        @mcp.prompt
+        def custom_level_prompt():
+            raise PromptError(
+                "Insufficient context, provide more details", log_level=logging.WARNING
+            )
+
+        async with Client(transport=FastMCPTransport(mcp)) as client:
+            with caplog.at_level(logging.WARNING):
+                with pytest.raises(Exception) as exc_info:
+                    await client.get_prompt("custom_level_prompt")
+
+        assert "Insufficient context" in str(exc_info.value)
+        assert any(
+            "Error rendering prompt" in record.message and record.levelname == "WARNING"
+            for record in caplog.records
+        )
+        assert not any(
+            "Error rendering prompt" in record.message and record.levelname == "ERROR"
+            for record in caplog.records
+        )
+
+    async def test_regular_prompt_error_logs_at_error(self, caplog):
+        """PromptError with default log_level logs at ERROR."""
+        mcp = FastMCP("TestServer")
+
+        @mcp.prompt
+        def regular_prompt():
+            raise PromptError("Something went wrong")
+
+        async with Client(transport=FastMCPTransport(mcp)) as client:
+            with caplog.at_level(logging.ERROR):
+                with pytest.raises(Exception) as exc_info:
+                    await client.get_prompt("regular_prompt")
+
+        assert "Something went wrong" in str(exc_info.value)
+        assert any(
+            "Error rendering prompt 'regular_prompt'" in record.message
+            and record.levelname == "ERROR"
+            for record in caplog.records
+        )
+
+    async def test_sampling_tool_error_with_custom_log_level(self, caplog):
+        """ToolError with custom log_level in sampling should log at specified level."""
+
+        async def custom_level_sampling_tool(x: int) -> int:
+            raise ToolError("Expected sampling error", log_level=logging.WARNING)
+
+        tool = SamplingTool.from_function(custom_level_sampling_tool)
+        tool_use = ToolUseContent(
+            type="tool_use",
+            id="test-id",
+            name="custom_level_sampling_tool",
+            input={"x": 42},
+        )
+
+        with caplog.at_level(logging.WARNING):
+            results = await execute_tools(
+                tool_calls=[tool_use],
+                tool_map={"custom_level_sampling_tool": tool},
+                mask_error_details=False,
+            )
+
+        assert len(results) == 1
+        assert results[0].isError
+        assert "Expected sampling error" in results[0].content[0].text  # type: ignore
+        assert any(
+            "Error calling sampling tool" in record.message
+            and record.levelname == "WARNING"
+            for record in caplog.records
+        )
+        assert not any(
+            "Error calling sampling tool" in record.message
+            and record.levelname == "ERROR"
+            for record in caplog.records
+        )

tests/prompts/test_prompt.py

@@ -273,11 +273,13 @@ class TestPromptTypeConversion:
 
         prompt = Prompt.from_function(typed_prompt)
 
-        # Test with invalid JSON - should raise PromptError due to exception handling in render()
+        # Test with invalid JSON - should raise PromptError with type conversion details
         with pytest.raises(PromptError) as exc_info:
             await prompt.render(arguments={"numbers": "not valid json"})
 
-        assert f"Error rendering prompt {prompt.name!r}" in str(exc_info.value)
+        # PromptError passes through unchanged
+        assert "Could not convert argument 'numbers'" in str(exc_info.value)
+        assert "list[int]" in str(exc_info.value)
 
     async def test_json_parsing_fallback(self):
         """Test that JSON parsing falls back to direct validation when needed."""