Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
64 changes: 60 additions & 4 deletions src/anthropic/lib/tools/_beta_functions.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
import logging
from abc import ABC, abstractmethod
from typing import Any, Union, Generic, TypeVar, Callable, Iterable, Coroutine, cast, overload
from inspect import isawaitable, isasyncgenfunction, iscoroutinefunction, isgeneratorfunction
from inspect import Parameter, signature, isawaitable, isasyncgenfunction, iscoroutinefunction, isgeneratorfunction
from collections.abc import Awaitable
from typing_extensions import Literal, TypeAlias, override

Expand Down Expand Up @@ -61,6 +61,28 @@ def __init__(self, content: BetaFunctionToolResultType) -> None:
self.content = content


def _positional_only_parameter_names(func: Callable[..., Any]) -> list[str]:
"""Return the positional-only parameter names of ``func``, in declaration order.

Tool inputs arrive as JSON objects and are handed to the function by name, so
``*args`` cannot be expressed at all and is rejected up front; positional-only
parameters are routed back into positional slots by :meth:`BaseFunctionTool._split_input`.
"""
try:
parameters = signature(func).parameters.values()
except (TypeError, ValueError):
return []

for parameter in parameters:
if parameter.kind is Parameter.VAR_POSITIONAL:
raise TypeError(
f"Tool function {getattr(func, '__name__', func)!r} declares *{parameter.name}; "
"tool inputs are JSON objects passed by name, so variadic positional parameters are not supported"
)

return [parameter.name for parameter in parameters if parameter.kind is Parameter.POSITIONAL_ONLY]


Function = Callable[..., BetaFunctionToolResultType]
FunctionT = TypeVar("FunctionT", bound=Function)

Expand Down Expand Up @@ -163,6 +185,7 @@ def __init__(
raise RuntimeError("Tool functions are only supported with Pydantic v2")

self.func = func
self._positional_only_params = _positional_only_parameter_names(func)
self._func_with_validate = pydantic.validate_call(func)
self.name = name or func.__name__
self._defer_loading = defer_loading
Expand All @@ -186,6 +209,24 @@ def __init__(
def __call__(self) -> CallableT:
return self.func

def _split_input(self, input: dict[object, object]) -> tuple[list[object], dict[object, object]]:
"""Split a tool input object into the positional and keyword arguments for ``func``.

Positional-only parameters cannot be passed by keyword, so their values are
pulled out of the input by name and forwarded positionally. Anything missing
or unexpected is left for ``validate_call`` to report.
"""
if not self._positional_only_params:
return [], input

kwargs = dict(input)
positional: list[object] = []
for param_name in self._positional_only_params:
if param_name not in kwargs:
break
positional.append(kwargs.pop(param_name))
return positional, kwargs

def to_dict(self) -> BetaToolParam:
defn: BetaToolParam = {
"name": self.name,
Expand Down Expand Up @@ -224,7 +265,7 @@ def _create_schema_from_function(self) -> InputSchema:

from pydantic_core import CoreSchema
from pydantic.json_schema import JsonSchemaValue, GenerateJsonSchema
from pydantic_core.core_schema import ArgumentsParameter
from pydantic_core.core_schema import ArgumentsSchema, ArgumentsParameter

class CustomGenerateJsonSchema(GenerateJsonSchema):
def __init__(self, *, func: Callable[..., Any], parsed_docstring: Any) -> None:
Expand All @@ -235,6 +276,19 @@ def __init__(self, *, func: Callable[..., Any], parsed_docstring: Any) -> None:
def __call__(self, *_args: Any, **_kwds: Any) -> "CustomGenerateJsonSchema": # noqa: ARG002
return self

@override
def arguments_schema(self, schema: ArgumentsSchema) -> JsonSchemaValue:
# Tool inputs are JSON objects passed by name, so positional-only parameters
# must become named properties. Left alone, pydantic renders them (and only
# them) in its positional form: a `type: array` schema the API cannot accept.
arguments = [
cast(ArgumentsParameter, {**argument, "mode": "positional_or_keyword"})
if argument.get("mode") == "positional_only"
else argument
for argument in schema["arguments_schema"]
]
return super().arguments_schema(cast(ArgumentsSchema, {**schema, "arguments_schema": arguments}))

@override
def kw_arguments_schema(
self,
Expand Down Expand Up @@ -276,8 +330,9 @@ def call(self, input: object) -> BetaFunctionToolResultType:
if not is_dict(input):
raise TypeError(f"Input must be a dictionary, got {type(input).__name__}")

args, kwargs = self._split_input(input)
try:
return self._func_with_validate(**cast(Any, input))
return self._func_with_validate(*args, **cast(Any, kwargs))
except pydantic.ValidationError as e:
raise ValueError(f"Invalid arguments for function {self.name}") from e

Expand All @@ -290,8 +345,9 @@ async def call(self, input: object) -> BetaFunctionToolResultType:
if not is_dict(input):
raise TypeError(f"Input must be a dictionary, got {type(input).__name__}")

args, kwargs = self._split_input(input)
try:
return await self._func_with_validate(**cast(Any, input))
return await self._func_with_validate(*args, **cast(Any, kwargs))
except pydantic.ValidationError as e:
raise ValueError(f"Invalid arguments for function {self.name}") from e

Expand Down
71 changes: 71 additions & 0 deletions tests/lib/tools/test_functions.py
Original file line number Diff line number Diff line change
Expand Up @@ -440,6 +440,77 @@ def simple_add(a: int, b: int) -> str:

assert function_tool.input_schema == expected_schema

def test_positional_only_parameters(self) -> None:
"""Positional-only parameters are exposed as named properties and routed back positionally."""

def lookup(user_id: int, /, field: str = "name") -> str:
"""Look up a field on a user record."""
return f"{user_id}:{field}"

function_tool = beta_tool(lookup)

# pydantic's own JSON schema for this signature is the positional `type: array`
# form, which is not a valid tool `input_schema`.
assert function_tool.input_schema == {
"additionalProperties": False,
"type": "object",
"properties": {
"user_id": {"title": "User Id", "type": "integer"},
"field": {"title": "Field", "type": "string", "default": "name"},
},
"required": ["user_id"],
}
assert function_tool.call({"user_id": 7, "field": "email"}) == "7:email"
assert function_tool.call({"user_id": 7}) == "7:name"

with pytest.raises(ValueError, match="Invalid arguments for function lookup"):
function_tool.call({"field": "email"})
with pytest.raises(ValueError, match="Invalid arguments for function lookup"):
function_tool.call({"user_id": "seven"})

def test_positional_only_with_keyword_only_parameters(self) -> None:
def combine(a: int, /, *, b: str) -> str:
"""Combine two values."""
return f"{a}{b}"

function_tool = beta_tool(combine)

assert function_tool.input_schema == {
"additionalProperties": False,
"type": "object",
"properties": {"a": {"title": "A", "type": "integer"}, "b": {"title": "B", "type": "string"}},
"required": ["a", "b"],
}
assert function_tool.call({"a": 1, "b": "z"}) == "1z"

async def test_async_positional_only_parameters(self) -> None:
from anthropic.lib.tools._beta_functions import beta_async_tool

async def lookup(user_id: int, /, field: str = "name") -> str:
"""Look up a field on a user record."""
return f"{user_id}:{field}"

function_tool = beta_async_tool(lookup)

assert function_tool.input_schema == {
"additionalProperties": False,
"type": "object",
"properties": {
"user_id": {"title": "User Id", "type": "integer"},
"field": {"title": "Field", "type": "string", "default": "name"},
},
"required": ["user_id"],
}
assert await function_tool.call({"user_id": 3, "field": "email"}) == "3:email"

def test_var_positional_parameter_raises(self) -> None:
def collect(*values: int) -> str:
"""Sum values."""
return str(sum(values))

with pytest.raises(TypeError, match=r"declares \*values"):
beta_tool(collect)


def _get_parameters_info(fn: BaseFunctionTool[Any]) -> dict[str, str]:
param_info: dict[str, str] = {}
Expand Down