src/fastmcp/tools/tool_transform.py

from __future__ import annotations

import inspect
import warnings
from collections.abc import Callable
from contextvars import ContextVar
from dataclasses import dataclass
from typing import Annotated, Any, Literal, cast

import pydantic_core
from mcp.types import ToolAnnotations
from pydantic import ConfigDict
from pydantic.fields import Field
from pydantic.functional_validators import BeforeValidator

import fastmcp
from fastmcp.tools.tool import ParsedFunction, Tool, ToolResult, _convert_to_content
from fastmcp.utilities.components import _convert_set_default_none
from fastmcp.utilities.json_schema import compress_schema
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.types import (
    FastMCPBaseModel,
    NotSet,
    NotSetT,
    get_cached_typeadapter,
)

logger = get_logger(__name__)


# Context variable to store current transformed tool
_current_tool: ContextVar[TransformedTool | None] = ContextVar(  # type: ignore[assignment]
    "_current_tool", default=None
)


async def forward(**kwargs) -> ToolResult:
    """Forward to parent tool with argument transformation applied.

    This function can only be called from within a transformed tool's custom
    function. It applies argument transformation (renaming, validation) before
    calling the parent tool.

    For example, if the parent tool has args `x` and `y`, but the transformed
    tool has args `a` and `b`, and an `transform_args` was provided that maps `x` to
    `a` and `y` to `b`, then `forward(a=1, b=2)` will call the parent tool with
    `x=1` and `y=2`.

    Args:
        **kwargs: Arguments to forward to the parent tool (using transformed names).

    Returns:
        The ToolResult from the parent tool execution.

    Raises:
        RuntimeError: If called outside a transformed tool context.
        TypeError: If provided arguments don't match the transformed schema.
    """
    tool = _current_tool.get()
    if tool is None:
        raise RuntimeError("forward() can only be called within a transformed tool")

    # Use the forwarding function that handles mapping
    return await tool.forwarding_fn(**kwargs)


async def forward_raw(**kwargs) -> ToolResult:
    """Forward directly to parent tool without transformation.

    This function bypasses all argument transformation and validation, calling the parent
    tool directly with the provided arguments. Use this when you need to call the parent
    with its original parameter names and structure.

    For example, if the parent tool has args `x` and `y`, then `forward_raw(x=1,
    y=2)` will call the parent tool with `x=1` and `y=2`.

    Args:
        **kwargs: Arguments to pass directly to the parent tool (using original names).

    Returns:
        The ToolResult from the parent tool execution.

    Raises:
        RuntimeError: If called outside a transformed tool context.
    """
    tool = _current_tool.get()
    if tool is None:
        raise RuntimeError("forward_raw() can only be called within a transformed tool")

    return await tool.parent_tool.run(kwargs)


@dataclass(kw_only=True)
class ArgTransform:
    """Configuration for transforming a parent tool's argument.

    This class allows fine-grained control over how individual arguments are transformed
    when creating a new tool from an existing one. You can rename arguments, change their
    descriptions, add default values, or hide them from clients while passing constants.

    Attributes:
        name: New name for the argument. Use None to keep original name, or ... for no change.
        description: New description for the argument. Use None to remove description, or ... for no change.
        default: New default value for the argument. Use ... for no change.
        default_factory: Callable that returns a default value. Cannot be used with default.
        type: New type for the argument. Use ... for no change.
        hide: If True, hide this argument from clients but pass a constant value to parent.
        required: If True, make argument required (remove default). Use ... for no change.
        examples: Examples for the argument. Use ... for no change.

    Examples:
        Rename argument 'old_name' to 'new_name'
        ```python
        ArgTransform(name="new_name")
        ```

        Change description only
        ```python
        ArgTransform(description="Updated description")
        ```

        Add a default value (makes argument optional)
        ```python
        ArgTransform(default=42)
        ```

        Add a default factory (makes argument optional)
        ```python
        ArgTransform(default_factory=lambda: time.time())
        ```

        Change the type
        ```python
        ArgTransform(type=str)
        ```

        Hide the argument entirely from clients
        ```python
        ArgTransform(hide=True)
        ```

        Hide argument but pass a constant value to parent
        ```python
        ArgTransform(hide=True, default="constant_value")
        ```

        Hide argument but pass a factory-generated value to parent
        ```python
        ArgTransform(hide=True, default_factory=lambda: uuid.uuid4().hex)
        ```

        Make an optional parameter required (removes any default)
        ```python
        ArgTransform(required=True)
        ```

        Combine multiple transformations
        ```python
        ArgTransform(name="new_name", description="New desc", default=None, type=int)
        ```
    """

    name: str | NotSetT = NotSet
    description: str | NotSetT = NotSet
    default: Any | NotSetT = NotSet
    default_factory: Callable[[], Any] | NotSetT = NotSet
    type: Any | NotSetT = NotSet
    hide: bool = False
    required: Literal[True] | NotSetT = NotSet
    examples: Any | NotSetT = NotSet

    def __post_init__(self):
        """Validate that only one of default or default_factory is provided."""
        has_default = self.default is not NotSet
        has_factory = self.default_factory is not NotSet

        if has_default and has_factory:
            raise ValueError(
                "Cannot specify both 'default' and 'default_factory' in ArgTransform. "
                "Use either 'default' for a static value or 'default_factory' for a callable."
            )

        if has_factory and not self.hide:
            raise ValueError(
                "default_factory can only be used with hide=True. "
                "Visible parameters must use static 'default' values since JSON schema "
                "cannot represent dynamic factories."
            )

        if self.required is True and (has_default or has_factory):
            raise ValueError(
                "Cannot specify 'required=True' with 'default' or 'default_factory'. "
                "Required parameters cannot have defaults."
            )

        if self.hide and self.required is True:
            raise ValueError(
                "Cannot specify both 'hide=True' and 'required=True'. "
                "Hidden parameters cannot be required since clients cannot provide them."
            )

        if self.required is False:
            raise ValueError(
                "Cannot specify 'required=False'. Set a default value instead."
            )


class ArgTransformConfig(FastMCPBaseModel):
    """A model for requesting a single argument transform."""

    name: str | None = Field(default=None, description="The new name for the argument.")
    description: str | None = Field(
        default=None, description="The new description for the argument."
    )
    default: str | int | float | bool | None = Field(
        default=None, description="The new default value for the argument."
    )
    hide: bool = Field(
        default=False, description="Whether to hide the argument from the tool."
    )
    required: Literal[True] | None = Field(
        default=None, description="Whether the argument is required."
    )
    examples: Any | None = Field(default=None, description="Examples of the argument.")

    def to_arg_transform(self) -> ArgTransform:
        """Convert the argument transform to a FastMCP argument transform."""

        return ArgTransform(**self.model_dump(exclude_unset=True))  # pyright: ignore[reportAny]


class TransformedTool(Tool):
    """A tool that is transformed from another tool.

    This class represents a tool that has been created by transforming another tool.
    It supports argument renaming, schema modification, custom function injection,
    structured output control, and provides context for the forward() and forward_raw() functions.

    The transformation can be purely schema-based (argument renaming, dropping, etc.)
    or can include a custom function that uses forward() to call the parent tool
    with transformed arguments. Output schemas and structured outputs are automatically
    inherited from the parent tool but can be overridden or disabled.

    Attributes:
        parent_tool: The original tool that this tool was transformed from.
        fn: The function to execute when this tool is called (either the forwarding
            function for pure transformations or a custom user function).
        forwarding_fn: Internal function that handles argument transformation and
            validation when forward() is called from custom functions.
    """

    model_config = ConfigDict(extra="allow", arbitrary_types_allowed=True)

    parent_tool: Tool
    fn: Callable[..., Any]
    forwarding_fn: Callable[..., Any]  # Always present, handles arg transformation
    transform_args: dict[str, ArgTransform]

    async def run(self, arguments: dict[str, Any]) -> ToolResult:
        """Run the tool with context set for forward() functions.

        This method executes the tool's function while setting up the context
        that allows forward() and forward_raw() to work correctly within custom
        functions.

        Args:
            arguments: Dictionary of arguments to pass to the tool's function.

        Returns:
            ToolResult object containing content and optional structured output.
        """

        # Fill in missing arguments with schema defaults to ensure
        # ArgTransform defaults take precedence over function defaults
        arguments = arguments.copy()
        properties = self.parameters.get("properties", {})

        for param_name, param_schema in properties.items():
            if param_name not in arguments and "default" in param_schema:
                # Check if this parameter has a default_factory from transform_args
                # We need to call the factory for each run, not use the cached schema value
                has_factory_default = False
                if self.transform_args:
                    # Find the original parameter name that maps to this param_name
                    for orig_name, transform in self.transform_args.items():
                        transform_name = (
                            transform.name
                            if transform.name is not NotSet
                            else orig_name
                        )
                        if (
                            transform_name == param_name
                            and transform.default_factory is not NotSet
                        ):
                            # Type check to ensure default_factory is callable
                            if callable(transform.default_factory):
                                arguments[param_name] = transform.default_factory()
                                has_factory_default = True
                                break

                if not has_factory_default:
                    arguments[param_name] = param_schema["default"]

        token = _current_tool.set(self)
        try:
            result = await self.fn(**arguments)

            # If transform function returns ToolResult, respect our output_schema setting
            if isinstance(result, ToolResult):
                if self.output_schema is None:
                    # Check if this is from a custom function that returns ToolResult

                    return_annotation = inspect.signature(self.fn).return_annotation
                    if return_annotation is ToolResult:
                        # Custom function returns ToolResult - preserve its content
                        return result
                    else:
                        # Forwarded call with no explicit schema - preserve parent's structured content
                        # The parent tool may have generated structured content via its own fallback logic
                        return result
                elif self.output_schema.get(
                    "type"
                ) != "object" and not self.output_schema.get("x-fastmcp-wrap-result"):
                    # Non-object explicit schemas disable structured content
                    return ToolResult(
                        content=result.content,
                        structured_content=None,
                    )
                else:
                    return result

            # Otherwise convert to content and create ToolResult with proper structured content

            unstructured_result = _convert_to_content(
                result, serializer=self.serializer
            )

            structured_output = None
            # First handle structured content based on output schema, if any
            if self.output_schema is not None:
                if self.output_schema.get("x-fastmcp-wrap-result"):
                    # Schema says wrap - always wrap in result key
                    structured_output = {"result": result}
                else:
                    structured_output = result
            # If no output schema, try to serialize the result. If it is a dict, use
            # it as structured content. If it is not a dict, ignore it.
            if structured_output is None:
                try:
                    structured_output = pydantic_core.to_jsonable_python(result)
                    if not isinstance(structured_output, dict):
                        structured_output = None
                except Exception:
                    pass

            return ToolResult(
                content=unstructured_result,
                structured_content=structured_output,
            )
        finally:
            _current_tool.reset(token)

    @classmethod
    def from_tool(
        cls,
        tool: Tool,
        name: str | None = None,
        title: str | None | NotSetT = NotSet,
        description: str | None | NotSetT = NotSet,
        tags: set[str] | None = None,
        transform_fn: Callable[..., Any] | None = None,
        transform_args: dict[str, ArgTransform] | None = None,
        annotations: ToolAnnotations | None | NotSetT = NotSet,
        output_schema: dict[str, Any] | None | NotSetT | Literal[False] = NotSet,
        serializer: Callable[[Any], str] | None | NotSetT = NotSet,
        meta: dict[str, Any] | None | NotSetT = NotSet,
        enabled: bool | None = None,
    ) -> TransformedTool:
        """Create a transformed tool from a parent tool.

        Args:
            tool: The parent tool to transform.
            transform_fn: Optional custom function. Can use forward() and forward_raw()
                to call the parent tool. Functions with **kwargs receive transformed
                argument names.
            name: New name for the tool. Defaults to parent tool's name.
            title: New title for the tool. Defaults to parent tool's title.
            transform_args: Optional transformations for parent tool arguments.
                Only specified arguments are transformed, others pass through unchanged:
                - Simple rename (str)
                - Complex transformation (rename/description/default/drop) (ArgTransform)
                - Drop the argument (None)
            description: New description. Defaults to parent's description.
            tags: New tags. Defaults to parent's tags.
            annotations: New annotations. Defaults to parent's annotations.
            output_schema: Control output schema for structured outputs:
                - None (default): Inherit from transform_fn if available, then parent tool
                - dict: Use custom output schema
                - False: Disable output schema and structured outputs
            serializer: New serializer. Defaults to parent's serializer.
            meta: Control meta information:
                - NotSet (default): Inherit from parent tool
                - dict: Use custom meta information
                - None: Remove meta information

        Returns:
            TransformedTool with the specified transformations.

        Examples:
            # Transform specific arguments only
            ```python
            Tool.from_tool(parent, transform_args={"old": "new"})  # Others unchanged
            ```

            # Custom function with partial transforms
            ```python
            async def custom(x: int, y: int) -> str:
                result = await forward(x=x, y=y)
                return f"Custom: {result}"

            Tool.from_tool(parent, transform_fn=custom, transform_args={"a": "x", "b": "y"})
            ```

            # Using **kwargs (gets all args, transformed and untransformed)
            ```python
            async def flexible(**kwargs) -> str:
                result = await forward(**kwargs)
                return f"Got: {kwargs}"

            Tool.from_tool(parent, transform_fn=flexible, transform_args={"a": "x"})
            ```

            # Control structured outputs and schemas
            ```python
            # Custom output schema
            Tool.from_tool(parent, output_schema={
                "type": "object",
                "properties": {"status": {"type": "string"}}
            })

            # Disable structured outputs
            Tool.from_tool(parent, output_schema=False)

            # Return ToolResult for full control
            async def custom_output(**kwargs) -> ToolResult:
                result = await forward(**kwargs)
                return ToolResult(
                    content=[TextContent(text="Summary")],
                    structured_content={"processed": True}
                )
            ```
        """
        transform_args = transform_args or {}

        if transform_fn is not None:
            parsed_fn = ParsedFunction.from_function(transform_fn, validate=False)
        else:
            parsed_fn = None

        # Validate transform_args
        parent_params = set(tool.parameters.get("properties", {}).keys())
        unknown_args = set(transform_args.keys()) - parent_params
        if unknown_args:
            raise ValueError(
                f"Unknown arguments in transform_args: {', '.join(sorted(unknown_args))}. "
                f"Parent tool `{tool.name}` has: {', '.join(sorted(parent_params))}"
            )

        # Always create the forwarding transform
        schema, forwarding_fn = cls._create_forwarding_transform(tool, transform_args)

        # Handle output schema
        if output_schema is NotSet:
            # Use smart fallback: try custom function, then parent
            if transform_fn is not None:
                # parsed fn is not none here
                final_output_schema = cast(ParsedFunction, parsed_fn).output_schema
                if final_output_schema is None:
                    # Check if function returns ToolResult - if so, don't fall back to parent
                    return_annotation = inspect.signature(
                        transform_fn
                    ).return_annotation
                    if return_annotation is ToolResult:
                        final_output_schema = None
                    else:
                        final_output_schema = tool.output_schema
            else:
                final_output_schema = tool.output_schema
        elif output_schema is False:
            # Handle False as deprecated synonym for None (deprecated in 2.11.4)
            if fastmcp.settings.deprecation_warnings:
                warnings.warn(
                    "Passing output_schema=False is deprecated. Use output_schema=None instead.",
                    DeprecationWarning,
                    stacklevel=2,
                )
            final_output_schema = None
        else:
            final_output_schema = cast(dict | None, output_schema)

        if transform_fn is None:
            # User wants pure transformation - use forwarding_fn as the main function
            final_fn = forwarding_fn
            final_schema = schema
        else:
            # parsed fn is not none here
            parsed_fn = cast(ParsedFunction, parsed_fn)
            # User provided custom function - merge schemas
            final_fn = transform_fn

            has_kwargs = cls._function_has_kwargs(transform_fn)

            # Validate function parameters against transformed schema
            fn_params = set(parsed_fn.input_schema.get("properties", {}).keys())
            transformed_params = set(schema.get("properties", {}).keys())

            if not has_kwargs:
                # Without **kwargs, function must declare all transformed params
                # Check if function is missing any parameters required after transformation
                missing_params = transformed_params - fn_params
                if missing_params:
                    raise ValueError(
                        f"Function missing parameters required after transformation: "
                        f"{', '.join(sorted(missing_params))}. "
                        f"Function declares: {', '.join(sorted(fn_params))}"
                    )

                # ArgTransform takes precedence over function signature
                # Start with function schema as base, then override with transformed schema
                final_schema = cls._merge_schema_with_precedence(
                    parsed_fn.input_schema, schema
                )
            else:
                # With **kwargs, function can access all transformed params
                # ArgTransform takes precedence over function signature
                # No validation needed - kwargs makes everything accessible

                # Start with function schema as base, then override with transformed schema
                final_schema = cls._merge_schema_with_precedence(
                    parsed_fn.input_schema, schema
                )

        # Additional validation: check for naming conflicts after transformation
        if transform_args:
            new_names = []
            for old_name, transform in transform_args.items():
                if not transform.hide:
                    if transform.name is not NotSet:
                        new_names.append(transform.name)
                    else:
                        new_names.append(old_name)

            # Check for duplicate names after transformation
            name_counts = {}
            for arg_name in new_names:
                name_counts[arg_name] = name_counts.get(arg_name, 0) + 1

            duplicates = [
                arg_name for arg_name, count in name_counts.items() if count > 1
            ]
            if duplicates:
                raise ValueError(
                    f"Multiple arguments would be mapped to the same names: "
                    f"{', '.join(sorted(duplicates))}"
                )

        final_name = name or tool.name
        final_description = (
            description if not isinstance(description, NotSetT) else tool.description
        )
        final_title = title if not isinstance(title, NotSetT) else tool.title
        final_meta = meta if not isinstance(meta, NotSetT) else tool.meta
        final_annotations = (
            annotations if not isinstance(annotations, NotSetT) else tool.annotations
        )
        final_serializer = (
            serializer if not isinstance(serializer, NotSetT) else tool.serializer
        )
        final_enabled = enabled if enabled is not None else tool.enabled

        transformed_tool = cls(
            fn=final_fn,
            forwarding_fn=forwarding_fn,
            parent_tool=tool,
            name=final_name,
            title=final_title,
            description=final_description,
            parameters=final_schema,
            output_schema=final_output_schema,
            tags=tags or tool.tags,
            annotations=final_annotations,
            serializer=final_serializer,
            meta=final_meta,
            transform_args=transform_args,
            enabled=final_enabled,
        )

        return transformed_tool

    @classmethod
    def _create_forwarding_transform(
        cls,
        parent_tool: Tool,
        transform_args: dict[str, ArgTransform] | None,
    ) -> tuple[dict[str, Any], Callable[..., Any]]:
        """Create schema and forwarding function that encapsulates all transformation logic.

        This method builds a new JSON schema for the transformed tool and creates a
        forwarding function that validates arguments against the new schema and maps
        them back to the parent tool's expected arguments.

        Args:
            parent_tool: The original tool to transform.
            transform_args: Dictionary defining how to transform each argument.

        Returns:
            A tuple containing:
            - The new JSON schema for the transformed tool as a dictionary
            - Async function that validates and forwards calls to the parent tool
        """

        # Build transformed schema and mapping
        parent_defs = parent_tool.parameters.get("$defs", {})
        parent_props = parent_tool.parameters.get("properties", {}).copy()
        parent_required = set(parent_tool.parameters.get("required", []))

        new_props = {}
        new_required = set()
        new_to_old = {}
        hidden_defaults = {}  # Track hidden parameters with constant values

        for old_name, old_schema in parent_props.items():
            # Check if parameter is in transform_args
            if transform_args and old_name in transform_args:
                transform = transform_args[old_name]
            else:
                # Default behavior - pass through (no transformation)
                transform = ArgTransform()  # Default ArgTransform with no changes

            # Handle hidden parameters with defaults
            if transform.hide:
                # Validate that hidden parameters without user defaults have parent defaults
                has_user_default = (
                    transform.default is not NotSet
                    or transform.default_factory is not NotSet
                )
                if not has_user_default and old_name in parent_required:
                    raise ValueError(
                        f"Hidden parameter '{old_name}' has no default value in parent tool "
                        f"and no default or default_factory provided in ArgTransform. Either provide a default "
                        f"or default_factory in ArgTransform or don't hide required parameters."
                    )
                if has_user_default:
                    # Store info for later factory calling or direct value
                    hidden_defaults[old_name] = transform
                # Skip adding to schema (not exposed to clients)
                continue

            transform_result = cls._apply_single_transform(
                old_name,
                old_schema,
                transform,
                old_name in parent_required,
            )

            if transform_result:
                new_name, new_schema, is_required = transform_result
                new_props[new_name] = new_schema
                new_to_old[new_name] = old_name
                if is_required:
                    new_required.add(new_name)

        schema = {
            "type": "object",
            "properties": new_props,
            "required": list(new_required),
        }

        if parent_defs:
            schema["$defs"] = parent_defs
            schema = compress_schema(schema, prune_defs=True)

        # Create forwarding function that closes over everything it needs
        async def _forward(**kwargs):
            # Validate arguments
            valid_args = set(new_props.keys())
            provided_args = set(kwargs.keys())
            unknown_args = provided_args - valid_args

            if unknown_args:
                raise TypeError(
                    f"Got unexpected keyword argument(s): {', '.join(sorted(unknown_args))}"
                )

            # Check required arguments
            missing_args = new_required - provided_args
            if missing_args:
                raise TypeError(
                    f"Missing required argument(s): {', '.join(sorted(missing_args))}"
                )

            # Map arguments to parent names
            parent_args = {}
            for new_name, value in kwargs.items():
                old_name = new_to_old.get(new_name, new_name)
                parent_args[old_name] = value

            # Add hidden defaults (constant values for hidden parameters)
            for old_name, transform in hidden_defaults.items():
                if transform.default is not NotSet:
                    parent_args[old_name] = transform.default
                elif transform.default_factory is not NotSet:
                    # Type check to ensure default_factory is callable
                    if callable(transform.default_factory):
                        parent_args[old_name] = transform.default_factory()

            return await parent_tool.run(parent_args)

        return schema, _forward

    @staticmethod
    def _apply_single_transform(
        old_name: str,
        old_schema: dict[str, Any],
        transform: ArgTransform,
        is_required: bool,
    ) -> tuple[str, dict[str, Any], bool] | None:
        """Apply transformation to a single parameter.

        This method handles the transformation of a single argument according to
        the specified transformation rules.

        Args:
            old_name: Original name of the parameter.
            old_schema: Original JSON schema for the parameter.
            transform: ArgTransform object specifying how to transform the parameter.
            is_required: Whether the original parameter was required.

        Returns:
            Tuple of (new_name, new_schema, new_is_required) if parameter should be kept,
            None if parameter should be dropped.
        """
        if transform.hide:
            return None

        # Handle name transformation - ensure we always have a string
        if transform.name is not NotSet:
            new_name = transform.name if transform.name is not None else old_name
        else:
            new_name = old_name

        # Ensure new_name is always a string
        if not isinstance(new_name, str):
            new_name = old_name

        new_schema = old_schema.copy()

        # Handle description transformation
        if transform.description is not NotSet:
            if transform.description is None:
                new_schema.pop("description", None)  # Remove description
            else:
                new_schema["description"] = transform.description

        # Handle required transformation first
        if transform.required is not NotSet:
            is_required = bool(transform.required)
            if transform.required is True:
                # Remove any existing default when making required
                new_schema.pop("default", None)

        # Handle default value transformation (only if not making required)
        if transform.default is not NotSet and transform.required is not True:
            new_schema["default"] = transform.default
            is_required = False

        # Handle type transformation
        if transform.type is not NotSet:
            # Use TypeAdapter to get proper JSON schema for the type
            type_schema = get_cached_typeadapter(transform.type).json_schema()
            # Update the schema with the type information from TypeAdapter
            new_schema.update(type_schema)

        # Handle examples transformation
        if transform.examples is not NotSet:
            new_schema["examples"] = transform.examples

        return new_name, new_schema, is_required

    @staticmethod
    def _merge_schema_with_precedence(
        base_schema: dict[str, Any], override_schema: dict[str, Any]
    ) -> dict[str, Any]:
        """Merge two schemas, with the override schema taking precedence.

        Args:
            base_schema: Base schema to start with
            override_schema: Schema that takes precedence for overlapping properties

        Returns:
            Merged schema with override taking precedence
        """
        merged_props = base_schema.get("properties", {}).copy()
        merged_required = set(base_schema.get("required", []))

        override_props = override_schema.get("properties", {})
        override_required = set(override_schema.get("required", []))

        # Override properties
        for param_name, param_schema in override_props.items():
            if param_name in merged_props:
                # Merge the schemas, with override taking precedence
                base_param = merged_props[param_name].copy()
                base_param.update(param_schema)
                merged_props[param_name] = base_param
            else:
                merged_props[param_name] = param_schema.copy()

        # Handle required parameters - override takes complete precedence
        # Start with override's required set
        final_required = override_required.copy()

        # For parameters not in override, inherit base requirement status
        # but only if they don't have a default in the final merged properties
        for param_name in merged_required:
            if param_name not in override_props:
                # Parameter not mentioned in override, keep base requirement status
                final_required.add(param_name)
            elif (
                param_name in override_props
                and "default" not in merged_props[param_name]
            ):
                # Parameter in override but no default, keep required if it was required in base
                if param_name not in override_required:
                    # Override doesn't specify it as required, and it has no default,
                    # so inherit from base
                    final_required.add(param_name)

        # Remove any parameters that have defaults (they become optional)
        for param_name, param_schema in merged_props.items():
            if "default" in param_schema:
                final_required.discard(param_name)

        return {
            "type": "object",
            "properties": merged_props,
            "required": list(final_required),
        }

    @staticmethod
    def _function_has_kwargs(fn: Callable[..., Any]) -> bool:
        """Check if function accepts **kwargs.

        This determines whether a custom function can accept arbitrary keyword arguments,
        which affects how schemas are merged during tool transformation.

        Args:
            fn: Function to inspect.

        Returns:
            True if the function has a **kwargs parameter, False otherwise.
        """
        sig = inspect.signature(fn)
        return any(
            p.kind == inspect.Parameter.VAR_KEYWORD for p in sig.parameters.values()
        )


class ToolTransformConfig(FastMCPBaseModel):
    """Provides a way to transform a tool."""

    name: str | None = Field(default=None, description="The new name for the tool.")

    title: str | None = Field(
        default=None,
        description="The new title of the tool.",
    )
    description: str | None = Field(
        default=None,
        description="The new description of the tool.",
    )
    tags: Annotated[set[str], BeforeValidator(_convert_set_default_none)] = Field(
        default_factory=set,
        description="The new tags for the tool.",
    )
    meta: dict[str, Any] | None = Field(
        default=None,
        description="The new meta information for the tool.",
    )

    enabled: bool = Field(
        default=True,
        description="Whether the tool is enabled.",
    )

    arguments: dict[str, ArgTransformConfig] = Field(
        default_factory=dict,
        description="A dictionary of argument transforms to apply to the tool.",
    )

    def apply(self, tool: Tool) -> TransformedTool:
        """Create a TransformedTool from a provided tool and this transformation configuration."""

        tool_changes: dict[str, Any] = self.model_dump(
            exclude_unset=True, exclude={"arguments"}
        )

        return TransformedTool.from_tool(
            tool=tool,
            **tool_changes,
            transform_args={k: v.to_arg_transform() for k, v in self.arguments.items()},
        )


def apply_transformations_to_tools(
    tools: dict[str, Tool],
    transformations: dict[str, ToolTransformConfig],
) -> dict[str, Tool]:
    """Apply a list of transformations to a list of tools. Tools that do not have any transforamtions
    are left unchanged.
    """

    transformed_tools: dict[str, Tool] = {}

    for tool_name, tool in tools.items():
        if transformation := transformations.get(tool_name):
            transformed_tools[transformation.name or tool_name] = transformation.apply(
                tool
            )
            continue

        transformed_tools[tool_name] = tool

    return transformed_tools