Log MCPServer handler exceptions by kind and keep crash details off the wire (#3314)

Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>

src/mcp/server/mcpserver/context.py

@@ -169,8 +169,12 @@ class Context(BaseModel, Generic[LifespanContextT, RequestT]):
             The resource content as either text or bytes
 
         Raises:
-            ResourceNotFoundError: If no resource or template matches the URI.
-            ResourceError: If template creation or resource reading fails.
+            ResourceNotFoundError: If no resource or template matches the URI, or the
+                handler raised it.
+            ResourceError: If the resource or template function raises `ResourceError`.
+            UnexpectedResourceError: If the resource or template function raises anything
+                else. `__cause__` is the original exception. Left uncaught in a tool, this
+                is logged as the tool's crash, while the two above are not.
             RuntimeError: If the resource returned an `InputRequiredResult`.
         """
         assert self._mcp_server is not None, "Context is not available outside of a request"

src/mcp/server/mcpserver/exceptions.py

@@ -6,20 +6,70 @@ class MCPServerError(Exception):
 
 
 class ResourceError(MCPServerError):
-    """Error in resource operations."""
+    """A resource failure you anticipated.
+
+    Raise this from a resource or resource template handler for a failure you saw
+    coming: the client receives a `-32603` protocol error carrying your message
+    (`ResourceNotFoundError` below is the `-32602` variant), and the server logs it
+    at INFO without a traceback. Any other exception (bar `MCPError`, which is a
+    protocol error) is treated as a crash: the client gets a generic message naming
+    only the URI, and the server logs the traceback at ERROR.
+
+    The SDK raises it too, and `UnexpectedResourceError` subclasses it, so
+    `except ResourceError` around `MCPServer.read_resource()` catches every read
+    failure, crash or not.
+    """
 
 
 class ResourceNotFoundError(ResourceError):
     """Resource does not exist.
 
-    Raise this from a resource template handler to signal that the requested instance does not exist;
-    clients receive `-32602` (invalid params) per
+    Raise this from a resource handler to signal that the requested instance does not exist.
+    Clients receive `-32602` (invalid params) per
     [SEP-2164](https://github.com/modelcontextprotocol/modelcontextprotocol/pull/2164).
     """
 
 
+class UnexpectedResourceError(ResourceError):
+    """A resource read failed with something other than `ResourceError` or `MCPError`.
+
+    The SDK raises this itself, around a crash in a resource or resource template
+    handler. You never raise it. `__cause__` is the original exception, which the
+    server logs with its traceback. The message names only the URI, so the
+    original text is withheld from the client.
+    """
+
+
 class ToolError(MCPServerError):
-    """Error in tool operations."""
+    """A tool failure you anticipated.
+
+    Raise this from a tool (or a resolver) for a failure you saw coming: the
+    call returns `is_error=True` with your message in `content` for the model to
+    read, and the server logs it at INFO without a traceback. A `ResourceError`
+    that escapes the tool (say from `ctx.read_resource()`) counts the same. Any
+    other exception bar `MCPError` (a protocol error) is treated as a crash: the
+    model sees only `Error executing tool <name>`, and the server logs the
+    traceback at ERROR. Inside a pydantic validator, raise `ValueError` as pydantic
+    expects; it arrives as an argument-validation failure, which is anticipated too.
+
+    The SDK raises it too, for an unknown tool name and for arguments that fail
+    the input schema, and `UnexpectedToolError` subclasses it, so `except ToolError`
+    around `MCPServer.call_tool()` catches every tool failure, crash or not.
+    """
+
+
+class UnexpectedToolError(ToolError):
+    """A tool call failed with something other than `ToolError`, `ResourceError`, or `MCPError`.
+
+    The SDK raises this itself, around a crash in the tool (or a resolver) or a
+    return value that fails output conversion. You never raise it. The message is
+    only `Error executing tool <name>` (followed by the same for a nested tool or
+    resource that crashed), so nothing from the original reaches the client.
+    `__cause__` is the original exception (or, for a nested tool or resource
+    crash, its `Unexpected...Error` wrapper), which the server logs with its
+    traceback before returning the `is_error=True` result. Catch it around
+    `MCPServer.call_tool()` to tell a crash from a deliberate `ToolError`.
+    """
 
 
 class InvalidSignature(Exception):

src/mcp/server/mcpserver/prompts/base.py

@@ -209,5 +209,5 @@ class Prompt(BaseModel):
             return messages
         except MCPError:
             raise
-        except Exception as e:
-            raise ValueError(f"Error rendering prompt {self.name}: {e}")
+        except Exception as exc:
+            raise ValueError(f"Error rendering prompt {self.name}") from exc

src/mcp/server/mcpserver/resolve.py

@@ -575,7 +575,15 @@ async def _fulfil(marker: _Marker, key: str, res: _Resolution) -> ElicitationRes
         if res.context.session.can_send_request:
             _require_capability(res.context, marker, key)
         if isinstance(marker, Elicit):
-            return await res.context.elicit(marker.message, marker.schema)
+            try:
+                return await res.context.elicit(marker.message, marker.schema)
+            except ValueError as e:
+                # Accepted with no content, or content that fails the schema: the same
+                # client mistake the input_required path below reports as a ToolError.
+                # (A pydantic ValidationError here means a non-conformant client sent a
+                # malformed ElicitResult; its text is not repeated back.)
+                detail = "received an invalid elicitation response" if isinstance(e, ValidationError) else str(e)
+                raise ToolError(f"Resolver {key!r}: {detail}") from e
         result = await res.context.session.send_request(
             _render_request(marker),
             _result_type(marker),

src/mcp/server/mcpserver/resources/templates.py

@@ -11,18 +11,15 @@ import anyio.to_thread
 from mcp_types import Annotations, Icon, InputRequiredResult
 from pydantic import BaseModel, Field, validate_call
 
-from mcp.server.mcpserver.exceptions import ResourceError
+from mcp.server.mcpserver.exceptions import ResourceError, UnexpectedResourceError
 from mcp.server.mcpserver.resources.types import FunctionResource, Resource
 from mcp.server.mcpserver.utilities.context_injection import find_context_parameter, inject_context
 from mcp.server.mcpserver.utilities.func_metadata import func_metadata
-from mcp.server.mcpserver.utilities.logging import get_logger
 from mcp.shared._callable_inspection import is_async_callable
 from mcp.shared.exceptions import MCPError
 from mcp.shared.path_security import contains_path_traversal, is_absolute_path
 from mcp.shared.uri_template import UriTemplate
 
-logger = get_logger(__name__)
-
 if TYPE_CHECKING:
     from mcp.server.context import LifespanContextT, RequestT
     from mcp.server.mcpserver.context import Context
@@ -217,7 +214,9 @@ class ResourceTemplate(BaseModel):
         carrying the echoed opaque state.
 
         Raises:
-            ResourceError: If creating the resource fails.
+            ResourceError: If the template function raises `ResourceError`.
+            UnexpectedResourceError: If the template function raises anything other
+                than `ResourceError` or `MCPError`. `__cause__` is the original exception.
         """
         try:
             # Add context to params if needed
@@ -246,5 +245,4 @@ class ResourceTemplate(BaseModel):
         except (ResourceError, MCPError):
             raise
         except Exception as exc:
-            logger.exception(f"Error creating resource from template {uri}")
-            raise ResourceError(f"Error creating resource from template {uri}") from exc
+            raise UnexpectedResourceError(f"Error creating resource from template {uri}") from exc

src/mcp/server/mcpserver/resources/types.py

@@ -18,7 +18,6 @@ from pydantic import Field, validate_call
 
 from mcp.server.mcpserver.resources.base import Resource
 from mcp.shared._callable_inspection import is_async_callable
-from mcp.shared.exceptions import MCPError
 
 # `application/*` types that are textual but predate the `+json`/`+xml`
 # structured-syntax suffixes, so the suffix rule below can't catch them.
@@ -80,33 +79,28 @@ class FunctionResource(Resource):
 
     async def read(self) -> str | bytes:
         """Read the resource by calling the wrapped function."""
-        try:
-            fn = self.fn
-            if is_async_callable(fn):
-                result = await fn()
-            else:
-                result = await anyio.to_thread.run_sync(self.fn)
+        fn = self.fn
+        if is_async_callable(fn):
+            result = await fn()
+        else:
+            result = await anyio.to_thread.run_sync(self.fn)
 
-            if isinstance(result, InputRequiredResult):
-                # A static resource function can never read the retry's
-                # input_responses (it takes no Context), so this can only be a
-                # mistake — reject it instead of JSON-dumping it as content.
-                raise ValueError(
-                    "static resources cannot return InputRequiredResult; only resource "
-                    "template functions participate in the multi-round-trip flow"
-                )
-            if isinstance(result, Resource):  # pragma: no cover
-                return await result.read()
-            elif isinstance(result, bytes):
-                return result
-            elif isinstance(result, str):
-                return result
-            else:
-                return pydantic_core.to_json(result, fallback=str, indent=2).decode()
-        except MCPError:
-            raise
-        except Exception as e:
-            raise ValueError(f"Error reading resource {self.uri}: {e}")
+        if isinstance(result, InputRequiredResult):
+            # A static resource function can never read the retry's
+            # input_responses (it takes no Context), so this can only be a
+            # mistake — reject it instead of JSON-dumping it as content.
+            raise ValueError(
+                "static resources cannot return InputRequiredResult; only resource "
+                "template functions participate in the multi-round-trip flow"
+            )
+        if isinstance(result, Resource):  # pragma: no cover
+            return await result.read()
+        elif isinstance(result, bytes):
+            return result
+        elif isinstance(result, str):
+            return result
+        else:
+            return pydantic_core.to_json(result, fallback=str, indent=2).decode()
 
     @classmethod
     def from_function(
@@ -183,12 +177,9 @@ class FileResource(Resource):
 
     async def read(self) -> str | bytes:
         """Read the file content."""
-        try:
-            if self.encoding is None:
-                return await anyio.to_thread.run_sync(self.path.read_bytes)
-            return await anyio.to_thread.run_sync(partial(self.path.read_text, encoding=self.encoding))
-        except Exception as e:
-            raise ValueError(f"Error reading file {self.path}: {e}")
+        if self.encoding is None:
+            return await anyio.to_thread.run_sync(self.path.read_bytes)
+        return await anyio.to_thread.run_sync(partial(self.path.read_text, encoding=self.encoding))
 
 
 class HttpResource(Resource):
@@ -228,18 +219,12 @@ class DirectoryResource(Resource):
         if not self.path.is_dir():
             raise NotADirectoryError(f"Not a directory: {self.path}")
 
-        try:
-            if self.pattern:
-                return list(self.path.glob(self.pattern)) if not self.recursive else list(self.path.rglob(self.pattern))
-            return list(self.path.glob("*")) if not self.recursive else list(self.path.rglob("*"))
-        except Exception as e:
-            raise ValueError(f"Error listing directory {self.path}: {e}")
+        if self.pattern:
+            return list(self.path.glob(self.pattern)) if not self.recursive else list(self.path.rglob(self.pattern))
+        return list(self.path.glob("*")) if not self.recursive else list(self.path.rglob("*"))
 
     async def read(self) -> str:  # Always returns JSON string  # pragma: no cover
         """Read the directory listing."""
-        try:
-            files = await anyio.to_thread.run_sync(self.list_files)
-            file_list = [str(f.relative_to(self.path)) for f in files if f.is_file()]
-            return json.dumps({"files": file_list}, indent=2)
-        except Exception as e:
-            raise ValueError(f"Error reading directory {self.path}: {e}")
+        files = await anyio.to_thread.run_sync(self.list_files)
+        file_list = [str(f.relative_to(self.path)) for f in files if f.is_file()]
+        return json.dumps({"files": file_list}, indent=2)

src/mcp/server/mcpserver/server.py

@@ -6,7 +6,7 @@ import base64
 import inspect
 from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Mapping, Sequence
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
-from typing import Any, Generic, Literal, TypeVar, overload
+from typing import Any, Generic, Literal, TypeVar, cast, overload
 
 import anyio
 import pydantic_core
@@ -44,7 +44,7 @@ from mcp_types import PromptArgument as MCPPromptArgument
 from mcp_types import Resource as MCPResource
 from mcp_types import ResourceTemplate as MCPResourceTemplate
 from mcp_types import Tool as MCPTool
-from pydantic import BaseModel
+from pydantic import BaseModel, ValidationError
 from pydantic.networks import AnyUrl
 from starlette.applications import Starlette
 from starlette.middleware import Middleware
@@ -71,7 +71,13 @@ from mcp.server.lowlevel.helper_types import ReadResourceContents
 from mcp.server.lowlevel.server import LifespanResultT, Server
 from mcp.server.lowlevel.server import lifespan as default_lifespan
 from mcp.server.mcpserver.context import Context
-from mcp.server.mcpserver.exceptions import ResourceError, ResourceNotFoundError
+from mcp.server.mcpserver.exceptions import (
+    ResourceError,
+    ResourceNotFoundError,
+    ToolError,
+    UnexpectedResourceError,
+    UnexpectedToolError,
+)
 from mcp.server.mcpserver.prompts import Prompt, PromptManager
 from mcp.server.mcpserver.resources import (
     DEFAULT_RESOURCE_SECURITY,
@@ -421,8 +427,18 @@ class MCPServer(Generic[LifespanResultT]):
             return await self.call_tool(params.name, params.arguments or {}, context)
         except MCPError:
             raise
-        except Exception as e:
-            return CallToolResult(content=[TextContent(type="text", text=str(e))], is_error=True)
+        except Exception as exc:
+            if isinstance(exc, ToolError) and not isinstance(exc, UnexpectedToolError):
+                if isinstance(exc.__cause__, ValidationError):
+                    # Field names only: the rejected values are the caller's data.
+                    fields = sorted({".".join(str(part) for part in err["loc"]) for err in exc.__cause__.errors()})
+                    logger.info("Tool %r rejected arguments: %r", params.name, fields)
+                else:
+                    # %r keeps peer-supplied text on one line.
+                    logger.info("Tool %r failed: %r", params.name, str(exc))
+            else:
+                logger.exception("Tool %r raised an unexpected exception", params.name)
+            return CallToolResult(content=[TextContent(type="text", text=str(exc))], is_error=True)
 
     async def _handle_list_resources(
         self, ctx: ServerRequestContext[LifespanResultT], params: PaginatedRequestParams | None
@@ -435,10 +451,13 @@ class MCPServer(Generic[LifespanResultT]):
         context = Context(request_context=ctx, mcp_server=self, input_params=params, subscriptions=self._subscriptions)
         try:
             results = await self.read_resource(params.uri, context)
-        except ResourceNotFoundError as err:
-            raise MCPError(code=INVALID_PARAMS, message=str(err), data={"uri": str(params.uri)})
         except ResourceError as err:
-            raise MCPError(code=INTERNAL_ERROR, message=str(err), data={"uri": str(params.uri)})
+            if isinstance(err, UnexpectedResourceError):
+                logger.exception("Resource %r raised an unexpected exception", str(params.uri))
+            else:
+                logger.info("Resource %r failed: %r", str(params.uri), str(err))
+            code = INVALID_PARAMS if isinstance(err, ResourceNotFoundError) else INTERNAL_ERROR
+            raise MCPError(code=code, message=str(err), data={"uri": str(params.uri)})
         if isinstance(results, InputRequiredResult):
             return results
         contents: list[TextResourceContents | BlobResourceContents] = []
@@ -499,7 +518,16 @@ class MCPServer(Generic[LifespanResultT]):
     async def call_tool(
         self, name: str, arguments: dict[str, Any], context: Context[LifespanResultT, Any] | None = None
     ) -> CallToolResult | InputRequiredResult:
-        """Call a tool by name with arguments."""
+        """Call a tool by name with arguments.
+
+        Raises:
+            ToolError: If the tool is unknown, the arguments fail validation, or the
+                tool (or a resolver) raises `ToolError` or `ResourceError`.
+            UnexpectedToolError: If the tool (or a resolver) raises anything else, or
+                its return value fails output conversion. `__cause__` is the original
+                exception (or, for a nested tool or resource crash, its wrapper).
+            MCPError: Raised by the tool or a resolver; passed through unchanged.
+        """
         if context is None:
             context = Context(mcp_server=self, subscriptions=self._subscriptions)
         return await self._tool_manager.call_tool(name, arguments, context, convert_result=True)
@@ -550,23 +578,27 @@ class MCPServer(Generic[LifespanResultT]):
 
         Raises:
             ResourceNotFoundError: If no resource or template matches the URI.
-            ResourceError: If template creation or resource reading fails.
+            ResourceError: If the resource or template function raises `ResourceError`.
+            UnexpectedResourceError: If reading the resource (or creating it from a
+                template) raises anything other than `ResourceError` or `MCPError`.
+                `__cause__` is the original exception.
+            MCPError: Raised by the resource or template function; passed through unchanged.
         """
         if context is None:
             context = Context(mcp_server=self, subscriptions=self._subscriptions)
-        resource = await self._resource_manager.get_resource(uri, context)
-        if isinstance(resource, InputRequiredResult):
-            return resource
-
         try:
-            content = await resource.read()
+            resource = await self._resource_manager.get_resource(uri, context)
+            if isinstance(resource, InputRequiredResult):
+                return resource
+            # Checked at runtime because a Resource subclass may not honour the annotation.
+            content = cast(object, await resource.read())
+            if not isinstance(content, str | bytes):
+                raise TypeError(f"Resource.read() must return str or bytes, not {type(content).__name__}")
             return [ReadResourceContents(content=content, mime_type=resource.mime_type, meta=resource.meta)]
-        except MCPError:
+        except (MCPError, ResourceError):
             raise
         except Exception as exc:
-            logger.exception(f"Error getting resource {uri}")
-            # If an exception happens when reading the resource, we should not leak the exception to the client.
-            raise ResourceError(f"Error reading resource {uri}") from exc
+            raise UnexpectedResourceError(f"Error reading resource {uri}") from exc
 
     def add_tool(
         self,
@@ -712,10 +744,18 @@ class MCPServer(Generic[LifespanResultT]):
             async def handler(
                 ctx: ServerRequestContext[LifespanResultT], params: CompleteRequestParams
             ) -> CompleteResult:
-                result = await func(params.ref, params.argument, params.context)
-                return CompleteResult(
-                    completion=result if result is not None else Completion(values=[], total=None, has_more=None),
-                )
+                try:
+                    result = await func(params.ref, params.argument, params.context)
+                    return CompleteResult(
+                        completion=result if result is not None else Completion(values=[], total=None, has_more=None),
+                    )
+                except MCPError:
+                    raise
+                except Exception as exc:
+                    logger.exception("Completion for argument %r raised an unexpected exception", params.argument.name)
+                    raise MCPError(
+                        code=INTERNAL_ERROR, message=f"Error completing argument {params.argument.name}"
+                    ) from exc
 
             self._lowlevel_server.add_request_handler("completion/complete", CompleteRequestParams, handler)
             return func
@@ -1299,7 +1339,7 @@ class MCPServer(Generic[LifespanResultT]):
         except MCPError:
             raise
         except Exception as e:
-            logger.exception(f"Error getting prompt {name}")
+            # Not logged here: the dispatcher boundary logs it once.
             raise ValueError(str(e)) from e
 
 

src/mcp/server/mcpserver/tools/base.py

@@ -5,9 +5,15 @@ from functools import cached_property
 from typing import TYPE_CHECKING, Any
 
 from mcp_types import Icon, InputRequiredResult, ToolAnnotations
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, ValidationError
 
-from mcp.server.mcpserver.exceptions import InvalidSignature, ToolError
+from mcp.server.mcpserver.exceptions import (
+    InvalidSignature,
+    ResourceError,
+    ToolError,
+    UnexpectedResourceError,
+    UnexpectedToolError,
+)
 from mcp.server.mcpserver.resolve import (
     build_resolver_plans,
     find_resolved_parameters,
@@ -128,39 +134,52 @@ class Tool(BaseModel):
     ) -> Any:
         """Run the tool with arguments.
 
+        Every failure other than `MCPError` is raised as a `ToolError` whose message
+        starts `Error executing tool <name>` and whose `__cause__` is what was raised.
+        An anticipated failure keeps its own text after the prefix. A crash does not,
+        so nothing from an unexpected exception reaches the client.
+
         Raises:
-            ToolError: If the tool function raises during execution.
+            ToolError: If the arguments fail validation against the input schema, or
+                the tool function (or a resolver) raises `ToolError` or `ResourceError`.
+            UnexpectedToolError: If argument validation, the tool function, or a
+                resolver raises anything else, or the return value fails output conversion.
         """
+        try:
+            validated = self.fn_metadata.validate_arguments(arguments)
+        except ValidationError as exc:
+            # The caller's arguments don't match the input schema: the model's mistake
+            # to read and correct, so it is reported like a deliberate ToolError.
+            raise ToolError(f"Error executing tool {self.name}: {exc}") from exc
+        except MCPError:
+            raise
+        except Exception as exc:
+            # A custom validator or default_factory that raises is a crash.
+            raise UnexpectedToolError(f"Error executing tool {self.name}") from exc
+
         try:
             pass_directly: dict[str, Any] = {}
             if self.context_kwarg is not None:
                 pass_directly[self.context_kwarg] = context
 
-            # Resolvers see the same validated arguments the tool body receives:
-            # validate once and reuse it, so a `default_factory`/stateful validator
-            # can't hand a by-name resolver a different value than the body.
-            pre_validated: dict[str, Any] | None = None
+            # Resolvers see the same validated arguments the tool body receives, so a
+            # `default_factory`/stateful validator can't hand a by-name resolver a
+            # different value than the body.
             if self.resolved_params:
-                pre_validated = self.fn_metadata.validate_arguments(arguments)
-                resolved = await resolve_arguments(self.resolved_params, self.resolver_plans, pre_validated, context)
+                resolved = await resolve_arguments(self.resolved_params, self.resolver_plans, validated, context)
                 if isinstance(resolved, InputRequiredResult):
                     # A resolver still needs client input (>= 2026-07-28): surface the
                     # batched questions instead of running the tool body this round.
                     return self.fn_metadata.convert_result(resolved) if convert_result else resolved
                 pass_directly |= resolved
 
-            result = await self.fn_metadata.call_fn_with_arg_validation(
-                self.fn,
-                self.is_async,
-                arguments,
-                pass_directly or None,
-                pre_validated=pre_validated,
-            )
+            result = await self.fn_metadata.call_fn(self.fn, self.is_async, validated, pass_directly)
 
             # Registration rejects the annotated form of this combination; this covers
-            # a body that returns an InputRequiredResult without declaring it.
+            # a body that returns an InputRequiredResult without declaring it. It is
+            # an authoring bug, so it is raised as a crash rather than a ToolError.
             if self.resolved_params and isinstance(result, InputRequiredResult):
-                raise ToolError(
+                raise RuntimeError(
                     "the tool returned an InputRequiredResult but its parameters use Resolve(...); "
                     "a call has one input_required channel, so the multi-round flow is driven "
                     "either by resolvers or by the tool body, not both"
@@ -177,5 +196,15 @@ class Tool(BaseModel):
             # it as a top-level JSON-RPC error rather than wrapping it as a
             # `CallToolResult(isError=True)` execution failure.
             raise
-        except Exception as e:
-            raise ToolError(f"Error executing tool {self.name}: {e}") from e
+        # Everything else reaches the model as an is_error result under this tool's
+        # name, and the wrapper's type tells the server whether to log a crash.
+        except (UnexpectedToolError, UnexpectedResourceError) as exc:
+            # A nested tool call or resource read crashed: still a crash here. Its
+            # message is already the generic one, so it is safe to carry along.
+            raise UnexpectedToolError(f"Error executing tool {self.name}: {exc}") from exc
+        except (ToolError, ResourceError) as exc:
+            # Raised deliberately by the tool, a resolver, or a resource it read.
+            raise ToolError(f"Error executing tool {self.name}: {exc}") from exc
+        except Exception as exc:
+            # A crash: the exception's own text stays on the server.
+            raise UnexpectedToolError(f"Error executing tool {self.name}") from exc

src/mcp/server/mcpserver/utilities/func_metadata.py

@@ -23,7 +23,7 @@ from pydantic import (
 )
 from pydantic.fields import FieldInfo
 from pydantic.json_schema import GenerateJsonSchema, JsonSchemaWarningKind
-from typing_extensions import NotRequired, ReadOnly, TypedDict, get_type_hints, is_typeddict
+from typing_extensions import NotRequired, ReadOnly, TypedDict, deprecated, get_type_hints, is_typeddict
 from typing_inspection.introspection import (
     UNKNOWN,
     AnnotationSource,
@@ -35,6 +35,7 @@ from typing_inspection.introspection import (
 from mcp.server.mcpserver.exceptions import InvalidSignature
 from mcp.server.mcpserver.utilities.logging import get_logger
 from mcp.server.mcpserver.utilities.types import Audio, Image
+from mcp.shared.exceptions import MCPDeprecationWarning
 
 logger = get_logger(__name__)
 
@@ -125,6 +126,28 @@ class FuncMetadata(BaseModel):
         arguments_parsed_model = self.arg_model.model_validate(arguments_pre_parsed)
         return arguments_parsed_model.model_dump_one_level()
 
+    async def call_fn(
+        self,
+        fn: Callable[..., Any | Awaitable[Any]],
+        fn_is_async: bool,
+        arguments: dict[str, Any],
+        arguments_to_pass_directly: dict[str, Any] | None = None,
+    ) -> Any:
+        """Call the function with already-validated `arguments` plus `arguments_to_pass_directly`.
+
+        `arguments` is the output of `validate_arguments`. A sync function runs on a
+        worker thread.
+        """
+        kwargs = arguments | (arguments_to_pass_directly or {})
+        if fn_is_async:
+            return await fn(**kwargs)
+        return await anyio.to_thread.run_sync(functools.partial(fn, **kwargs))
+
+    @deprecated(
+        "FuncMetadata.call_fn_with_arg_validation() is deprecated and will be removed in 3.0; "
+        "call validate_arguments() and then call_fn() instead.",
+        category=MCPDeprecationWarning,
+    )
     async def call_fn_with_arg_validation(
         self,
         fn: Callable[..., Any | Awaitable[Any]],
@@ -133,25 +156,12 @@ class FuncMetadata(BaseModel):
         arguments_to_pass_directly: dict[str, Any] | None,
         pre_validated: dict[str, Any] | None = None,
     ) -> Any:
-        """Call the given function with arguments validated and injected.
+        """Validate `arguments_to_validate` (unless `pre_validated` is given) and call the function.
 
-        Arguments are first attempted to be parsed from JSON, then validated against
-        the argument model, before being passed to the function. Pass `pre_validated`
-        (the output of `validate_arguments`) to reuse an earlier validation pass -
-        validating twice can re-run `default_factory`/stateful validators and hand the
-        function different values than a caller already observed.
+        Deprecated: call `validate_arguments` and then `call_fn`.
         """
-        # Copy so a caller-provided `pre_validated` dict is never mutated in place.
-        arguments_parsed_dict = dict(
-            pre_validated if pre_validated is not None else self.validate_arguments(arguments_to_validate)
-        )
-
-        arguments_parsed_dict |= arguments_to_pass_directly or {}
-
-        if fn_is_async:
-            return await fn(**arguments_parsed_dict)
-        else:
-            return await anyio.to_thread.run_sync(functools.partial(fn, **arguments_parsed_dict))
+        arguments = pre_validated if pre_validated is not None else self.validate_arguments(arguments_to_validate)
+        return await self.call_fn(fn, fn_is_async, arguments, arguments_to_pass_directly)
 
     def convert_result(self, result: Any) -> CallToolResult | InputRequiredResult:
         """Convert a function call result into a `CallToolResult`.
@@ -172,7 +182,7 @@ class FuncMetadata(BaseModel):
         # A schema published without a model (hand-built metadata) is advertised but not validated here.
         output_model = self.output_model if self.output_schema is not None else None
         if isinstance(result, CallToolResult):
-            if output_model is not None:
+            if output_model is not None and not result.is_error:
                 self._output_adapter(output_model).validate_python(result.structured_content)
             return result
 
@@ -224,8 +234,10 @@ class FuncMetadata(BaseModel):
             if isinstance(data_value, str) and field_info.annotation is not str:
                 try:
                     pre_parsed = json.loads(data_value)
-                except json.JSONDecodeError:
-                    continue  # Not JSON - skip
+                except (ValueError, RecursionError):
+                    # Not JSON, or JSON the parser refuses (over-long integers, deep
+                    # nesting): leave the string for validation to accept or reject.
+                    continue
                 if isinstance(pre_parsed, str | int | float):
                     # This is likely that the raw value is e.g. `"hello"` which we
                     # Should really be parsed as '"hello"' in Python - but if we parse