Replace custom parsing with TypeAdapter

src/fastmcp/server/openapi.py

@@ -18,7 +18,6 @@ from fastmcp.resources import Resource, ResourceTemplate
 from fastmcp.server.server import FastMCP
 from fastmcp.tools.tool import Tool, _convert_to_content
 from fastmcp.utilities import openapi
-from fastmcp.utilities.func_metadata import func_metadata
 from fastmcp.utilities.logging import get_logger
 from fastmcp.utilities.openapi import (
     _combine_schemas,
@@ -123,7 +122,6 @@ class OpenAPITool(Tool):
         name: str,
         description: str,
         parameters: dict[str, Any],
-        fn_metadata: Any,
         is_async: bool = True,
         tags: set[str] = set(),
         timeout: float | None = None,
@@ -135,7 +133,6 @@ class OpenAPITool(Tool):
             description=description,
             parameters=parameters,
             fn=self._execute_request,  # We'll use an instance method instead of a global function
-            fn_metadata=fn_metadata,
             is_async=is_async,
             context_kwarg="context",  # Default context keyword argument
             tags=tags,
@@ -553,7 +550,6 @@ class FastMCPOpenAPI(FastMCP):
             name=tool_name,
             description=enhanced_description,
             parameters=combined_schema,
-            fn_metadata=func_metadata(_openapi_passthrough),
             is_async=True,
             tags=set(route.tags or []),
             timeout=self._timeout,

src/fastmcp/server/proxy.py

@@ -24,7 +24,6 @@ from fastmcp.resources import Resource, ResourceTemplate
 from fastmcp.server.context import Context
 from fastmcp.server.server import FastMCP
 from fastmcp.tools.tool import Tool
-from fastmcp.utilities.func_metadata import func_metadata
 from fastmcp.utilities.logging import get_logger
 
 if TYPE_CHECKING:
@@ -53,7 +52,6 @@ class ProxyTool(Tool):
             description=tool.description,
             parameters=tool.inputSchema,
             fn=_proxy_passthrough,
-            fn_metadata=func_metadata(_proxy_passthrough),
             is_async=True,
         )
 

src/fastmcp/tools/tool.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import inspect
+import json
 from collections.abc import Callable
 from typing import TYPE_CHECKING, Annotated, Any
 
@@ -10,11 +11,12 @@ from mcp.types import Tool as MCPTool
 from pydantic import BaseModel, BeforeValidator, Field
 
 from fastmcp.exceptions import ToolError
-from fastmcp.utilities.func_metadata import FuncMetadata, func_metadata
+from fastmcp.utilities.json_schema import prune_params
 from fastmcp.utilities.logging import get_logger
 from fastmcp.utilities.types import (
     Image,
     _convert_set_defaults,
+    get_cached_typeadapter,
     is_class_member_of_type,
 )
 
@@ -38,10 +40,6 @@ class Tool(BaseModel):
     name: str = Field(description="Name of the tool")
     description: str = Field(description="Description of what the tool does")
     parameters: dict[str, Any] = Field(description="JSON schema for tool parameters")
-    fn_metadata: FuncMetadata = Field(
-        description="Metadata about the function including a pydantic model for tool"
-        " arguments"
-    )
     is_async: bool = Field(description="Whether the tool is async")
     context_kwarg: str | None = Field(
         None, description="Name of the kwarg that should receive context"
@@ -78,35 +76,26 @@ class Tool(BaseModel):
         func_doc = description or fn.__doc__ or ""
         is_async = inspect.iscoroutinefunction(fn)
 
+        if inspect.ismethod(fn) and hasattr(fn, "__func__"):
+            sig = inspect.signature(fn.__func__)
+        else:
+            sig = inspect.signature(fn)
         if context_kwarg is None:
-            if inspect.ismethod(fn) and hasattr(fn, "__func__"):
-                sig = inspect.signature(fn.__func__)
-            else:
-                sig = inspect.signature(fn)
             for param_name, param in sig.parameters.items():
                 if is_class_member_of_type(param.annotation, Context):
                     context_kwarg = param_name
                     break
 
-        # Use callable typing to ensure fn is treated as a callable despite being a classmethod
-        fn_callable: Callable[..., Any] = fn
-        func_arg_metadata = func_metadata(
-            fn_callable,
-            skip_names=[context_kwarg] if context_kwarg is not None else [],
-        )
-        try:
-            parameters = func_arg_metadata.arg_model.model_json_schema()
-        except Exception as e:
-            raise TypeError(
-                f'Unable to parse parameters for function "{fn.__name__}": {e}'
-            ) from e
+        type_adapter = get_cached_typeadapter(fn)
+        schema = type_adapter.json_schema()
+        if context_kwarg:
+            schema = prune_params(schema, params=[context_kwarg])
 
         return cls(
-            fn=fn_callable,
+            fn=fn,
             name=func_name,
             description=func_doc,
-            parameters=parameters,
-            fn_metadata=func_arg_metadata,
+            parameters=schema,
             is_async=is_async,
             context_kwarg=context_kwarg,
             tags=tags or set(),
@@ -121,17 +110,30 @@ class Tool(BaseModel):
     ) -> list[TextContent | ImageContent | EmbeddedResource]:
         """Run the tool with arguments."""
         try:
-            pass_args = (
-                {self.context_kwarg: context}
-                if self.context_kwarg is not None
-                else None
-            )
-            result = await self.fn_metadata.call_fn_with_arg_validation(
-                fn=self.fn,
-                fn_is_async=self.is_async,
-                arguments_to_validate=arguments,
-                arguments_to_pass_directly=pass_args,
+            injected_args = (
+                {self.context_kwarg: context} if self.context_kwarg is not None else {}
             )
+
+            parsed_args = arguments.copy()
+
+            # Pre-parse data from JSON in order to handle cases like `["a", "b", "c"]`
+            # being passed in as JSON inside a string rather than an actual list.
+            #
+            # Claude desktop is prone to this - in fact it seems incapable of NOT doing
+            # this. For sub-models, it tends to pass dicts (JSON objects) as JSON strings,
+            # which can be pre-parsed here.
+            for param_name in self.parameters["properties"]:
+                if isinstance(parsed_args.get(param_name, None), str):
+                    try:
+                        parsed_args[param_name] = json.loads(parsed_args[param_name])
+                    except Exception:
+                        pass
+
+            type_adapter = get_cached_typeadapter(self.fn)
+            result = type_adapter.validate_python(parsed_args | injected_args)
+            if inspect.isawaitable(result):
+                result = await result
+
             return _convert_to_content(result, serializer=self.serializer)
         except Exception as e:
             raise ToolError(f"Error executing tool {self.name}: {e}") from e

src/fastmcp/utilities/func_metadata.py

@@ -1,229 +0,0 @@
-import inspect
-import json
-from collections.abc import Awaitable, Callable, Sequence
-from typing import (
-    Annotated,
-    Any,
-    ForwardRef,
-)
-
-from pydantic import (
-    BaseModel,
-    ConfigDict,
-    Field,
-    TypeAdapter,
-    ValidationError,
-    WithJsonSchema,
-    create_model,
-)
-from pydantic._internal._typing_extra import eval_type_backport
-from pydantic.fields import FieldInfo
-from pydantic_core import PydanticUndefined
-
-from fastmcp.exceptions import InvalidSignature
-from fastmcp.utilities.logging import get_logger
-
-logger = get_logger(__name__)
-
-
-class ArgModelBase(BaseModel):
-    """A model representing the arguments to a function."""
-
-    def model_dump_one_level(self) -> dict[str, Any]:
-        """Return a dict of the model's fields, one level deep.
-
-        That is, sub-models etc are not dumped - they are kept as pydantic models.
-        """
-        kwargs: dict[str, Any] = {}
-        for field_name in self.__class__.model_fields.keys():
-            kwargs[field_name] = getattr(self, field_name)
-        return kwargs
-
-    model_config = ConfigDict(
-        arbitrary_types_allowed=True,
-    )
-
-
-class FuncMetadata(BaseModel):
-    arg_model: Annotated[type[ArgModelBase], WithJsonSchema(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_is_async: bool,
-        arguments_to_validate: dict[str, Any],
-        arguments_to_pass_directly: dict[str, Any] | None,
-    ) -> Any:
-        """Call the given function with arguments validated and injected.
-
-        Arguments are first attempted to be parsed from JSON, then validated against
-        the argument model, before being passed to the function.
-        """
-        arguments_pre_parsed = self.pre_parse_json(arguments_to_validate)
-        arguments_parsed_model = self.arg_model.model_validate(arguments_pre_parsed)
-        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")
-
-    def pre_parse_json(self, data: dict[str, Any]) -> dict[str, Any]:
-        """Pre-parse data from JSON.
-
-        Return a dict with same keys as input but with values parsed from JSON
-        if appropriate.
-
-        This is to handle cases like `["a", "b", "c"]` being passed in as JSON inside
-        a string rather than an actual list. Claude desktop is prone to this - in fact
-        it seems incapable of NOT doing this. For sub-models, it tends to pass
-        dicts (JSON objects) as JSON strings, which can be pre-parsed here.
-        """
-        new_data = data.copy()  # Shallow copy
-        for field_name, field_info in self.arg_model.model_fields.items():
-            if field_name not in data.keys():
-                continue
-            if isinstance(data[field_name], str):
-                try:
-                    pre_parsed = json.loads(data[field_name])
-
-                    # Check if the pre_parsed value is valid for the field
-                    validator = TypeAdapter(field_info.annotation)
-                    validator.validate_python(pre_parsed)
-                except (json.JSONDecodeError, ValidationError):
-                    continue  # Not JSON or invalid for the field
-                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
-                    # it as JSON it'll turn into just 'hello'. So we skip it.
-                    continue
-                new_data[field_name] = pre_parsed
-        assert new_data.keys() == data.keys()
-        return new_data
-
-    model_config = ConfigDict(
-        arbitrary_types_allowed=True,
-    )
-
-
-def func_metadata(
-    func: Callable[..., Any], skip_names: Sequence[str] = ()
-) -> FuncMetadata:
-    """Given a function, return metadata including a pydantic model representing its
-    signature.
-
-    The use case for this is
-    ```
-    meta = func_to_pyd(func)
-    validated_args = meta.arg_model.model_validate(some_raw_data_dict)
-    return func(**validated_args.model_dump_one_level())
-    ```
-
-    **critically** it also provides pre-parse helper to attempt to parse things from
-    JSON.
-
-    Args:
-        func: The function to convert to a pydantic model
-        skip_names: A list of parameter names to skip. These will not be included in
-            the model.
-    Returns:
-        A pydantic model representing the function's signature.
-    """
-    if isinstance(func, classmethod):
-        sig = _get_typed_signature(func.__func__)
-    else:
-        sig = _get_typed_signature(func)
-    params = sig.parameters
-    dynamic_pydantic_model_params: dict[str, Any] = {}
-    globalns = getattr(func, "__globals__", {})
-    for param in params.values():
-        if param.name.startswith("_"):
-            raise InvalidSignature(
-                f"Parameter {param.name} of {func.__name__} cannot start with '_'"
-            )
-        if param.name in skip_names:
-            continue
-        annotation = param.annotation
-
-        # `x: None` / `x: None = None`
-        if annotation is None:
-            annotation = Annotated[
-                None,
-                Field(
-                    default=param.default
-                    if param.default is not inspect.Parameter.empty
-                    else PydanticUndefined
-                ),
-            ]
-
-        # Untyped field
-        if annotation is inspect.Parameter.empty:
-            annotation = Annotated[
-                Any,
-                Field(),
-                # 🤷
-                WithJsonSchema({"title": param.name, "type": "string"}),
-            ]
-
-        field_info = FieldInfo.from_annotated_attribute(
-            _get_typed_annotation(annotation, globalns),
-            param.default
-            if param.default is not inspect.Parameter.empty
-            else PydanticUndefined,
-        )
-        dynamic_pydantic_model_params[param.name] = (field_info.annotation, field_info)
-        continue
-
-    arguments_model = create_model(
-        f"{func.__name__}Arguments",
-        **dynamic_pydantic_model_params,
-        __base__=ArgModelBase,
-    )
-    resp = FuncMetadata(arg_model=arguments_model)
-    return resp
-
-
-def _get_typed_annotation(annotation: Any, globalns: dict[str, Any]) -> Any:
-    def try_eval_type(
-        value: Any, globalns: dict[str, Any], localns: dict[str, Any]
-    ) -> tuple[Any, bool]:
-        try:
-            return eval_type_backport(value, globalns, localns), True
-        except NameError:
-            return value, False
-
-    if isinstance(annotation, str):
-        annotation = ForwardRef(annotation)
-        annotation, status = try_eval_type(annotation, globalns, globalns)
-
-        # This check and raise could perhaps be skipped, and we (FastMCP) just call
-        # model_rebuild right before using it 🤷
-        if status is False:
-            raise InvalidSignature(f"Unable to evaluate type annotation {annotation}")
-
-    return annotation
-
-
-def _get_typed_signature(call: Callable[..., Any]) -> inspect.Signature:
-    """Get function signature while evaluating forward references"""
-    signature = inspect.signature(call)
-    globalns = getattr(call, "__globals__", {})
-    typed_params = [
-        inspect.Parameter(
-            name=param.name,
-            kind=param.kind,
-            default=param.default,
-            annotation=_get_typed_annotation(param.annotation, globalns),
-        )
-        for param in signature.parameters.values()
-    ]
-    typed_signature = inspect.Signature(typed_params)
-    return typed_signature

src/fastmcp/utilities/json_schema.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+import copy
+from collections.abc import Mapping, Sequence
+
+
+def _prune_param(schema: dict, param: str) -> dict:
+    """Return a new schema with *param* removed from `properties`, `required`,
+    and (if no longer referenced) `$defs`.
+    """
+
+    # ── 1. drop from properties/required ──────────────────────────────
+    props = schema.get("properties", {})
+    removed = props.pop(param, None)
+    if removed is None:  # nothing to do
+        return schema
+    # Keep empty properties object rather than removing it entirely
+    schema["properties"] = props
+    if param in schema.get("required", []):
+        schema["required"].remove(param)
+        if not schema["required"]:
+            schema.pop("required")
+
+    # ── 2. collect all remaining local $ref targets ───────────────────
+    used_defs: set[str] = set()
+
+    def walk(node: object) -> None:  # depth-first traversal
+        if isinstance(node, Mapping):
+            ref = node.get("$ref")
+            if isinstance(ref, str) and ref.startswith("#/$defs/"):
+                used_defs.add(ref.split("/")[-1])
+            for v in node.values():
+                walk(v)
+        elif isinstance(node, Sequence) and not isinstance(node, str | bytes):
+            for v in node:
+                walk(v)
+
+    walk(schema)
+
+    # ── 3. remove orphaned definitions ────────────────────────────────
+    defs = schema.get("$defs", {})
+    for def_name in list(defs):
+        if def_name not in used_defs:
+            defs.pop(def_name)
+    if not defs:
+        schema.pop("$defs", None)
+
+    return schema
+
+
+def prune_params(schema: dict, params: list[str]) -> dict:
+    """
+    Remove the given parameters from the schema.
+
+    """
+    schema = copy.deepcopy(schema)
+    for param in params:
+        schema = _prune_param(schema, param=param)
+    return schema

src/fastmcp/utilities/types.py

@@ -1,15 +1,28 @@
 """Common types used across FastMCP."""
 
 import base64
+from functools import lru_cache
 from pathlib import Path
 from types import UnionType
 from typing import Annotated, TypeVar, Union, get_args, get_origin
 
 from mcp.types import ImageContent
+from pydantic import TypeAdapter
 
 T = TypeVar("T")
 
 
+@lru_cache(maxsize=5000)
+def get_cached_typeadapter(cls: T) -> TypeAdapter[T]:
+    """
+    TypeAdapters are heavy objects, and in an application context we'd typically
+    create them once in a global scope and reuse them as often as possible.
+    However, this isn't feasible for user-generated functions. Instead, we use a
+    cache to minimize the cost of creating them as much as possible.
+    """
+    return TypeAdapter(cls)
+
+
 def issubclass_safe(cls: type, base: type) -> bool:
     """Check if cls is a subclass of base, even if cls is a type variable."""
     try:

tests/server/test_server_interactions.py

@@ -460,7 +460,7 @@ class TestToolParameters:
             pass
 
         async with Client(mcp) as client:
-            with pytest.raises(ClientError, match="Field required"):
+            with pytest.raises(ClientError, match="Missing required argument"):
                 await client.call_tool("analyze", {})
 
     async def test_literal_type_validation_error(self):
@@ -537,7 +537,7 @@ class TestToolParameters:
             assert isinstance(result[0], TextContent)
             assert result[0].text == "1.0"
 
-            with pytest.raises(ClientError, match="2 validation errors for analyze"):
+            with pytest.raises(ClientError, match="2 validation errors"):
                 await client.call_tool("analyze", {"x": "not a number"})
 
     async def test_path_type(self):

tests/tools/test_tool.py

@@ -0,0 +1,206 @@
+import pytest
+from mcp.types import ImageContent, TextContent
+from pydantic import BaseModel
+
+from fastmcp import Image
+from fastmcp.tools.tool import Tool
+
+
+class TestToolFromFunction:
+    def test_basic_function(self):
+        """Test registering and running a basic function."""
+
+        def add(a: int, b: int) -> int:
+            """Add two numbers."""
+            return a + b
+
+        tool = Tool.from_function(add)
+
+        assert tool.name == "add"
+        assert tool.description == "Add two numbers."
+        assert tool.is_async is False
+        assert tool.parameters["properties"]["a"]["type"] == "integer"
+        assert tool.parameters["properties"]["b"]["type"] == "integer"
+
+    async def test_async_function(self):
+        """Test registering and running an async function."""
+
+        async def fetch_data(url: str) -> str:
+            """Fetch data from URL."""
+            return f"Data from {url}"
+
+        tool = Tool.from_function(fetch_data)
+
+        assert tool.name == "fetch_data"
+        assert tool.description == "Fetch data from URL."
+        assert tool.is_async is True
+        assert tool.parameters["properties"]["url"]["type"] == "string"
+
+    def test_pydantic_model_function(self):
+        """Test registering a function that takes a Pydantic model."""
+
+        class UserInput(BaseModel):
+            name: str
+            age: int
+
+        def create_user(user: UserInput, flag: bool) -> dict:
+            """Create a new user."""
+            return {"id": 1, **user.model_dump()}
+
+        tool = Tool.from_function(create_user)
+
+        assert tool.name == "create_user"
+        assert tool.description == "Create a new user."
+        assert tool.is_async is False
+        assert "name" in tool.parameters["$defs"]["UserInput"]["properties"]
+        assert "age" in tool.parameters["$defs"]["UserInput"]["properties"]
+        assert "flag" in tool.parameters["properties"]
+
+    async def test_tool_with_image_return(self):
+        def image_tool(data: bytes) -> Image:
+            return Image(data=data)
+
+        tool = Tool.from_function(image_tool)
+
+        result = await tool.run({"data": "test.png"})
+        assert tool.parameters["properties"]["data"]["type"] == "string"
+        assert isinstance(result[0], ImageContent)
+
+    def test_add_invalid_tool(self):
+        with pytest.raises(AttributeError):
+            Tool.from_function(1)  # type: ignore
+
+    def test_add_lambda(self):
+        tool = Tool.from_function(lambda x: x, name="my_tool")
+        assert tool.name == "my_tool"
+
+    def test_add_lambda_with_no_name(self):
+        with pytest.raises(
+            ValueError, match="You must provide a name for lambda functions"
+        ):
+            Tool.from_function(lambda x: x)
+
+    def test_no_private_arguments(self):
+        def add(_a: int, _b: int) -> int:
+            """Add two numbers."""
+            return _a + _b
+
+        tool = Tool.from_function(add)
+        assert tool.parameters["properties"]["_a"]["type"] == "integer"
+        assert tool.parameters["properties"]["_b"]["type"] == "integer"
+
+
+class TestToolJsonParsing:
+    """Tests for Tool's JSON pre-parsing functionality."""
+
+    async def test_json_string_arguments(self):
+        """Test that JSON string arguments are parsed and validated correctly"""
+
+        def simple_func(x: int, y: list[str]) -> str:
+            return f"{x}-{','.join(y)}"
+
+        # Create a tool to use its JSON pre-parsing logic
+        tool = Tool.from_function(simple_func)
+
+        # Prepare arguments where some are JSON strings
+        json_args = {
+            "x": 1,
+            "y": '["a", "b", "c"]',  # JSON string
+        }
+
+        # Run the tool which will do JSON parsing
+        result = await tool.run(json_args)
+        assert len(result) == 1
+        assert isinstance(result[0], TextContent)
+        assert result[0].text == "1-a,b,c"
+
+    async def test_str_vs_list_str(self):
+        """Test handling of string vs list[str] type annotations."""
+
+        def func_with_str_types(str_or_list: str | list[str]) -> str | list[str]:
+            return str_or_list
+
+        tool = Tool.from_function(func_with_str_types)
+
+        # Test regular string input (should remain a string)
+        result = await tool.run({"str_or_list": "hello"})
+        assert len(result) == 1
+        assert isinstance(result[0], TextContent)
+        assert result[0].text == "hello"
+
+        # Test JSON string input (should be parsed as a string)
+        result = await tool.run({"str_or_list": '"hello"'})
+        assert len(result) == 1
+        assert isinstance(result[0], TextContent)
+        assert result[0].text == "hello"
+
+        # Test JSON list input (should be parsed as a list)
+        result = await tool.run({"str_or_list": '["hello", "world"]'})
+        assert len(result) == 1
+        assert isinstance(result[0], TextContent)
+
+        # The exact formatting might vary, so we just check that it contains the key elements
+        text_without_whitespace = result[0].text.replace(" ", "").replace("\n", "")
+        assert "hello" in text_without_whitespace
+        assert "world" in text_without_whitespace
+        assert "[" in text_without_whitespace
+        assert "]" in text_without_whitespace
+
+    async def test_keep_str_as_str(self):
+        """Test that string arguments are kept as strings when they're not valid JSON"""
+
+        def func_with_str_types(string: str) -> str:
+            return string
+
+        tool = Tool.from_function(func_with_str_types)
+
+        # Invalid JSON should remain a string
+        invalid_json = "{'nice to meet you': 'hello', 'goodbye': 5}"
+        result = await tool.run({"string": invalid_json})
+        assert len(result) == 1
+        assert isinstance(result[0], TextContent)
+        assert result[0].text == invalid_json
+
+    async def test_keep_str_union_as_str(self):
+        """Test that string arguments are kept as strings when parsing would create an invalid value"""
+
+        def func_with_str_types(
+            string: str | dict[int, str] | None,
+        ) -> str | dict[int, str] | None:
+            return string
+
+        tool = Tool.from_function(func_with_str_types)
+
+        # Invalid JSON for the union type should remain a string
+        invalid_json = "{'nice to meet you': 'hello', 'goodbye': 5}"
+        result = await tool.run({"string": invalid_json})
+        assert len(result) == 1
+        assert isinstance(result[0], TextContent)
+        assert result[0].text == invalid_json
+
+    async def test_complex_type_validation(self):
+        """Test that parsed JSON is validated against complex types"""
+
+        class SomeModel(BaseModel):
+            x: int
+            y: dict[int, str]
+
+        def func_with_complex_type(data: SomeModel) -> SomeModel:
+            return data
+
+        tool = Tool.from_function(func_with_complex_type)
+
+        # Valid JSON for the model
+        valid_json = '{"x": 1, "y": {"1": "hello"}}'
+        result = await tool.run({"data": valid_json})
+        assert len(result) == 1
+        assert isinstance(result[0], TextContent)
+        assert '"x": 1' in result[0].text
+        assert '"y": {' in result[0].text
+        assert '"1": "hello"' in result[0].text
+
+        # Invalid JSON for the model (y has string keys, not int keys)
+        # Should throw a validation error
+        invalid_json = '{"x": 1, "y": {"invalid": "hello"}}'
+        with pytest.raises(Exception):
+            await tool.run({"data": invalid_json})

tests/tools/test_tool_manager.py

@@ -369,7 +369,7 @@ class TestCallTools:
         assert isinstance(result, list)
         assert len(result) == 1
         assert isinstance(result[0], TextContent)
-        assert result[0].text == '"a"'
+        assert result[0].text == "a"
 
     async def test_call_tool_with_complex_model(self):
         class MyShrimpTank(BaseModel):
@@ -379,7 +379,7 @@ class TestCallTools:
             shrimp: list[Shrimp]
             x: None
 
-        def name_shrimp(tank: MyShrimpTank, ctx: Context) -> list[str]:
+        def name_shrimp(tank: MyShrimpTank, ctx: Context | None) -> list[str]:
             return [x.name for x in tank.shrimp]
 
         manager = ToolManager()
@@ -449,7 +449,24 @@ class TestToolSchema:
         tool = manager.add_tool_from_fn(something)
         assert "ctx" not in json.dumps(tool.parameters)
         assert "Context" not in json.dumps(tool.parameters)
-        assert "ctx" not in tool.fn_metadata.arg_model.model_fields
+
+    async def test_optional_context_arg_excluded_from_schema(self):
+        def something(a: int, ctx: Context | None) -> int:
+            return a
+
+        manager = ToolManager()
+        tool = manager.add_tool_from_fn(something)
+        assert "ctx" not in json.dumps(tool.parameters)
+        assert "Context" not in json.dumps(tool.parameters)
+
+    async def test_annotated_context_arg_excluded_from_schema(self):
+        def something(a: int, ctx: Annotated[Context | int | None, "ctx"]) -> int:
+            return a
+
+        manager = ToolManager()
+        tool = manager.add_tool_from_fn(something)
+        assert "ctx" not in json.dumps(tool.parameters)
+        assert "Context" not in json.dumps(tool.parameters)
 
 
 class TestContextHandling:

tests/utilities/test_func_metadata.py

@@ -1,481 +0,0 @@
-from typing import Annotated
-
-import annotated_types
-import pytest
-from pydantic import BaseModel, Field
-
-from fastmcp.utilities.func_metadata import func_metadata
-
-
-class SomeInputModelA(BaseModel):
-    pass
-
-
-class SomeInputModelB(BaseModel):
-    class InnerModel(BaseModel):
-        x: int
-
-    how_many_shrimp: Annotated[int, Field(description="How many shrimp in the tank???")]
-    ok: InnerModel
-    y: None
-
-
-def complex_arguments_fn(
-    an_int: int,
-    must_be_none: None,
-    must_be_none_dumb_annotation: Annotated[None, "blah"],
-    list_of_ints: list[int],
-    # list[str] | str is an interesting case because if it comes in as JSON like
-    # "[\"a\", \"b\"]" then it will be naively parsed as a string.
-    list_str_or_str: list[str] | str,
-    an_int_annotated_with_field: Annotated[
-        int, Field(description="An int with a field")
-    ],
-    an_int_annotated_with_field_and_others: Annotated[
-        int,
-        str,  # Should be ignored, really
-        Field(description="An int with a field"),
-        annotated_types.Gt(1),
-    ],
-    an_int_annotated_with_junk: Annotated[
-        int,
-        "123",
-        456,
-    ],
-    field_with_default_via_field_annotation_before_nondefault_arg: Annotated[
-        int, Field(1)
-    ],
-    unannotated,
-    my_model_a: SomeInputModelA,
-    my_model_a_forward_ref: "SomeInputModelA",
-    my_model_b: SomeInputModelB,
-    an_int_annotated_with_field_default: Annotated[
-        int,
-        Field(1, description="An int with a field"),
-    ],
-    unannotated_with_default=5,
-    my_model_a_with_default: SomeInputModelA = SomeInputModelA(),  # noqa: B008
-    an_int_with_default: int = 1,
-    must_be_none_with_default: None = None,
-    an_int_with_equals_field: int = Field(1, ge=0),
-    int_annotated_with_default: Annotated[int, Field(description="hey")] = 5,
-) -> str:
-    _ = (
-        an_int,
-        must_be_none,
-        must_be_none_dumb_annotation,
-        list_of_ints,
-        list_str_or_str,
-        an_int_annotated_with_field,
-        an_int_annotated_with_field_and_others,
-        an_int_annotated_with_junk,
-        field_with_default_via_field_annotation_before_nondefault_arg,
-        unannotated,
-        an_int_annotated_with_field_default,
-        unannotated_with_default,
-        my_model_a,
-        my_model_a_forward_ref,
-        my_model_b,
-        my_model_a_with_default,
-        an_int_with_default,
-        must_be_none_with_default,
-        an_int_with_equals_field,
-        int_annotated_with_default,
-    )
-    return "ok!"
-
-
-async def test_complex_function_runtime_arg_validation_non_json():
-    """Test that basic non-JSON arguments are validated correctly"""
-    meta = func_metadata(complex_arguments_fn)
-
-    # Test with minimum required arguments
-    result = await meta.call_fn_with_arg_validation(
-        complex_arguments_fn,
-        fn_is_async=False,
-        arguments_to_validate={
-            "an_int": 1,
-            "must_be_none": None,
-            "must_be_none_dumb_annotation": None,
-            "list_of_ints": [1, 2, 3],
-            "list_str_or_str": "hello",
-            "an_int_annotated_with_field": 42,
-            "an_int_annotated_with_field_and_others": 5,
-            "an_int_annotated_with_junk": 100,
-            "unannotated": "test",
-            "my_model_a": {},
-            "my_model_a_forward_ref": {},
-            "my_model_b": {"how_many_shrimp": 5, "ok": {"x": 1}, "y": None},
-        },
-        arguments_to_pass_directly=None,
-    )
-    assert result == "ok!"
-
-    # Test with invalid types
-    with pytest.raises(ValueError):
-        await meta.call_fn_with_arg_validation(
-            complex_arguments_fn,
-            fn_is_async=False,
-            arguments_to_validate={"an_int": "not an int"},
-            arguments_to_pass_directly=None,
-        )
-
-
-async def test_complex_function_runtime_arg_validation_with_json():
-    """Test that JSON string arguments are parsed and validated correctly"""
-    meta = func_metadata(complex_arguments_fn)
-
-    result = await meta.call_fn_with_arg_validation(
-        complex_arguments_fn,
-        fn_is_async=False,
-        arguments_to_validate={
-            "an_int": 1,
-            "must_be_none": None,
-            "must_be_none_dumb_annotation": None,
-            "list_of_ints": "[1, 2, 3]",  # JSON string
-            "list_str_or_str": '["a", "b", "c"]',  # JSON string
-            "an_int_annotated_with_field": 42,
-            "an_int_annotated_with_field_and_others": "5",  # JSON string
-            "an_int_annotated_with_junk": 100,
-            "unannotated": "test",
-            "my_model_a": "{}",  # JSON string
-            "my_model_a_forward_ref": "{}",  # JSON string
-            "my_model_b": '{"how_many_shrimp": 5, "ok": {"x": 1}, "y": null}',
-        },
-        arguments_to_pass_directly=None,
-    )
-    assert result == "ok!"
-
-
-def test_str_vs_list_str():
-    """Test handling of string vs list[str] type annotations.
-
-    This is tricky as '"hello"' can be parsed as a JSON string or a Python string.
-    We want to make sure it's kept as a python string.
-    """
-
-    def func_with_str_types(str_or_list: str | list[str]):
-        return str_or_list
-
-    meta = func_metadata(func_with_str_types)
-
-    # Test string input for union type
-    result = meta.pre_parse_json({"str_or_list": "hello"})
-    assert result["str_or_list"] == "hello"
-
-    # Test string input that contains valid JSON for union type
-    # We want to see here that the JSON-vali string is NOT parsed as JSON, but rather
-    # kept as a raw string
-    result = meta.pre_parse_json({"str_or_list": '"hello"'})
-    assert result["str_or_list"] == '"hello"'
-
-    # Test list input for union type
-    result = meta.pre_parse_json({"str_or_list": '["hello", "world"]'})
-    assert result["str_or_list"] == ["hello", "world"]
-
-
-def test_keep_str_as_str():
-    """Test that string arguments are kept as strings"""
-
-    def func_with_str_types(string: str):
-        return string
-
-    meta = func_metadata(func_with_str_types)
-    result = meta.pre_parse_json(
-        {"string": "{'nice to meet you': 'hello', 'goodbye': 5}"}
-    )
-    assert result["string"] == "{'nice to meet you': 'hello', 'goodbye': 5}"
-
-
-def test_missing_annotation():
-    """Test that missing annotations don't cause errors"""
-
-    def fn(x, y):
-        return x + y
-
-    meta = func_metadata(fn)
-    result = meta.pre_parse_json({"x": "1", "y": "2"})
-    assert result["x"] == "1"
-    assert result["y"] == "2"
-
-
-def test_keep_str_union_as_str():
-    """Test that string arguments are kept as strings"""
-
-    def func_with_str_types(string: str | dict[int, str] | None):
-        return string
-
-    meta = func_metadata(func_with_str_types)
-    result = meta.pre_parse_json(
-        {"string": "{'nice to meet you': 'hello', 'goodbye': 5}"}
-    )
-    assert result["string"] == "{'nice to meet you': 'hello', 'goodbye': 5}"
-
-
-def test_keep_str_complex_type_as_str():
-    """Test that string arguments are kept as strings because it's invalid for the field"""
-
-    class SomeModel(BaseModel):
-        x: int
-        y: dict[int, str]
-
-    def func_with_str_types(string: str | SomeModel | None):
-        return string
-
-    meta = func_metadata(func_with_str_types)
-    result = meta.pre_parse_json({"string": '{"x": 1, "y": {"invalid": "hello"}}'})
-    assert result["string"] == '{"x": 1, "y": {"invalid": "hello"}}'
-
-
-def test_convert_str_to_complex_type():
-    """Test that string arguments are converted to the complex type because it's valid for the field"""
-
-    class SomeModel(BaseModel):
-        x: int
-        y: dict[int, str]
-
-    def func_with_str_types(string: str | SomeModel | None):
-        return string
-
-    meta = func_metadata(func_with_str_types)
-    result = meta.pre_parse_json({"string": '{"x": 1, "y": {"1": "hello"}}'})
-    assert result["string"] == {"x": 1, "y": {"1": "hello"}}
-
-
-def test_skip_names():
-    """Test that skipped parameters are not included in the model"""
-
-    def func_with_many_params(
-        keep_this: int, skip_this: str, also_keep: float, also_skip: bool
-    ):
-        return keep_this, skip_this, also_keep, also_skip
-
-    # Skip some parameters
-    meta = func_metadata(func_with_many_params, skip_names=["skip_this", "also_skip"])
-
-    # Check model fields
-    assert "keep_this" in meta.arg_model.model_fields
-    assert "also_keep" in meta.arg_model.model_fields
-    assert "skip_this" not in meta.arg_model.model_fields
-    assert "also_skip" not in meta.arg_model.model_fields
-
-    # Validate that we can call with only non-skipped parameters
-    model: BaseModel = meta.arg_model.model_validate({"keep_this": 1, "also_keep": 2.5})  # type: ignore
-    assert model.keep_this == 1  # type: ignore
-    assert model.also_keep == 2.5  # type: ignore
-
-
-async def test_lambda_function():
-    """Test lambda function schema and validation"""
-    fn = lambda x, y=5: x  # noqa: E731
-    meta = func_metadata(lambda x, y=5: x)
-
-    # Test schema
-    assert meta.arg_model.model_json_schema() == {
-        "properties": {
-            "x": {"title": "x", "type": "string"},
-            "y": {"default": 5, "title": "y", "type": "string"},
-        },
-        "required": ["x"],
-        "title": "<lambda>Arguments",
-        "type": "object",
-    }
-
-    async def check_call(args):
-        return await meta.call_fn_with_arg_validation(
-            fn,
-            fn_is_async=False,
-            arguments_to_validate=args,
-            arguments_to_pass_directly=None,
-        )
-
-    # Basic calls
-    assert await check_call({"x": "hello"}) == "hello"
-    assert await check_call({"x": "hello", "y": "world"}) == "hello"
-    assert await check_call({"x": '"hello"'}) == '"hello"'
-
-    # Missing required arg
-    with pytest.raises(ValueError):
-        await check_call({"y": "world"})
-
-
-def test_complex_function_json_schema():
-    """Test JSON schema generation for complex function arguments.
-
-    Note: Different versions of pydantic output slightly different
-    JSON Schema formats for model fields with defaults. The format changed in 2.9.0:
-
-    1. Before 2.9.0:
-       {
-         "allOf": [{"$ref": "#/$defs/Model"}],
-         "default": {}
-       }
-
-    2. Since 2.9.0:
-       {
-         "$ref": "#/$defs/Model",
-         "default": {}
-       }
-
-    Both formats are valid and functionally equivalent. This test accepts either format
-    to ensure compatibility across our supported pydantic versions.
-
-    This change in format does not affect runtime behavior since:
-    1. Both schemas validate the same way
-    2. The actual model classes and validation logic are unchanged
-    3. func_metadata uses model_validate/model_dump, not the schema directly
-    """
-    meta = func_metadata(complex_arguments_fn)
-    actual_schema = meta.arg_model.model_json_schema()
-
-    # Create a copy of the actual schema to normalize
-    normalized_schema = actual_schema.copy()
-
-    # Normalize the my_model_a_with_default field to handle both pydantic formats
-    if "allOf" in actual_schema["properties"]["my_model_a_with_default"]:
-        normalized_schema["properties"]["my_model_a_with_default"] = {
-            "$ref": "#/$defs/SomeInputModelA",
-            "default": {},
-        }
-
-    assert normalized_schema == {
-        "$defs": {
-            "InnerModel": {
-                "properties": {"x": {"title": "X", "type": "integer"}},
-                "required": ["x"],
-                "title": "InnerModel",
-                "type": "object",
-            },
-            "SomeInputModelA": {
-                "properties": {},
-                "title": "SomeInputModelA",
-                "type": "object",
-            },
-            "SomeInputModelB": {
-                "properties": {
-                    "how_many_shrimp": {
-                        "description": "How many shrimp in the tank???",
-                        "title": "How Many Shrimp",
-                        "type": "integer",
-                    },
-                    "ok": {"$ref": "#/$defs/InnerModel"},
-                    "y": {"title": "Y", "type": "null"},
-                },
-                "required": ["how_many_shrimp", "ok", "y"],
-                "title": "SomeInputModelB",
-                "type": "object",
-            },
-        },
-        "properties": {
-            "an_int": {"title": "An Int", "type": "integer"},
-            "must_be_none": {"title": "Must Be None", "type": "null"},
-            "must_be_none_dumb_annotation": {
-                "title": "Must Be None Dumb Annotation",
-                "type": "null",
-            },
-            "list_of_ints": {
-                "items": {"type": "integer"},
-                "title": "List Of Ints",
-                "type": "array",
-            },
-            "list_str_or_str": {
-                "anyOf": [
-                    {"items": {"type": "string"}, "type": "array"},
-                    {"type": "string"},
-                ],
-                "title": "List Str Or Str",
-            },
-            "an_int_annotated_with_field": {
-                "description": "An int with a field",
-                "title": "An Int Annotated With Field",
-                "type": "integer",
-            },
-            "an_int_annotated_with_field_and_others": {
-                "description": "An int with a field",
-                "exclusiveMinimum": 1,
-                "title": "An Int Annotated With Field And Others",
-                "type": "integer",
-            },
-            "an_int_annotated_with_junk": {
-                "title": "An Int Annotated With Junk",
-                "type": "integer",
-            },
-            "field_with_default_via_field_annotation_before_nondefault_arg": {
-                "default": 1,
-                "title": "Field With Default Via Field Annotation Before Nondefault Arg",
-                "type": "integer",
-            },
-            "unannotated": {"title": "unannotated", "type": "string"},
-            "my_model_a": {"$ref": "#/$defs/SomeInputModelA"},
-            "my_model_a_forward_ref": {"$ref": "#/$defs/SomeInputModelA"},
-            "my_model_b": {"$ref": "#/$defs/SomeInputModelB"},
-            "an_int_annotated_with_field_default": {
-                "default": 1,
-                "description": "An int with a field",
-                "title": "An Int Annotated With Field Default",
-                "type": "integer",
-            },
-            "unannotated_with_default": {
-                "default": 5,
-                "title": "unannotated_with_default",
-                "type": "string",
-            },
-            "my_model_a_with_default": {
-                "$ref": "#/$defs/SomeInputModelA",
-                "default": {},
-            },
-            "an_int_with_default": {
-                "default": 1,
-                "title": "An Int With Default",
-                "type": "integer",
-            },
-            "must_be_none_with_default": {
-                "default": None,
-                "title": "Must Be None With Default",
-                "type": "null",
-            },
-            "an_int_with_equals_field": {
-                "default": 1,
-                "minimum": 0,
-                "title": "An Int With Equals Field",
-                "type": "integer",
-            },
-            "int_annotated_with_default": {
-                "default": 5,
-                "description": "hey",
-                "title": "Int Annotated With Default",
-                "type": "integer",
-            },
-        },
-        "required": [
-            "an_int",
-            "must_be_none",
-            "must_be_none_dumb_annotation",
-            "list_of_ints",
-            "list_str_or_str",
-            "an_int_annotated_with_field",
-            "an_int_annotated_with_field_and_others",
-            "an_int_annotated_with_junk",
-            "unannotated",
-            "my_model_a",
-            "my_model_a_forward_ref",
-            "my_model_b",
-        ],
-        "title": "complex_arguments_fnArguments",
-        "type": "object",
-    }
-
-
-def test_str_vs_int():
-    """
-    Test that string values are kept as strings even when they contain numbers,
-    while numbers are parsed correctly.
-    """
-
-    def func_with_str_and_int(a: str, b: int):
-        return a
-
-    meta = func_metadata(func_with_str_and_int)
-    result = meta.pre_parse_json({"a": "123", "b": 123})
-    assert result["a"] == "123"
-    assert result["b"] == 123

tests/utilities/test_json_schema.py

@@ -0,0 +1,110 @@
+from fastmcp.utilities.json_schema import _prune_param, prune_params
+
+
+def test_prune_param_nonexistent():
+    """Test pruning a parameter that doesn't exist."""
+    schema = {"properties": {"foo": {"type": "string"}}}
+    result = _prune_param(schema, "bar")
+    assert result == schema  # Schema should be unchanged
+
+
+def test_prune_param_exists():
+    """Test pruning a parameter that exists."""
+    schema = {"properties": {"foo": {"type": "string"}, "bar": {"type": "integer"}}}
+    result = _prune_param(schema, "bar")
+    assert result["properties"] == {"foo": {"type": "string"}}
+
+
+def test_prune_param_last_property():
+    """Test pruning the only/last parameter, should leave empty properties object."""
+    schema = {"properties": {"foo": {"type": "string"}}}
+    result = _prune_param(schema, "foo")
+    assert "properties" in result
+    assert result["properties"] == {}
+
+
+def test_prune_param_from_required():
+    """Test pruning a parameter that's in the required list."""
+    schema = {
+        "properties": {"foo": {"type": "string"}, "bar": {"type": "integer"}},
+        "required": ["foo", "bar"],
+    }
+    result = _prune_param(schema, "bar")
+    assert result["required"] == ["foo"]
+
+
+def test_prune_param_last_required():
+    """Test pruning the last required parameter, should remove required field."""
+    schema = {
+        "properties": {"foo": {"type": "string"}, "bar": {"type": "integer"}},
+        "required": ["foo"],
+    }
+    result = _prune_param(schema, "foo")
+    assert "required" not in result
+
+
+def test_prune_param_with_refs():
+    """Test pruning a parameter that has references in $defs."""
+    schema = {
+        "properties": {
+            "foo": {"$ref": "#/$defs/foo_def"},
+            "bar": {"$ref": "#/$defs/bar_def"},
+        },
+        "$defs": {
+            "foo_def": {"type": "string"},
+            "bar_def": {"type": "integer"},
+        },
+    }
+    result = _prune_param(schema, "bar")
+    assert "bar_def" not in result["$defs"]
+    assert "foo_def" in result["$defs"]
+
+
+def test_prune_param_all_refs():
+    """Test pruning all parameters with refs, should remove $defs."""
+    schema = {
+        "properties": {
+            "foo": {"$ref": "#/$defs/foo_def"},
+        },
+        "$defs": {
+            "foo_def": {"type": "string"},
+        },
+    }
+    result = _prune_param(schema, "foo")
+    assert "$defs" not in result
+
+
+def test_prune_params_multiple():
+    """Test pruning multiple parameters at once."""
+    schema = {
+        "properties": {
+            "foo": {"type": "string"},
+            "bar": {"type": "integer"},
+            "baz": {"type": "boolean"},
+        },
+        "required": ["foo", "bar"],
+    }
+    result = prune_params(schema, ["foo", "baz"])
+    assert result["properties"] == {"bar": {"type": "integer"}}
+    assert result["required"] == ["bar"]
+
+
+def test_prune_params_nested_refs():
+    """Test pruning with nested references."""
+    schema = {
+        "properties": {
+            "foo": {
+                "type": "object",
+                "properties": {"nested": {"$ref": "#/$defs/nested_def"}},
+            },
+            "bar": {"$ref": "#/$defs/bar_def"},
+        },
+        "$defs": {
+            "nested_def": {"type": "string"},
+            "bar_def": {"type": "integer"},
+        },
+    }
+    # Removing foo should keep nested_def as it's not referenced anymore
+    result = _prune_param(schema, "foo")
+    assert "nested_def" not in result["$defs"]
+    assert "bar_def" in result["$defs"]

tests/utilities/test_typeadapter.py

@@ -0,0 +1,244 @@
+"""
+This test file adapts tests from test_func_metadata.py which tested a custom implementation
+that has been replaced by pydantic TypeAdapters.
+
+The tests ensure our TypeAdapter-based approach covers all the edge cases the old custom
+implementation handled. Since we're now using standard pydantic functionality, these tests
+may be redundant with pydantic's own tests and could potentially be removed in the future.
+"""
+
+from typing import Annotated
+
+import annotated_types
+import pytest
+from pydantic import BaseModel, Field
+
+from fastmcp.utilities.json_schema import prune_params
+from fastmcp.utilities.types import get_cached_typeadapter
+
+
+# Models must be defined at the module level for forward references to work
+class SomeInputModelA(BaseModel):
+    pass
+
+
+class SomeInputModelB(BaseModel):
+    class InnerModel(BaseModel):
+        x: int
+
+    how_many_shrimp: Annotated[int, Field(description="How many shrimp in the tank???")]
+    ok: InnerModel
+    y: None
+
+
+# Define additional models needed in tests
+class SomeComplexModel(BaseModel):
+    x: int
+    y: dict[int, str]
+
+
+def complex_arguments_fn(
+    an_int: int,
+    must_be_none: None,
+    must_be_none_dumb_annotation: Annotated[None, "blah"],
+    list_of_ints: list[int],
+    # list[str] | str is an interesting case because if it comes in as JSON like
+    # "[\"a\", \"b\"]" then it will be naively parsed as a string.
+    list_str_or_str: list[str] | str,
+    an_int_annotated_with_field: Annotated[
+        int, Field(description="An int with a field")
+    ],
+    an_int_annotated_with_field_and_others: Annotated[
+        int,
+        str,  # Should be ignored, really
+        Field(description="An int with a field"),
+        annotated_types.Gt(1),
+    ],
+    an_int_annotated_with_junk: Annotated[
+        int,
+        "123",
+        456,
+    ],
+    field_with_default_via_field_annotation_before_nondefault_arg: Annotated[
+        int, Field(1)
+    ],
+    unannotated,
+    my_model_a: SomeInputModelA,
+    my_model_a_forward_ref: "SomeInputModelA",
+    my_model_b: SomeInputModelB,
+    an_int_annotated_with_field_default: Annotated[
+        int,
+        Field(1, description="An int with a field"),
+    ],
+    unannotated_with_default=5,
+    my_model_a_with_default: SomeInputModelA = SomeInputModelA(),  # noqa: B008
+    an_int_with_default: int = 1,
+    must_be_none_with_default: None = None,
+    an_int_with_equals_field: int = Field(1, ge=0),
+    int_annotated_with_default: Annotated[int, Field(description="hey")] = 5,
+) -> str:
+    _ = (
+        an_int,
+        must_be_none,
+        must_be_none_dumb_annotation,
+        list_of_ints,
+        list_str_or_str,
+        an_int_annotated_with_field,
+        an_int_annotated_with_field_and_others,
+        an_int_annotated_with_junk,
+        field_with_default_via_field_annotation_before_nondefault_arg,
+        unannotated,
+        an_int_annotated_with_field_default,
+        unannotated_with_default,
+        my_model_a,
+        my_model_a_forward_ref,
+        my_model_b,
+        my_model_a_with_default,
+        an_int_with_default,
+        must_be_none_with_default,
+        an_int_with_equals_field,
+        int_annotated_with_default,
+    )
+    return "ok!"
+
+
+def get_simple_func_adapter():
+    """Get a TypeAdapter for a simple function to avoid forward reference issues"""
+
+    def simple_func(x: int, y: str = "default") -> str:
+        return f"{x}-{y}"
+
+    return get_cached_typeadapter(simple_func)
+
+
+async def test_complex_function_runtime_arg_validation_non_json():
+    """Test that basic non-JSON arguments are validated correctly using a simpler function"""
+    type_adapter = get_simple_func_adapter()
+
+    # Test with minimum required arguments
+    args = {"x": 1}
+    result = type_adapter.validate_python(args)
+    assert (
+        result == "1-default"
+    )  # Don't call result() as TypeAdapter returns the value directly
+
+    # Test with all arguments
+    args = {"x": 1, "y": "hello"}
+    result = type_adapter.validate_python(args)
+    assert result == "1-hello"
+
+    # Test with invalid types
+    with pytest.raises(Exception):
+        type_adapter.validate_python({"x": "not an int"})
+
+
+def test_missing_annotation():
+    """Test that missing annotations don't cause errors"""
+
+    def func_no_annotations(x, y):
+        return x + y
+
+    type_adapter = get_cached_typeadapter(func_no_annotations)
+    result = type_adapter.validate_python({"x": "1", "y": "2"})
+    assert result == "12"  # String concatenation since no type info
+
+
+def test_convert_str_to_complex_type():
+    """Test that string arguments are converted to the complex type when valid"""
+
+    def func_with_str_types(string: SomeComplexModel):
+        return string
+
+    # Create a valid model instance
+    input_data = {"x": 1, "y": {1: "hello"}}
+
+    # Validate with model directly
+    SomeComplexModel.model_validate(input_data)
+
+    # Now check if type adapter validates correctly
+    type_adapter = get_cached_typeadapter(func_with_str_types)
+    result = type_adapter.validate_python({"string": input_data})
+
+    assert isinstance(result, SomeComplexModel)
+    assert result.x == 1
+    assert result.y == {1: "hello"}
+
+
+def test_skip_names():
+    """Test that skipped parameters are not included in the schema"""
+
+    def func_with_many_params(
+        keep_this: int, skip_this: str, also_keep: float, also_skip: bool
+    ):
+        return keep_this, skip_this, also_keep, also_skip
+
+    # Get schema and prune parameters
+    type_adapter = get_cached_typeadapter(func_with_many_params)
+    schema = type_adapter.json_schema()
+    pruned_schema = prune_params(schema, params=["skip_this", "also_skip"])
+
+    # Check that only the desired parameters remain
+    assert "keep_this" in pruned_schema["properties"]
+    assert "also_keep" in pruned_schema["properties"]
+    assert "skip_this" not in pruned_schema["properties"]
+    assert "also_skip" not in pruned_schema["properties"]
+
+    # The pruned parameters should also be removed from required
+    if "required" in pruned_schema:
+        assert "skip_this" not in pruned_schema["required"]
+        assert "also_skip" not in pruned_schema["required"]
+
+
+async def test_lambda_function():
+    """Test lambda function schema and validation"""
+    fn = lambda x, y=5: str(x)  # noqa: E731
+    type_adapter = get_cached_typeadapter(fn)
+
+    # Basic calls - validate_python returns the result directly
+    result = type_adapter.validate_python({"x": "hello"})
+    assert result == "hello"
+
+    result = type_adapter.validate_python({"x": "hello", "y": "world"})
+    assert result == "hello"
+
+    # Missing required arg
+    with pytest.raises(Exception):
+        type_adapter.validate_python({"y": "world"})
+
+
+def test_basic_json_schema():
+    """Test JSON schema generation for a simple function"""
+
+    def simple_func(a: int, b: str = "default") -> str:
+        return f"{a}-{b}"
+
+    type_adapter = get_cached_typeadapter(simple_func)
+    schema = type_adapter.json_schema()
+
+    # Check basic properties
+    assert "properties" in schema
+    assert "a" in schema["properties"]
+    assert "b" in schema["properties"]
+    assert schema["properties"]["a"]["type"] == "integer"
+    assert schema["properties"]["b"]["type"] == "string"
+    assert "default" in schema["properties"]["b"]
+    assert schema["properties"]["b"]["default"] == "default"
+
+    # Check required
+    assert "required" in schema
+    assert "a" in schema["required"]
+    assert "b" not in schema["required"]
+
+
+def test_str_vs_int():
+    """
+    Test that string values are kept as strings even when they contain numbers,
+    while numbers are parsed correctly.
+    """
+
+    def func_with_str_and_int(a: str, b: int):
+        return a
+
+    type_adapter = get_cached_typeadapter(func_with_str_and_int)
+    result = type_adapter.validate_python({"a": "123", "b": 123})
+    assert result == "123"