rebase on lowlevel schema validation, address comments (wip)

Author: bhosmer-ant

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

@@ -4,16 +4,13 @@
 import inspect
 from collections.abc import Callable, Sequence
 from functools import cached_property
-from itertools import chain
 from typing import TYPE_CHECKING, Any, get_origin
 
-import pydantic_core
 from pydantic import BaseModel, Field
 
 from mcp.server.fastmcp.exceptions import ToolError
 from mcp.server.fastmcp.utilities.func_metadata import FuncMetadata, func_metadata
-from mcp.server.fastmcp.utilities.types import Image
-from mcp.types import ContentBlock, TextContent, ToolAnnotations
+from mcp.types import ContentBlock, ToolAnnotations
 
 if TYPE_CHECKING:
     from mcp.server.fastmcp.server import Context
@@ -112,28 +109,8 @@ async def run(
             raise ToolError(f"Error executing tool {self.name}: {e}") from e
 
     def convert_result(self, result: Any) -> Sequence[ContentBlock] | dict[str, Any]:
-        """Validate tool result and convert to appropriate output format."""
-        output_model = self.fn_metadata.output_model
-        if output_model:
-            # This will raise a ToolError if validation fails
-            return self.fn_metadata.to_validated_dict(result)
-        else:
-            if result is None:
-                return []
-
-            if isinstance(result, ContentBlock):
-                return [result]
-
-            if isinstance(result, Image):
-                return [result.to_image_content()]
-
-            if isinstance(result, list | tuple):
-                return list(chain.from_iterable(self.convert_result(item) for item in result))  # type: ignore[reportUnknownVariableType]
-
-            if not isinstance(result, str):
-                result = pydantic_core.to_json(result, fallback=str, indent=2).decode()
-
-            return [TextContent(type="text", text=result)]
+        """convert to appropriate output format."""
+        return self.fn_metadata.convert_result(result)
 
 
 def _is_async_callable(obj: Any) -> bool:

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

@@ -2,16 +2,10 @@
 import json
 from collections.abc import Awaitable, Callable, Sequence
 from dataclasses import asdict, is_dataclass
-from typing import (
-    Annotated,
-    Any,
-    ForwardRef,
-    Literal,
-    get_args,
-    get_origin,
-    get_type_hints,
-)
+from itertools import chain
+from typing import Annotated, Any, ForwardRef, Literal, get_args, get_origin, get_type_hints
 
+import pydantic_core
 from pydantic import (
     BaseModel,
     ConfigDict,
@@ -24,12 +18,14 @@
 from pydantic.fields import FieldInfo
 from pydantic_core import PydanticUndefined
 
-from mcp.server.fastmcp.exceptions import InvalidSignature, ToolError
+from mcp.server.fastmcp.exceptions import InvalidSignature
 from mcp.server.fastmcp.utilities.logging import get_logger
+from mcp.server.fastmcp.utilities.types import Image
+from mcp.types import ContentBlock, TextContent
 
 logger = get_logger(__name__)
 
-OutputConversion = Literal["none", "wrapped", "namedtuple", "class"]
+OutputConversion = Literal["none", "basemodel", "wrapped", "namedtuple", "class"]
 
 
 class ArgModelBase(BaseModel):
@@ -54,16 +50,13 @@ class FuncMetadata(BaseModel):
     arg_model: Annotated[type[ArgModelBase], WithJsonSchema(None)]
     output_model: Annotated[type[BaseModel], WithJsonSchema(None)] | None = None
     output_conversion: OutputConversion = "none"
-    # We can add things in the future like
-    #  - Maybe some args are excluded from attempting to parse from JSON
-    #  - Maybe some args are special (like context) for dependency injection
 
     async def call_fn_with_arg_validation(
         self,
-        fn: Callable[..., Any] | Awaitable[Any],
+        fn: Callable[..., Any | Awaitable[Any]],
         fn_is_async: bool,
         arguments_to_validate: dict[str, Any],
-        arguments_to_pass_directly: dict[str, Any] | None,
+        arguments_to_pass_directly: dict[str, Any] | None = None,
     ) -> Any:
         """Call the given function with arguments validated and injected.
 
@@ -75,36 +68,62 @@ async def call_fn_with_arg_validation(
         arguments_parsed_dict = arguments_parsed_model.model_dump_one_level()
 
         arguments_parsed_dict |= arguments_to_pass_directly or {}
-
         if fn_is_async:
-            if isinstance(fn, Awaitable):
-                return await fn
             return await fn(**arguments_parsed_dict)
-        if isinstance(fn, Callable):
-            return fn(**arguments_parsed_dict)
-        raise TypeError("fn must be either Callable or Awaitable")
+        return fn(**arguments_parsed_dict)
+
+    async def call_fn(
+        self,
+        fn: Callable[..., Any | Awaitable[Any]],
+        fn_is_async: bool,
+        args: dict[str, Any],
+        implicit_args: dict[str, Any] | None = None,
+    ) -> Any:
+        parsed_args = self.arg_model.model_construct(**args).model_dump_one_level()
+        kwargs = parsed_args | (implicit_args or {})
+        if fn_is_async:
+            return await fn(**kwargs)
+        else:
+            return fn(**kwargs)
 
-    def to_validated_dict(self, result: Any) -> dict[str, Any]:
-        """Validate and convert the result to a dict after validation."""
-        if self.output_model is None:
-            raise ValueError("No output model to validate against")
+    def convert_result(self, result: Any) -> Sequence[ContentBlock] | dict[str, Any]:
+        """convert result to dict"""
+        if self.output_model:
+            return self._convert_structured_result(result)
+        else:
+            return self._convert_unstructured_result(result)
 
+    def _convert_structured_result(self, result: Any) -> dict[str, Any]:
         match self.output_conversion:
+            case "none":
+                return result
+            case "basemodel":
+                return result.model_dump()
             case "wrapped":
-                converted = _convert_wrapped_result(result)
+                return {"result": result}
             case "namedtuple":
-                converted = _convert_namedtuple_result(result)
+                return result._asdict()
             case "class":
-                converted = _convert_class_result(result)
-            case "none":
-                converted = result
+                if is_dataclass(result) and not isinstance(result, type):
+                    return asdict(result)
+                return dict(vars(result))
 
-        try:
-            validated = self.output_model.model_validate(converted)
-        except Exception as e:
-            raise ToolError(f"Output validation failed: {e}") from e
+    def _convert_unstructured_result(self, result: Any) -> Sequence[ContentBlock]:
+        if result is None:
+            return []
 
-        return validated.model_dump()
+        if isinstance(result, ContentBlock):
+            return [result]
+
+        if isinstance(result, Image):
+            return [result.to_image_content()]
+
+        if isinstance(result, list | tuple):
+            return list(chain.from_iterable(self._convert_unstructured_result(item) for item in result))  # type: ignore
+
+        if not isinstance(result, str):
+            result = pydantic_core.to_json(result, fallback=str, indent=2).decode()
+        return [TextContent(type="text", text=result)]
 
     def pre_parse_json(self, data: dict[str, Any]) -> dict[str, Any]:
         """Pre-parse data from JSON.
@@ -177,6 +196,7 @@ def func_metadata(
         A FuncMetadata object containing:
         - arg_model: A pydantic model representing the function's arguments
         - output_model: A pydantic model for the return type (if structured_output=True)
+        - output_conversion: Records how function output should be converted before returning.
     """
     sig = _get_typed_signature(func)
     params = sig.parameters
@@ -236,7 +256,7 @@ def func_metadata(
         elif isinstance(annotation, type):
             if issubclass(annotation, BaseModel):
                 output_model = annotation
-                output_conversion = "none"
+                output_conversion = "basemodel"
             elif _is_typeddict(annotation):
                 output_model = _create_model_from_typeddict(annotation, globalns)
                 output_conversion = "none"
@@ -338,13 +358,6 @@ def _create_model_from_class(cls: type[Any], globalns: dict[str, Any]) -> type[B
     return create_model(cls.__name__, **model_fields, __base__=BaseModel)
 
 
-def _convert_class_result(result: Any) -> dict[str, Any]:
-    if is_dataclass(result) and not isinstance(result, type):
-        return asdict(result)
-
-    return dict(vars(result))
-
-
 def _create_model_from_typeddict(td_type: type[Any], globalns: dict[str, Any]) -> type[BaseModel]:
     """Create a Pydantic model from a TypedDict.
 
@@ -387,10 +400,6 @@ def _create_model_from_namedtuple(nt_type: type[Any], globalns: dict[str, Any])
     return create_model(nt_type.__name__, **model_fields, __base__=BaseModel)
 
 
-def _convert_namedtuple_result(result: Any) -> dict[str, Any]:
-    return result._asdict()
-
-
 def _create_wrapped_model(func_name: str, annotation: Any, field_info: FieldInfo) -> type[BaseModel]:
     """Create a model that wraps a type in a 'result' field.
 
@@ -405,10 +414,6 @@ def _create_wrapped_model(func_name: str, annotation: Any, field_info: FieldInfo
     return create_model(model_name, result=(annotation, field_info), __base__=BaseModel)
 
 
-def _convert_wrapped_result(result: Any) -> dict[str, Any]:
-    return {"result": result}
-
-
 def _create_dict_model(func_name: str, dict_annotation: Any) -> type[BaseModel]:
     """Create a RootModel for dict[str, T] types."""