Add suppress_log parameter to FastMCP errors

Add suppress_log parameter to FastMCPError and subclasses to skip server-side logging for expected control-flow errors.

Author: daniel-tsiang

src/fastmcp/exceptions.py

@@ -15,6 +15,10 @@ class FastMCPDeprecationWarning(DeprecationWarning):
 class FastMCPError(Exception):
     """Base error for FastMCP."""
 
+    def __init__(self, *args: object, suppress_log: bool = False) -> None:
+        super().__init__(*args)
+        self.suppress_log = suppress_log
+
 
 class ValidationError(FastMCPError):
     """Error in validating parameters or return values."""

src/fastmcp/prompts/function_prompt.py

@@ -24,7 +24,7 @@
 
 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 @@ async def render(
                     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,8 @@ async def _execute_single_tool(tool_use: ToolUseContent) -> ToolResultContent:
             )
         except ToolError as e:
             # ToolError is the escape hatch - always pass message through
-            logger.exception(f"Error calling sampling tool '{tool_use.name}'")
+            if not e.suppress_log:
+                logger.exception(f"Error calling sampling tool '{tool_use.name}'")
             return ToolResultContent(
                 type="tool_result",
                 toolUseId=tool_use.id,

src/fastmcp/server/server.py

@@ -1259,8 +1259,9 @@ async def call_tool(
                     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:
+                    if not e.suppress_log:
+                        logger.exception(f"Error calling tool {name!r}")
                     raise
                 except (ValidationError, PydanticValidationError):
                     logger.exception(f"Error validating tool {name!r}")
@@ -1389,8 +1390,9 @@ async def read_resource(
                         task_meta = replace(task_meta, fn_key=resource.key)
                     try:
                         return await resource._read(task_meta=task_meta)
-                    except (FastMCPError, McpError):
-                        logger.exception(f"Error reading resource {uri!r}")
+                    except (FastMCPError, McpError) as e:
+                        if not getattr(e, "suppress_log", False):
+                            logger.exception(f"Error reading resource {uri!r}")
                         raise
                     except Exception as e:
                         logger.exception(f"Error reading resource {uri!r}")
@@ -1428,8 +1430,9 @@ async def read_resource(
                     task_meta = replace(task_meta, fn_key=template.key)
                 try:
                     return await template._read(uri, params, task_meta=task_meta)
-                except (FastMCPError, McpError):
-                    logger.exception(f"Error reading resource {uri!r}")
+                except (FastMCPError, McpError) as e:
+                    if not getattr(e, "suppress_log", False):
+                        logger.exception(f"Error reading resource {uri!r}")
                     raise
                 except Exception as e:
                     logger.exception(f"Error reading resource {uri!r}")
@@ -1542,8 +1545,9 @@ async def render_prompt(
                     task_meta = replace(task_meta, fn_key=prompt.key)
                 try:
                     return await prompt._render(arguments, task_meta=task_meta)
-                except (FastMCPError, McpError):
-                    logger.exception(f"Error rendering prompt {name!r}")
+                except (FastMCPError, McpError) as e:
+                    if not getattr(e, "suppress_log", False):
+                        logger.exception(f"Error rendering prompt {name!r}")
                     raise
                 except Exception as e:
                     logger.exception(f"Error rendering prompt {name!r}")

tests/client/client/test_client.py

@@ -284,7 +284,7 @@ def strict_typed_prompt(numbers: list[int]) -> str:
     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,5 +1,7 @@
 """Client error handling tests."""
 
+import logging
+
 import mcp.types
 import pytest
 from mcp.types import TextContent
@@ -8,7 +10,7 @@
 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.server import FastMCP
 
 
@@ -265,3 +267,161 @@ async def test_error_with_structured_content_does_not_parse_data(self):
         assert parsed.is_error is True
         assert parsed.data is None
         assert parsed.structured_content == {"key": "value"}
+
+
+class TestSuppressLog:
+    async def test_tool_error_with_suppress_log_does_not_log(self, caplog):
+        """ToolError with suppress_log=True should skip logger.exception()."""
+        mcp = FastMCP("TestServer")
+
+        @mcp.tool
+        def suppressed_error_tool():
+            raise ToolError("Missing required parameter", suppress_log=True)
+
+        async with Client(transport=FastMCPTransport(mcp)) as client:
+            with caplog.at_level(logging.ERROR):
+                result = await client.call_tool_mcp("suppressed_error_tool", {})
+
+        assert result.isError
+        assert isinstance(result.content[0], TextContent)
+        assert "Missing required parameter" in result.content[0].text
+        assert not any(
+            "Error calling tool" in record.message and record.levelname == "ERROR"
+            for record in caplog.records
+        )
+
+    async def test_regular_tool_error_still_logs(self, caplog):
+        """ToolError without suppress_log still triggers logger.exception()."""
+        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_suppress_log_does_not_log(self, caplog):
+        """ResourceError with suppress_log=True should skip logger.exception()."""
+        mcp = FastMCP("TestServer")
+
+        @mcp.resource("test://suppressed")
+        def suppressed_resource():
+            raise ResourceError(
+                "Resource unavailable, try again later", suppress_log=True
+            )
+
+        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://suppressed")
+
+        assert "Resource unavailable, try again later" in str(exc_info.value)
+        assert not any(
+            "Error reading resource" in record.message and record.levelname == "ERROR"
+            for record in caplog.records
+        )
+
+    async def test_regular_resource_error_still_logs(self, caplog):
+        """ResourceError without suppress_log still triggers logger.exception()."""
+        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_suppress_log_does_not_log(self, caplog):
+        """PromptError with suppress_log=True should skip logger.exception()."""
+        mcp = FastMCP("TestServer")
+
+        @mcp.prompt
+        def suppressed_prompt():
+            raise PromptError(
+                "Insufficient context, provide more details", suppress_log=True
+            )
+
+        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("suppressed_prompt")
+
+        assert "Insufficient context" in str(exc_info.value)
+        assert not any(
+            "Error rendering prompt" in record.message and record.levelname == "ERROR"
+            for record in caplog.records
+        )
+
+    async def test_regular_prompt_error_still_logs(self, caplog):
+        """PromptError without suppress_log still triggers logger.exception()."""
+        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_suppress_log_does_not_log(self, caplog):
+        """ToolError with suppress_log=True in sampling should skip logging."""
+        from mcp.types import ToolUseContent
+
+        from fastmcp.server.sampling.run import SamplingTool, execute_tools
+
+        async def suppressed_sampling_tool(x: int) -> int:
+            raise ToolError("Expected sampling error", suppress_log=True)
+
+        tool = SamplingTool.from_function(suppressed_sampling_tool)
+        tool_use = ToolUseContent(
+            type="tool_use",
+            id="test-id",
+            name="suppressed_sampling_tool",
+            input={"x": 42},
+        )
+
+        with caplog.at_level(logging.ERROR):
+            results = await execute_tools(
+                tool_calls=[tool_use],
+                tool_map={"suppressed_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 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 @@ def typed_prompt(numbers: list[int]) -> str:
 
         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."""