ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-21 13:04:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

Add documentation

Browse source
This commit is contained in:
Jeremiah Lowin 2025-06-09 17:18:04 -04:00
commit f8854fb3fd
4 changed files with 887 additions and 247 deletions
Download patch file Download diff file Expand all files Collapse all files

1
docs/docs.json
View file

@ -125,6 +125,7 @@
{
"group": "Patterns",
"pages": [
"patterns/tool-transformation",
"patterns/decorating-methods",
"patterns/http-requests",
"patterns/testing",

426
docs/patterns/tool-transformation.mdx Normal file
View file

@ -0,0 +1,426 @@
---
title: Tool Transformation
sidebarTitle: Tool Transformation
description: Create enhanced tool variants with modified schemas, argument mappings, and custom behavior.
icon: wand-magic-sparkles
---
import { VersionBadge } from '/snippets/version-badge.mdx'
<VersionBadge version="2.8.0" />
Tool transformation allows you to create new, enhanced tools from existing ones. This powerful feature enables you to adapt tools for different contexts, simplify complex interfaces, or add custom logic without duplicating code.
## Why Transform Tools?
Often, an existing tool is *almost* perfect for your use case, but it might have:
- A confusing description (or no description at all).
- Argument names or descriptions that are not intuitive for an LLM (e.g., `q` instead of `query`).
- Unnecessary parameters that you want to hide from the LLM.
- A need for input validation before the original tool is called.
- A need to modify or format the tool's output.
Instead of rewriting the tool from scratch, you can **transform** it to fit your needs.
## Basic Transformation
The primary way to create a transformed tool is with the `Tool.from_tool()` class method. At its simplest, you can use it to change a tool's top-level metadata like its `name`, `description`, or `tags`.
In the following simple example, we take a generic `search` tool and adjust its name and description to help an LLM client better understand its purpose.
```python {13-21}
from fastmcp import FastMCP
from fastmcp.tools import Tool
mcp = FastMCP()
# The original, generic tool
@mcp.tool
def search(query: str, category: str = "all") -> list[dict]:
"""Searches for items in the database."""
return database.search(query, category)
# Create a more domain-specific version by changing its metadata
product_search_tool = Tool.from_tool(
search,
name="find_products",
description="""
Search for products in the e-commerce catalog.
Use this when customers ask about finding specific items,
checking availability, or browsing product categories.
""",
)
mcp.add_tool(product_search_tool)
```
Now, clients see a tool named `find_products` with a clear, domain-specific purpose and relevant tags, even though it still uses the original generic `search` function's logic.
### Parameters
The `Tool.from_tool()` class method is the primary way to create a transformed tool. It takes the following parameters:
- `tool`: The tool to transform. This is the only required argument.
- `name`: An optional name for the new tool.
- `description`: An optional description for the new tool.
- `transform_args`: A dictionary of `ArgTransform` objects, one for each argument you want to modify.
- `transform_fn`: An optional function that will be called instead of the parent tool's logic.
- `tags`: An optional set of tags for the new tool.
- `annotations`: An optional set of `ToolAnnotations` for the new tool.
- `serializer`: An optional function that will be called to serialize the result of the new tool.
The result is a new `TransformedTool` object that wraps the parent tool and applies the transformations you specify. You can add this tool to your MCP server using its `add_tool()` method.
## Modifying Arguments
To modify a tool's parameters, provide a dictionary of `ArgTransform` objects to the `transform_args` parameter of `Tool.from_tool()`. Each key is the name of the *original* argument you want to modify.
<Tip>
You only need to provide a `transform_args` entry for arguments you want to modify. All other arguments will be passed through unchanged.
</Tip>
### The ArgTransform Class
To modify an argument, you need to create an `ArgTransform` object. This object has the following parameters:
- `name`: The new name for the argument.
- `description`: The new description for the argument.
- `default`: The new default value for the argument.
- `default_factory`: A function that will be called to generate a default value for the argument. This is useful for arguments that need to be generated for each tool call, such as timestamps or unique IDs.
- `hide`: Whether to hide the argument from the LLM.
- `required`: Whether the argument is required, usually used to make an optional argument be required instead.
- `type`: The new type for the argument.
<Tip>
Certain combinations of parameters are not allowed. For example, you can only use `default_factory` with `hide=True`, because dynamic defaults cannot be represented in a JSON schema for the client. You can only set required=True
</Tip>
### Descriptions
By far the most common reason to transform a tool, after its own description, is to improve its argument descriptions. A good description is crucial for helping an LLM understand how to use a parameter correctly. This is especially important when wrapping tools from external APIs, whose argument descriptions may be missing or written for developers, not LLMs.
In this example, we add a helpful description to the `user_id` argument:
```python {16-19}
from fastmcp import FastMCP
from fastmcp.tools import Tool
from fastmcp.tools.tool_transform import ArgTransform
mcp = FastMCP()
@mcp.tool
def find_user(user_id: str):
"""Finds a user by their ID."""
...
new_tool = Tool.from_tool(
find_user,
transform_args={
"user_id": ArgTransform(
description=(
"The unique identifier for the user, "
"usually in the format 'usr-xxxxxxxx'."
)
)
}
)
```
### Names
At times, you may want to rename an argument to make it more intuitive for an LLM.
For example, in the following example, we take a generic `q` argument and expand it to `search_query`:
```python {15}
from fastmcp import FastMCP
from fastmcp.tools import Tool
from fastmcp.tools.tool_transform import ArgTransform
mcp = FastMCP()
@mcp.tool
def search(q: str):
"""Searches for items in the database."""
return database.search(q)
new_tool = Tool.from_tool(
search,
transform_args={
"q": ArgTransform(name="search_query")
}
)
```
### Default Values
You can update the default value for any argument using the `default` parameter. Here, we change the default value of the `y` argument to 10:
```python{15}
from fastmcp import FastMCP
from fastmcp.tools import Tool
from fastmcp.tools.tool_transform import ArgTransform
mcp = FastMCP()
@mcp.tool
def add(x: int, y: int) -> int:
"""Adds two numbers."""
return x + y
new_tool = Tool.from_tool(
add,
transform_args={
"y": ArgTransform(default=10)
}
)
```
Default values are especially useful in combination with hidden arguments.
### Hiding Arguments
Sometimes a tool requires arguments that shouldn't be exposed to the LLM, such as API keys, configuration flags, or internal IDs. You can hide these parameters using `hide=True`. You can only hide arguments that already have a default value, or that you provide a new `default` or `default_factory` for.
```python {19-20}
import os
from fastmcp import FastMCP
from fastmcp.tools import Tool
from fastmcp.tools.tool_transform import ArgTransform
mcp = FastMCP()
@mcp.tool
def send_email(to: str, subject: str, body: str, api_key: str):
"""Sends an email."""
...
# Create a simplified version that hides the API key
new_tool = Tool.from_tool(
send_email,
name="send_notification",
transform_args={
"api_key": ArgTransform(
hide=True,
default=os.environ.get("EMAIL_API_KEY"),
)
}
)
```
The LLM now only sees the `to`, `subject`, and `body` parameters. The `api_key` is supplied automatically from an environment variable.
For values that must be generated for each tool call (like timestamps or unique IDs), use `default_factory`, which is called with no arguments every time the tool is called. For example,
```python {3-4}
transform_args = {
'timestamp': ArgTransform(
hide=True,
default_factory=lambda: datetime.now(),
)
}
```
<Warning>
`default_factory` can only be used with `hide=True`. This is because visible parameters need static defaults that can be represented in a JSON schema for the client.
</Warning>
### Required Values
In rare cases where you want to make an optional argument required, you can set `required=True`. This has no effect if the argument was already required.
```python {3}
transform_args = {
'user_id': ArgTransform(
required=True,
)
}
```
## Modifying Tool Behavior
<Warning>
With great power comes great responsibility. Modifying tool behavior is a very advanced feature.
</Warning>
In addition to changing a tool's schema, advanced users can also modify its behavior. This is useful for adding validation logic, or for post-processing the tool's output.
The `from_tool()` method takes a `transform_fn` parameter, which is an async function that replaces the parent tool's logic and gives you complete control over the tool's execution.
### The Transform Function
The `transform_fn` is an async function that **completely replaces** the parent tool's logic.
Critically, the transform function's arguments are used to determine the new tool's final schema. Any arguments that are not already present in the parent tool schema OR the `transform_args` will be added to the new tool's schema. Note that when `transform_args` and your function have the same argument name, the `transform_args` metadata will take precedence, if provided.
```python
async def my_custom_logic(user_input: str, max_length: int = 100) -> str:
# Your custom logic here - this completely replaces the parent tool
return f"Custom result for: {user_input[:max_length]}"
Tool.from_tool(transform_fn=my_custom_logic)
```
<Tip>
The name / docstring of the `transform_fn` are ignored. Only its arguments are used to determine the final schema.
</Tip>
### Calling the Parent Tool
Most of the time, you don't want to completely replace the parent tool's behavior. Instead, you want to add validation, modify inputs, or post-process outputs while still leveraging the parent tool's core functionality. For this, FastMCP provides the special `forward()` and `forward_raw()` functions.
Both `forward()` and `forward_raw()` are async functions that let you call the parent tool from within your `transform_fn`:
- **`forward()`** (recommended): Automatically handles argument mapping based on your `ArgTransform` configurations. Call it with the transformed argument names.
- **`forward_raw()`**: Bypasses all transformation and calls the parent tool directly with its original argument names. This is rarely needed unless you're doing complex argument manipulation, perhaps without `arg_transforms`.
The most common transformation pattern is to validate (potentially renamed) arguments before calling the parent tool. Here's an example that validates that `x` and `y` are positive before calling the parent tool:
<Tabs>
<Tab title="Using forward()">
In the simplest case, your parent tool and your transform function have the same arguments. You can call `forward()` with the same argument names as the parent tool:
```python {15}
from fastmcp import FastMCP
from fastmcp.tools import Tool
from fastmcp.tools.tool_transform import forward
mcp = FastMCP()
@mcp.tool
def add(x: int, y: int) -> int:
"""Adds two numbers."""
return x + y
async def ensure_positive(x: int, y: int) -> int:
if x <= 0 or y <= 0:
raise ValueError("x and y must be positive")
return await forward(x=x, y=y)
new_tool = Tool.from_tool(
add,
transform_fn=ensure_positive,
)
mcp.add_tool(new_tool)
```
</Tab>
<Tab title="Using forward() with renamed args">
When your transformed tool has different argument names than the parent tool, you can call `forward()` with the renamed arguments and it will automatically map the arguments to the parent tool's arguments:
```python {15, 20-23}
from fastmcp import FastMCP
from fastmcp.tools import Tool
from fastmcp.tools.tool_transform import forward
mcp = FastMCP()
@mcp.tool
def add(x: int, y: int) -> int:
"""Adds two numbers."""
return x + y
async def ensure_positive(a: int, b: int) -> int:
if a <= 0 or b <= 0:
raise ValueError("a and b must be positive")
return await forward(a=a, b=b)
new_tool = Tool.from_tool(
add,
transform_fn=ensure_positive,
transform_args={
"x": ArgTransform(name="a"),
"y": ArgTransform(name="b"),
}
)
mcp.add_tool(new_tool)
```
</Tab>
<Tab title="Using forward_raw()">
Finally, you can use `forward_raw()` to bypass all argument mapping and call the parent tool directly with its original argument names.
```python {15, 20-23}
from fastmcp import FastMCP
from fastmcp.tools import Tool
from fastmcp.tools.tool_transform import forward
mcp = FastMCP()
@mcp.tool
def add(x: int, y: int) -> int:
"""Adds two numbers."""
return x + y
async def ensure_positive(a: int, b: int) -> int:
if a <= 0 or b <= 0:
raise ValueError("a and b must be positive")
return await forward_raw(x=a, y=b)
new_tool = Tool.from_tool(
add,
transform_fn=ensure_positive,
transform_args={
"x": ArgTransform(name="a"),
"y": ArgTransform(name="b"),
}
)
mcp.add_tool(new_tool)
```
</Tab>
</Tabs>
### Passing Arguments with **kwargs
If your `transform_fn` includes `**kwargs` in its signature, it will receive **all arguments from the parent tool after `ArgTransform` configurations have been applied**. This is powerful for creating flexible validation functions that don't require you to add every argument to the function signature.
In the following example, we wrap a parent tool that accepts two arguments `x` and `y`. These are renamed to `a` and `b` in the transformed tool, and the transform only validates `a`, passing the other argument through as `**kwargs`.
```python {12, 15}
from fastmcp import FastMCP
from fastmcp.tools import Tool
from fastmcp.tools.tool_transform import forward
mcp = FastMCP()
@mcp.tool
def add(x: int, y: int) -> int:
"""Adds two numbers."""
return x + y
async def ensure_a_positive(a: int, **kwargs) -> int:
if a <= 0:
raise ValueError("a must be positive")
return await forward(a=a, **kwargs)
new_tool = Tool.from_tool(
add,
transform_fn=ensure_positive,
transform_args={
"x": ArgTransform(name="a"),
"y": ArgTransform(name="b"),
}
)
mcp.add_tool(new_tool)
```
<Tip>
In the above example, `**kwargs` receives the renamed argument `b`, not the original argument `y`. It is therefore recommended to use with `forward()`, not `forward_raw()`.
</Tip>
## Common Patterns
Tool transformation is a flexible feature that supports many powerful patterns. Here are a few common use cases to give you ideas.
### Adapting Remote or Generated Tools
This is one of the most common reasons to use tool transformation. Tools from remote servers (via a [proxy](/servers/proxy)) or generated from an [OpenAPI spec](/servers/openapi) are often too generic for direct use by an LLM. You can use transformation to create a simpler, more intuitive version for your specific needs.
### Chaining Transformations
You can chain transformations by using an already transformed tool as the parent for a new transformation. This lets you build up complex behaviors in layers, for example, first renaming arguments, and then adding validation logic to the renamed tool.
### Context-Aware Tool Factories
You can write functions that act as "factories," generating specialized versions of a tool for different contexts. For example, you could create a `get_my_data` tool that is specific to the currently logged-in user by hiding the `user_id` parameter and providing it automatically.

320
src/fastmcp/tools/tool_transform.py
View file

@ -1,107 +1,3 @@
"""# Tool Transformation
Transform existing tools with modified schemas, argument mappings, and custom behavior.
Use this for creating tool variants, adapting tools for different contexts, or adding
custom logic while preserving the original tool's functionality.
## Quick Reference
### Basic Argument Renaming
```python
# Transform specific parent arguments (others pass through unchanged)
new_tool = Tool.from_tool(
original_tool,
transform_args={"old_param": "new_param"} # Only transforms this one arg
)
```
### Complex Transformations
```python
from fastmcp.tools.tool_transform import ArgTransform
new_tool = Tool.from_tool(
original_tool,
transform_args={
"old_name": ArgTransform(name="new_name", description="Updated desc"),
"hidden_param": ArgTransform(hide=True, default="constant_value"),
"simple": "renamed"
}
)
```
### Custom Transform Functions
```python
async def my_transform(new_x: int, new_y: int) -> str:
# Use forward() with transformed argument names
result = await forward(new_x=new_x, new_y=new_y)
return f"Custom: {result}"
new_tool = Tool.from_tool(
original_tool,
transform_fn=my_transform,
transform_args={"x": "new_x", "y": "new_y"}
)
```
### Using **kwargs for Flexibility
```python
async def flexible_transform(**kwargs) -> str:
# kwargs contains all transformed arguments
result = await forward(**kwargs)
return f"Got: {kwargs}"
new_tool = Tool.from_tool(
original_tool,
transform_fn=flexible_transform,
transform_args={"x": "input_x", "y": "input_y"}
)
```
## Key Functions
- `forward(**kwargs)`: Call parent tool with transformed argument names
- `forward_raw(**kwargs)`: Call parent tool with original argument names
## Important Notes
- `transform_args` is optional - if empty/None, all parent arguments pass through unchanged
- Only arguments listed in `transform_args` are transformed, others remain as-is
- Functions with `**kwargs` receive both transformed and untransformed arguments
## ArgTransform Options
- `name`: Rename the argument
- `description`: Change the description
- `default`: Add/change default value
- `hide=True`: Hide the argument from clients (pass constant value to parent)
## Common Patterns
```python
# Chain transformations (partial transforms at each step)
tool1 = Tool.from_tool(original, transform_args={"a": "x"}) # Only transforms 'a'
tool2 = Tool.from_tool(tool1, transform_args={"x": "final"}) # Only transforms 'x'
# Pure passthrough (no transform_args needed)
enhanced = Tool.from_tool(
original,
name="enhanced_version",
description="Better tool",
tags={"v2", "enhanced"}
# No transform_args = all parent args pass through unchanged
)
# Hide specific arguments with constant values
simplified = Tool.from_tool(
complex_tool,
transform_args={
"api_key": ArgTransform(hide=True, default="secret_key"), # Hidden constant
"debug": ArgTransform(hide=True) # Hidden, uses parent's default
}
)
```
"""
from __future__ import annotations
import inspect
@ -109,19 +5,19 @@ from collections.abc import Callable
from contextvars import ContextVar
from dataclasses import dataclass
from types import EllipsisType
from typing import TYPE_CHECKING, Any
from typing import Any, Literal
from mcp.types import EmbeddedResource, ImageContent, TextContent, ToolAnnotations
from pydantic import ConfigDict
from fastmcp.tools.tool import ParsedFunction, Tool
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.types import get_cached_typeadapter
if TYPE_CHECKING:
pass
logger = get_logger(__name__)
NotSet = ...
# Context variable to store current transformed tool
_current_tool: ContextVar[TransformedTool | None] = ContextVar(
@ -197,8 +93,10 @@ class ArgTransform:
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:
# Rename argument 'old_name' to 'new_name'
@ -210,6 +108,9 @@ class ArgTransform:
# Add a default value (makes argument optional)
ArgTransform(default=42)
# Add a default factory (makes argument optional)
ArgTransform(default_factory=lambda: time.time())
# Change the type
ArgTransform(type=str)
@ -219,15 +120,53 @@ class ArgTransform:
# Hide argument but pass a constant value to parent
ArgTransform(hide=True, default="constant_value")
# Hide argument but pass a factory-generated value to parent
ArgTransform(hide=True, default_factory=lambda: uuid.uuid4().hex)
# Make an optional parameter required (removes any default)
ArgTransform(required=True)
# Combine multiple transformations
ArgTransform(name="new_name", description="New desc", default=None, type=int)
"""
name: str | None | EllipsisType = ...
description: str | None | EllipsisType = ...
default: Any | EllipsisType = ...
type: Any | EllipsisType = ...
name: str | EllipsisType = NotSet
description: str | EllipsisType = NotSet
default: Any | EllipsisType = NotSet
default_factory: Callable[[], Any] | EllipsisType = NotSet
type: Any | EllipsisType = NotSet
hide: bool = False
required: Literal[True] | EllipsisType = 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."
)
class TransformedTool(Tool):
@ -249,9 +188,12 @@ class TransformedTool(Tool):
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]
@ -278,7 +220,29 @@ class TransformedTool(Tool):
for param_name, param_schema in properties.items():
if param_name not in arguments and "default" in param_schema:
arguments[param_name] = param_schema["default"]
# 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:
@ -291,11 +255,11 @@ class TransformedTool(Tool):
def from_tool(
cls,
tool: Tool,
transform_fn: Callable[..., Any] | None = None,
name: str | None = None,
transform_args: dict[str, str | ArgTransform | None] | None = None,
description: str | None = None,
tags: set[str] | None = None,
transform_fn: Callable[..., Any] | None = None,
transform_args: dict[str, ArgTransform] | None = None,
annotations: ToolAnnotations | None = None,
serializer: Callable[[Any], str] | None = None,
) -> TransformedTool:
@ -338,16 +302,16 @@ class TransformedTool(Tool):
Tool.from_tool(parent, transform_fn=flexible, transform_args={"a": "x"})
"""
transform_args = transform_args or {}
# Validate transform_args early
if 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 has: {', '.join(sorted(parent_params))}"
)
# 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 has: {', '.join(sorted(parent_params))}"
)
# Always create the forwarding transform
schema, forwarding_fn = cls._create_forwarding_transform(tool, transform_args)
@ -397,10 +361,8 @@ class TransformedTool(Tool):
if transform_args:
new_names = []
for old_name, transform in transform_args.items():
if isinstance(transform, str):
new_names.append(transform)
elif isinstance(transform, ArgTransform) and not transform.hide:
if transform.name is not ... and transform.name is not None:
if not transform.hide:
if transform.name is not NotSet:
new_names.append(transform.name)
else:
new_names.append(old_name)
@ -421,7 +383,7 @@ class TransformedTool(Tool):
final_description = description if description is not None else tool.description
return cls(
transformed_tool = cls(
fn=final_fn,
forwarding_fn=forwarding_fn,
parent_tool=tool,
@ -431,13 +393,16 @@ class TransformedTool(Tool):
tags=tags or tool.tags,
annotations=annotations or tool.annotations,
serializer=serializer or tool.serializer,
transform_args=transform_args,
)
return transformed_tool
@classmethod
def _create_forwarding_transform(
cls,
parent_tool: Tool,
transform_args: dict[str, str | ArgTransform | None] | None,
transform_args: dict[str, ArgTransform] | None,
) -> tuple[dict[str, Any], Callable[..., Any]]:
"""Create schema and forwarding function that encapsulates all transformation logic.
@ -469,20 +434,25 @@ class TransformedTool(Tool):
if transform_args and old_name in transform_args:
transform = transform_args[old_name]
else:
transform = ... # Default behavior - pass through
# Default behavior - pass through (no transformation)
transform = ArgTransform() # Default ArgTransform with no changes
# Handle hidden parameters with defaults
if isinstance(transform, ArgTransform) and transform.hide:
if transform.hide:
# Validate that hidden parameters without user defaults have parent defaults
if transform.default is ... and old_name in parent_required:
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 provided in ArgTransform. Either provide a default "
f"in ArgTransform or don't hide required parameters."
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 transform.default is not ...:
# Hidden parameter with a constant value
hidden_defaults[old_name] = transform.default
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
@ -532,7 +502,13 @@ class TransformedTool(Tool):
parent_args[old_name] = value
# Add hidden defaults (constant values for hidden parameters)
parent_args.update(hidden_defaults)
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)
@ -542,7 +518,7 @@ class TransformedTool(Tool):
def _apply_single_transform(
old_name: str,
old_schema: dict[str, Any],
transform: str | ArgTransform | None | EllipsisType,
transform: ArgTransform,
is_required: bool,
) -> tuple[str, dict[str, Any], bool] | None:
"""Apply transformation to a single parameter.
@ -553,49 +529,55 @@ class TransformedTool(Tool):
Args:
old_name: Original name of the parameter.
old_schema: Original JSON schema for the parameter.
transform: Transformation to apply (string for rename, ArgTransform for complex,
None to drop, ... to pass through unchanged).
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 is ...:
# Not in transform_args - pass through
return old_name, old_schema.copy(), is_required
elif transform is None:
# Explicitly set to None in transform_args - drop the parameter
if transform.hide:
return None
if isinstance(transform, str):
# Simple rename
return transform, old_schema.copy(), is_required
# 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
if isinstance(transform, ArgTransform):
if transform.hide:
return None
# Ensure new_name is always a string
if not isinstance(new_name, str):
new_name = old_name
if transform.name is not ...:
new_name = transform.name or old_name # Handle None case
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_name = old_name
new_schema = old_schema.copy()
if transform.description is not ...:
new_schema["description"] = transform.description
if transform.default is not ...:
new_schema["default"] = transform.default
is_required = False
if transform.type is not ...:
# 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)
return new_name, new_schema, is_required # type: ignore[return-value]
# 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)
raise ValueError(f"Invalid transform: {transform}")
# 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)
return new_name, new_schema, is_required
@staticmethod
def _merge_schema_with_precedence(

383
tests/tools/test_tool_transform.py
View file

@ -37,35 +37,32 @@ def test_tool_from_tool_no_change(add_tool):
assert new_tool.description == add_tool.description
async def test_tool_change_arg_name_with_string(add_tool):
new_tool = Tool.from_tool(add_tool, transform_args={"old_x": "new_x"})
assert sorted(new_tool.parameters["properties"]) == ["new_x", "old_y"]
assert get_property(new_tool, "new_x") == get_property(add_tool, "old_x")
assert get_property(new_tool, "old_y") == get_property(add_tool, "old_y")
assert new_tool.parameters["required"] == ["new_x"]
result = await new_tool.run(arguments={"new_x": 1, "old_y": 2})
assert result[0].text == "3" # type: ignore
async def test_renamed_arg_description_is_maintained(add_tool):
new_tool = Tool.from_tool(add_tool, transform_args={"old_x": "new_x"})
assert get_property(new_tool, "new_x")["description"] == "old_x description"
new_tool = Tool.from_tool(
add_tool, transform_args={"old_x": ArgTransform(name="new_x")}
)
assert (
new_tool.parameters["properties"]["new_x"]["description"] == "old_x description"
)
async def test_tool_defaults_are_maintained_on_unmapped_args(add_tool):
new_tool = Tool.from_tool(add_tool, transform_args={"old_x": "new_x"})
new_tool = Tool.from_tool(
add_tool, transform_args={"old_x": ArgTransform(name="new_x")}
)
result = await new_tool.run(arguments={"new_x": 1})
assert result[0].text == "11" # type: ignore
async def test_tool_defaults_are_maintained_on_mapped_args(add_tool):
new_tool = Tool.from_tool(add_tool, transform_args={"old_y": "new_y"})
new_tool = Tool.from_tool(
add_tool, transform_args={"old_y": ArgTransform(name="new_y")}
)
result = await new_tool.run(arguments={"old_x": 1})
assert result[0].text == "11" # type: ignore
def test_tool_change_arg_name_with_arg_transform(add_tool):
def test_tool_change_arg_name(add_tool):
new_tool = Tool.from_tool(
add_tool, transform_args={"old_x": ArgTransform(name="new_x")}
)
@ -83,15 +80,7 @@ def test_tool_change_arg_description(add_tool):
assert get_property(new_tool, "old_x")["description"] == "new description"
async def test_tool_drop_arg_with_none(add_tool):
# drop the arg with a default value
new_tool = Tool.from_tool(add_tool, transform_args={"old_y": None})
assert sorted(new_tool.parameters["properties"]) == ["old_x"]
result = await new_tool.run(arguments={"old_x": 1})
assert result[0].text == "11" # type: ignore
async def test_tool_drop_arg_with_arg_transform(add_tool):
async def test_tool_drop_arg(add_tool):
new_tool = Tool.from_tool(
add_tool, transform_args={"old_y": ArgTransform(hide=True)}
)
@ -147,7 +136,7 @@ async def test_mixed_hidden_args_with_custom_function(add_tool):
add_tool,
transform_fn=custom_fn,
transform_args={
"old_x": "visible_x", # Rename and expose
"old_x": ArgTransform(name="visible_x"), # Rename and expose
"old_y": ArgTransform(hide=True, default=25), # Hidden with constant
},
)
@ -206,7 +195,10 @@ async def test_forward_with_argument_mapping(add_tool):
new_tool = Tool.from_tool(
add_tool,
transform_fn=custom_fn,
transform_args={"old_x": "new_x", "old_y": "new_y"},
transform_args={
"old_x": ArgTransform(name="new_x"),
"old_y": ArgTransform(name="new_y"),
},
)
result = await new_tool.run(arguments={"new_x": 2, "new_y": 3})
@ -221,7 +213,10 @@ async def test_forward_with_incorrect_args_raises_error(add_tool):
new_tool = Tool.from_tool(
add_tool,
transform_fn=custom_fn,
transform_args={"old_x": "new_x", "old_y": "new_y"},
transform_args={
"old_x": ArgTransform(name="new_x"),
"old_y": ArgTransform(name="new_y"),
},
)
with pytest.raises(
TypeError, match=re.escape("Got unexpected keyword argument(s): old_x, old_y")
@ -240,7 +235,10 @@ async def test_forward_raw_without_argument_mapping(add_tool):
new_tool = Tool.from_tool(
add_tool,
transform_fn=custom_fn,
transform_args={"old_x": "new_x", "old_y": "new_y"},
transform_args={
"old_x": ArgTransform(name="new_x"),
"old_y": ArgTransform(name="new_y"),
},
)
result = await new_tool.run(arguments={"new_x": 2, "new_y": 3})
@ -284,7 +282,9 @@ async def test_fn_with_kwargs_receives_transformed_arg_names(add_tool):
return result
new_tool = Tool.from_tool(
add_tool, transform_fn=custom_fn, transform_args={"old_x": "new_x"}
add_tool,
transform_fn=custom_fn,
transform_args={"old_x": ArgTransform(name="new_x")},
)
result = await new_tool.run(arguments={"new_x": 2, "old_y": 3})
assert result[0].text == "5" # type: ignore
@ -300,7 +300,9 @@ async def test_fn_with_kwargs_handles_partial_explicit_args(add_tool):
return result
new_tool = Tool.from_tool(
add_tool, transform_fn=custom_fn, transform_args={"old_x": "new_x"}
add_tool,
transform_fn=custom_fn,
transform_args={"old_x": ArgTransform(name="new_x")},
)
result = await new_tool.run(
arguments={"new_x": 3, "old_y": 7, "some_other_param": "test"}
@ -318,7 +320,9 @@ async def test_fn_with_kwargs_mixed_mapped_and_unmapped_args(add_tool):
return result
new_tool = Tool.from_tool(
add_tool, transform_fn=custom_fn, transform_args={"old_x": "new_x"}
add_tool,
transform_fn=custom_fn,
transform_args={"old_x": ArgTransform(name="new_x")},
) # only map 'a'
result = await new_tool.run(arguments={"new_x": 1, "old_y": 5})
assert result[0].text == "6" # type: ignore
@ -337,7 +341,10 @@ async def test_fn_with_kwargs_dropped_args_not_in_kwargs(add_tool):
new_tool = Tool.from_tool(
add_tool,
transform_fn=custom_fn,
transform_args={"old_x": "new_x", "old_y": None},
transform_args={
"old_x": ArgTransform(name="new_x"),
"old_y": ArgTransform(hide=True),
},
) # drop 'old_y'
result = await new_tool.run(arguments={"new_x": 8})
# 8 + 10 (default value of b in parent)
@ -367,21 +374,12 @@ def test_transform_args_validation_unknown_arg(add_tool):
with pytest.raises(
ValueError, match="Unknown arguments in transform_args: unknown_param"
):
Tool.from_tool(add_tool, transform_args={"unknown_param": "new_name"})
Tool.from_tool(
add_tool, transform_args={"unknown_param": ArgTransform(name="new_name")}
)
def test_transform_args_creates_duplicate_names(add_tool):
"""Test that transform_args creating duplicate parameter names raises ValueError."""
with pytest.raises(
ValueError,
match="Multiple arguments would be mapped to the same names: same_name",
):
Tool.from_tool(
add_tool, transform_args={"old_x": "same_name", "old_y": "same_name"}
)
def test_transform_args_creates_duplicate_names_with_arg_transform(add_tool):
"""Test that transform_args creating duplicate parameter names raises ValueError."""
with pytest.raises(
ValueError,
@ -391,16 +389,16 @@ def test_transform_args_creates_duplicate_names_with_arg_transform(add_tool):
add_tool,
transform_args={
"old_x": ArgTransform(name="same_name"),
"old_y": "same_name",
"old_y": ArgTransform(name="same_name"),
},
)
def test_function_without_kwargs_missing_params(add_tool):
"""Test that function without **kwargs must declare all transformed params."""
"""Test that function missing required transformed parameters raises ValueError."""
def invalid_fn(new_x: int, non_existent: str) -> str:
return "test"
return f"{new_x}_{non_existent}"
with pytest.raises(
ValueError,
@ -409,27 +407,33 @@ def test_function_without_kwargs_missing_params(add_tool):
Tool.from_tool(
add_tool,
transform_fn=invalid_fn,
transform_args={"old_x": "new_x", "old_y": "new_y"},
transform_args={
"old_x": ArgTransform(name="new_x"),
"old_y": ArgTransform(name="new_y"),
},
)
def test_function_without_kwargs_can_have_extra_params(add_tool):
"""Test that function without **kwargs can declare extra params beyond transformed ones."""
"""Test that function can have extra parameters not in parent tool."""
def valid_fn(new_x: int, new_y: int, extra_param: str = "default") -> str:
return f"{new_x + new_y}: {extra_param}"
return f"{new_x}_{new_y}_{extra_param}"
# This should work fine - function declares all required params plus an extra one
tool = Tool.from_tool(
# Should work - extra_param is fine as long as it has a default
new_tool = Tool.from_tool(
add_tool,
transform_fn=valid_fn,
transform_args={"old_x": "new_x", "old_y": "new_y"},
transform_args={
"old_x": ArgTransform(name="new_x"),
"old_y": ArgTransform(name="new_y"),
},
)
# The final schema should include all function parameters
assert "new_x" in tool.parameters["properties"]
assert "new_y" in tool.parameters["properties"]
assert "extra_param" in tool.parameters["properties"]
assert "new_x" in new_tool.parameters["properties"]
assert "new_y" in new_tool.parameters["properties"]
assert "extra_param" in new_tool.parameters["properties"]
def test_function_with_kwargs_can_add_params(add_tool):
@ -443,7 +447,10 @@ def test_function_with_kwargs_can_add_params(add_tool):
tool = Tool.from_tool(
add_tool,
transform_fn=valid_fn,
transform_args={"old_x": "new_x", "old_y": "new_y"},
transform_args={
"old_x": ArgTransform(name="new_x"),
"old_y": ArgTransform(name="new_y"),
},
)
# extra_param is added, new_x and new_y are available
@ -452,28 +459,27 @@ def test_function_with_kwargs_can_add_params(add_tool):
assert "new_y" in tool.parameters["properties"]
async def test_chaining_transformations(add_tool):
async def test_tool_transform_chaining(add_tool):
"""Test that transformed tools can be transformed again."""
# First transformation
tool1 = Tool.from_tool(add_tool, transform_args={"old_x": "x"})
# First transformation: a -> x
tool1 = Tool.from_tool(add_tool, transform_args={"old_x": ArgTransform(name="x")})
# Second transformation on the already-transformed tool
tool2 = Tool.from_tool(tool1, transform_args={"x": "final_x"})
# Second transformation: x -> final_x, using tool1
tool2 = Tool.from_tool(tool1, transform_args={"x": ArgTransform(name="final_x")})
# Should work with the final names
result = await tool2.run(arguments={"final_x": 5, "old_y": 3})
assert result[0].text == "8" # type: ignore
result = await tool2.run(arguments={"final_x": 5})
assert result[0].text == "15" # type: ignore
# And forward() in a custom function should work
async def custom(final_x: int, old_y: int) -> str:
# forward() goes to tool1, which has 'final_x' and 'old_y' after transformation
result = await forward(final_x=final_x, old_y=old_y)
return f"Chained: {result}"
# Transform tool1 with custom function that handles all parameters
async def custom(final_x: int, **kwargs) -> str:
result = await forward(final_x=final_x, **kwargs)
return f"custom {result[0].text}" # Extract text from content
tool3 = Tool.from_tool(tool1, transform_fn=custom, transform_args={"x": "final_x"})
result = await tool3.run(arguments={"final_x": 5, "old_y": 3})
assert "Chained:" in result[0].text # type: ignore
tool3 = Tool.from_tool(
tool1, transform_fn=custom, transform_args={"x": ArgTransform(name="final_x")}
)
result = await tool3.run(arguments={"final_x": 3, "old_y": 5})
assert result[0].text == "custom 8" # type: ignore
class MyModel(BaseModel):
@ -712,7 +718,9 @@ class TestProxy:
add_tool = await proxy_server.get_tool("add")
new_add_tool = Tool.from_tool(
add_tool, name="add_transformed", transform_args={"old_x": "new_x"}
add_tool,
name="add_transformed",
transform_args={"old_x": ArgTransform(name="new_x")},
)
proxy_server.add_tool(new_add_tool)
@ -720,3 +728,226 @@ class TestProxy:
# The tool should be registered with its transformed name
result = await client.call_tool("add_transformed", {"new_x": 1, "old_y": 2})
assert result[0].text == "3" # type: ignore
async def test_arg_transform_default_factory():
"""Test ArgTransform with default_factory for hidden parameters."""
@Tool.from_function
def base_tool(x: int, timestamp: float) -> str:
return f"{x}_{timestamp}"
# Create a tool with default_factory for hidden timestamp
new_tool = Tool.from_tool(
base_tool,
transform_args={
"timestamp": ArgTransform(hide=True, default_factory=lambda: 12345.0)
},
)
# Only x should be visible since timestamp is hidden
assert sorted(new_tool.parameters["properties"]) == ["x"]
# Should work without providing timestamp (gets value from factory)
result = await new_tool.run(arguments={"x": 42})
assert result[0].text == "42_12345.0" # type: ignore
async def test_arg_transform_default_factory_called_each_time():
"""Test that default_factory is called for each execution."""
call_count = 0
def counter_factory():
nonlocal call_count
call_count += 1
return call_count
@Tool.from_function
def base_tool(x: int, counter: int = 0) -> str:
return f"{x}_{counter}"
new_tool = Tool.from_tool(
base_tool,
transform_args={
"counter": ArgTransform(hide=True, default_factory=counter_factory)
},
)
# Only x should be visible since counter is hidden
assert sorted(new_tool.parameters["properties"]) == ["x"]
# First call
result1 = await new_tool.run(arguments={"x": 1})
assert result1[0].text == "1_1" # type: ignore
# Second call should get a different value
result2 = await new_tool.run(arguments={"x": 2})
assert result2[0].text == "2_2" # type: ignore
async def test_arg_transform_hidden_with_default_factory():
"""Test hidden parameter with default_factory."""
@Tool.from_function
def base_tool(x: int, request_id: str) -> str:
return f"{x}_{request_id}"
def make_request_id():
return "req_123"
new_tool = Tool.from_tool(
base_tool,
transform_args={
"request_id": ArgTransform(hide=True, default_factory=make_request_id)
},
)
# Only x should be visible
assert sorted(new_tool.parameters["properties"]) == ["x"]
# Should pass hidden request_id with factory value
result = await new_tool.run(arguments={"x": 42})
assert result[0].text == "42_req_123" # type: ignore
async def test_arg_transform_default_and_factory_raises_error():
"""Test that providing both default and default_factory raises an error."""
with pytest.raises(
ValueError, match="Cannot specify both 'default' and 'default_factory'"
):
ArgTransform(default=42, default_factory=lambda: 24)
async def test_arg_transform_default_factory_requires_hide():
"""Test that default_factory requires hide=True."""
with pytest.raises(
ValueError, match="default_factory can only be used with hide=True"
):
ArgTransform(default_factory=lambda: 42) # hide=False by default
async def test_arg_transform_required_true():
"""Test that required=True makes an optional parameter required."""
@Tool.from_function
def base_tool(optional_param: int = 42) -> str:
return f"value: {optional_param}"
# Make the optional parameter required
new_tool = Tool.from_tool(
base_tool, transform_args={"optional_param": ArgTransform(required=True)}
)
# Parameter should now be required (no default in schema)
assert "optional_param" in new_tool.parameters["required"]
assert "default" not in new_tool.parameters["properties"]["optional_param"]
# Should work when parameter is provided
result = await new_tool.run(arguments={"optional_param": 100})
assert result[0].text == "value: 100" # type: ignore
# Should fail when parameter is not provided
with pytest.raises(TypeError, match="Missing required argument"):
await new_tool.run(arguments={})
async def test_arg_transform_required_false():
"""Test that required=False makes a required parameter optional with default."""
@Tool.from_function
def base_tool(required_param: int) -> str:
return f"value: {required_param}"
# Make the required parameter optional with a default
new_tool = Tool.from_tool(
base_tool,
transform_args={"required_param": ArgTransform(required=False, default=99)},
)
# Parameter should now be optional (not in required list, has default)
assert "required_param" not in new_tool.parameters["required"]
assert new_tool.parameters["properties"]["required_param"]["default"] == 99
# Should work when parameter is not provided (uses default)
result = await new_tool.run(arguments={})
assert result[0].text == "value: 99" # type: ignore
# Should work when parameter is provided
result = await new_tool.run(arguments={"required_param": 123})
assert result[0].text == "value: 123" # type: ignore
async def test_arg_transform_required_with_rename():
"""Test that required works correctly with argument renaming."""
@Tool.from_function
def base_tool(optional_param: int = 42) -> str:
return f"value: {optional_param}"
# Rename and make required
new_tool = Tool.from_tool(
base_tool,
transform_args={
"optional_param": ArgTransform(name="new_param", required=True)
},
)
# New parameter name should be required
assert "new_param" in new_tool.parameters["required"]
assert "optional_param" not in new_tool.parameters["properties"]
assert "new_param" in new_tool.parameters["properties"]
assert "default" not in new_tool.parameters["properties"]["new_param"]
# Should work with new name
result = await new_tool.run(arguments={"new_param": 200})
assert result[0].text == "value: 200" # type: ignore
async def test_arg_transform_required_true_with_default_raises_error():
"""Test that required=True with default raises an error."""
with pytest.raises(
ValueError, match="Cannot specify 'required=True' with 'default'"
):
ArgTransform(required=True, default=42)
async def test_arg_transform_required_true_with_factory_raises_error():
"""Test that required=True with default_factory raises an error."""
with pytest.raises(
ValueError, match="default_factory can only be used with hide=True"
):
ArgTransform(required=True, default_factory=lambda: 42)
async def test_arg_transform_required_no_change():
"""Test that required=... (NotSet) leaves requirement status unchanged."""
@Tool.from_function
def base_tool(required_param: int, optional_param: int = 42) -> str:
return f"values: {required_param}, {optional_param}"
# Transform without changing required status
new_tool = Tool.from_tool(
base_tool,
transform_args={
"required_param": ArgTransform(name="req"),
"optional_param": ArgTransform(name="opt"),
},
)
# Required status should be unchanged
assert "req" in new_tool.parameters["required"]
assert "opt" not in new_tool.parameters["required"]
assert new_tool.parameters["properties"]["opt"]["default"] == 42
# Should work as expected
result = await new_tool.run(arguments={"req": 1})
assert result[0].text == "values: 1, 42" # type: ignore
async def test_arg_transform_hide_and_required_raises_error():
"""Test that hide=True and required=True together raises an error."""
with pytest.raises(
ValueError, match="Cannot specify both 'hide=True' and 'required=True'"
):
ArgTransform(hide=True, required=True)