Implementation:Anthropics Anthropic sdk python BetaFunctionTool Call

Knowledge Sources	Anthropic Python SDK Anthropic Tool Use Docs
Domains	Tool_Use, LLM, Function_Calling
Last Updated	2026-02-15 00:00 GMT

Overview

BetaFunctionTool.call() executes a registered tool function with the model-provided arguments, performing Pydantic-based input validation before invocation. It serves as the execution bridge between the model's ToolUseBlock.input dict and the Python function that was decorated with @beta_tool.

API Signature

Sync:

class BetaFunctionTool(BaseFunctionTool[FunctionT]):
    def call(self, input: object) -> BetaFunctionToolResultType: ...

Async:

class BetaAsyncFunctionTool(BaseFunctionTool[AsyncFunctionT]):
    async def call(self, input: object) -> BetaFunctionToolResultType: ...

Source Location

• Sync: src/anthropic/lib/tools/_beta_functions.py, lines 178-188
• Async: src/anthropic/lib/tools/_beta_functions.py, lines 192-202

Import

from anthropic import beta_tool, beta_async_tool
# The .call() method is on the BetaFunctionTool / BetaAsyncFunctionTool instances
# returned by the decorators

Parameters

Parameter	Type	Description
input	object	The arguments dict from ToolUseBlock.input. Must be a dict at runtime.

Return Value

BetaFunctionToolResultType = Union[str, Iterable[BetaContent]]

Where BetaContent is imported from anthropic.types.beta.beta_tool_result_block_param and represents structured content blocks (text, image, etc.).

The return value is what gets placed into the content field of a ToolResultBlockParam.

Implementation Details

Sync implementation (lines 178-188):

class BetaFunctionTool(BaseFunctionTool[FunctionT]):
    def call(self, input: object) -> BetaFunctionToolResultType:
        if iscoroutinefunction(self.func):
            raise RuntimeError(
                "Cannot call a coroutine function synchronously. Use `@async_tool` instead."
            )

        if not is_dict(input):
            raise TypeError(
                f"Input must be a dictionary, got {type(input).__name__}"
            )

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

Async implementation (lines 192-202):

class BetaAsyncFunctionTool(BaseFunctionTool[AsyncFunctionT]):
    async def call(self, input: object) -> BetaFunctionToolResultType:
        if not iscoroutinefunction(self.func):
            raise RuntimeError(
                "Cannot call a synchronous function asynchronously. Use `@tool` instead."
            )

        if not is_dict(input):
            raise TypeError(
                f"Input must be a dictionary, got {type(input).__name__}"
            )

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

Execution Flow

1. Coroutine check: Verifies the function matches the tool type (sync vs async)
2. Dict check: Validates that input is a dictionary (since the API always provides a dict)
3. Argument unpacking: The dict is unpacked as **kwargs to the Pydantic-validated wrapper
4. Pydantic validation: self._func_with_validate (created via pydantic.validate_call(func)) validates types, required fields, and constraints
5. Function execution: If validation passes, the original function body runs
6. Result return: The function's return value (string or content blocks) is returned

Usage Example

from anthropic import beta_tool

@beta_tool
def get_weather(city: str, unit: str = "celsius") -> str:
    """Get the current weather for a city.

    Args:
        city: The city name
        unit: Temperature unit (celsius or fahrenheit)
    """
    return f"The weather in {city} is 22 degrees {unit}"

# Simulating model-provided input
model_input = {"city": "London", "unit": "celsius"}

# Execute the tool
result = get_weather.call(model_input)
# Returns: "The weather in London is 22 degrees celsius"

In a tool use loop:

for block in response.content:
    if block.type == "tool_use":
        try:
            result = get_weather.call(block.input)
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": str(result),
            })
        except (ValueError, TypeError) as e:
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": repr(e),
                "is_error": True,
            })

Async example:

from anthropic import beta_async_tool

@beta_async_tool
async def fetch_data(url: str) -> str:
    """Fetch data from a URL."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(url)
        return resp.text

# In an async context
result = await fetch_data.call({"url": "https://example.com/data"})

Error Conditions

Error	Condition	Cause
RuntimeError	"Cannot call a coroutine function synchronously..."	BetaFunctionTool.call() invoked on an async function
RuntimeError	"Cannot call a synchronous function asynchronously..."	BetaAsyncFunctionTool.call() invoked on a sync function
TypeError	"Input must be a dictionary, got {type}"	input argument is not a dict
ValueError	"Invalid arguments for function {name}"	Pydantic validation fails (wrong types, missing required args, etc.)

The Validated Wrapper

The validation wrapper is created at tool registration time in BaseFunctionTool.__init__() (line 87):

self._func_with_validate = pydantic.validate_call(func)

This wraps the original function so that every call validates arguments against the function's type annotations. The wrapper handles:

• Type coercion (e.g., string "42" to int 42 where appropriate)
• Required parameter enforcement
• Default value application
• Enum/Literal constraint checking

Dependencies

• pydantic (v2): validate_call for argument validation, ValidationError for error handling
• inspect: iscoroutinefunction for async detection

Related Pages

Implements Principle

• Principle:Anthropics_Anthropic_sdk_python_Tool_Execution