Skip to content

Function tool

FunctionTool

Bases: Node, Generic[T]

A tool node for executing a specified function with the given input data.

Source code in dynamiq/nodes/tools/function_tool.py
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
class FunctionTool(Node, Generic[T]):
    """
    A tool node for executing a specified function with the given input data.
    """

    group: Literal[NodeGroup.TOOLS] = NodeGroup.TOOLS
    name: str = "Function Tool"
    description: str = Field(
        default="""Executes custom Python functions as workflow tools with automatic schema generation and parameter validation.

Key Capabilities:
- Automatic input schema generation from function signatures
- Dynamic parameter validation using function type hints
- Support for both sync and async function execution
- Integration with workflow dependency systems

Usage Strategy:
- Wrap existing utility functions for workflow integration
- Create custom tool implementations without full tool classes
- Rapid prototyping of workflow components
- Build reusable function libraries for specific domains

Parameter Guide:
- Function parameters matching wrapped function signature
- Automatic schema validation based on function type hints
- Must return serializable data types for workflow compatibility

Examples:
- {"x": 10, "y": 5} (for math functions)
- {"text": "Hello World", "reverse": true} (for string operations)
- {"data": [1,2,3,4,5], "operation": "sum"} (for data processing)"""  # noqa: E501
    )
    error_handling: ErrorHandling = Field(
        default_factory=lambda: ErrorHandling(timeout_seconds=600)
    )

    def run_func(self, **_: Any) -> Any:
        """
        Execute the function logic with provided arguments.

        This method must be implemented by subclasses.

        :param kwargs: Arguments to pass to the function.
        :return: Result of the function execution.
        """
        raise NotImplementedError("run_func must be implemented by subclasses")

    def execute(self, input_data: dict[str, Any], config: RunnableConfig = None, **kwargs) -> dict[str, Any]:
        """
        Execute the tool with the provided input data and configuration.

        :param input_data: Dictionary of input data to be passed to the tool.
        :param config: Optional configuration for the runnable instance.
        :return: Dictionary with the execution result.
        """
        logger.info(f"Tool {self.name} - {self.id}: started with INPUT DATA:\n{input_data.model_dump()}")
        config = ensure_config(config)
        self.run_on_node_execute_run(config.callbacks, **kwargs)

        result = self.run_func(input_data, config=config, **kwargs)

        logger.info(f"Tool {self.name} - {self.id}: finished with RESULT:\n{str(result)[:200]}...")
        return {"content": result}

    def get_schema(self):
        """
        Generate the schema for the input and output of the tool.

        :return: Dictionary representing the input and output schema.
        """
        cls = self.__class__
        run_tool_method = self.run_func
        if hasattr(cls, "_original_func"):
            run_tool_method = cls._original_func

        signature = inspect.signature(run_tool_method)
        parameters = signature.parameters

        fields = {}
        for name, param in parameters.items():
            if name == "self":
                continue
            annotation = (
                param.annotation if param.annotation != inspect.Parameter.empty else Any
            )
            default = ... if param.default == inspect.Parameter.empty else param.default
            fields[name] = (annotation, default)

        input_model = create_model(f"{cls.__name__}Input", **fields)

        return {
            "name": self.name,
            "description": self.description,
            "input_schema": input_model.schema(),
            "output_schema": {
                "type": "object",
                "properties": {"content": {"type": "any"}},
            },
        }

execute(input_data, config=None, **kwargs)

Execute the tool with the provided input data and configuration.

:param input_data: Dictionary of input data to be passed to the tool. :param config: Optional configuration for the runnable instance. :return: Dictionary with the execution result.

Source code in dynamiq/nodes/tools/function_tool.py
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
def execute(self, input_data: dict[str, Any], config: RunnableConfig = None, **kwargs) -> dict[str, Any]:
    """
    Execute the tool with the provided input data and configuration.

    :param input_data: Dictionary of input data to be passed to the tool.
    :param config: Optional configuration for the runnable instance.
    :return: Dictionary with the execution result.
    """
    logger.info(f"Tool {self.name} - {self.id}: started with INPUT DATA:\n{input_data.model_dump()}")
    config = ensure_config(config)
    self.run_on_node_execute_run(config.callbacks, **kwargs)

    result = self.run_func(input_data, config=config, **kwargs)

    logger.info(f"Tool {self.name} - {self.id}: finished with RESULT:\n{str(result)[:200]}...")
    return {"content": result}

get_schema()

Generate the schema for the input and output of the tool.

:return: Dictionary representing the input and output schema.

Source code in dynamiq/nodes/tools/function_tool.py
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def get_schema(self):
    """
    Generate the schema for the input and output of the tool.

    :return: Dictionary representing the input and output schema.
    """
    cls = self.__class__
    run_tool_method = self.run_func
    if hasattr(cls, "_original_func"):
        run_tool_method = cls._original_func

    signature = inspect.signature(run_tool_method)
    parameters = signature.parameters

    fields = {}
    for name, param in parameters.items():
        if name == "self":
            continue
        annotation = (
            param.annotation if param.annotation != inspect.Parameter.empty else Any
        )
        default = ... if param.default == inspect.Parameter.empty else param.default
        fields[name] = (annotation, default)

    input_model = create_model(f"{cls.__name__}Input", **fields)

    return {
        "name": self.name,
        "description": self.description,
        "input_schema": input_model.schema(),
        "output_schema": {
            "type": "object",
            "properties": {"content": {"type": "any"}},
        },
    }

run_func(**_)

Execute the function logic with provided arguments.

This method must be implemented by subclasses.

:param kwargs: Arguments to pass to the function. :return: Result of the function execution.

Source code in dynamiq/nodes/tools/function_tool.py
50
51
52
53
54
55
56
57
58
59
def run_func(self, **_: Any) -> Any:
    """
    Execute the function logic with provided arguments.

    This method must be implemented by subclasses.

    :param kwargs: Arguments to pass to the function.
    :return: Result of the function execution.
    """
    raise NotImplementedError("run_func must be implemented by subclasses")

function_tool(func)

Decorator to convert a function into a FunctionTool subclass.

:param func: Function to be converted into a tool. :return: A FunctionTool subclass that wraps the provided function.

Source code in dynamiq/nodes/tools/function_tool.py
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
def function_tool(func: Callable[..., T]) -> type[FunctionTool[T]]:
    """
    Decorator to convert a function into a FunctionTool subclass.

    :param func: Function to be converted into a tool.
    :return: A FunctionTool subclass that wraps the provided function.
    """

    def create_input_schema(func) -> type[BaseModel]:
        signature = inspect.signature(func)

        params_dict = {}

        for param in signature.parameters.values():
            if param.name == "kwargs" or param.name == "config":
                continue
            if param.default is inspect.Parameter.empty:
                params_dict[param.name] = (param.annotation, ...)
            else:
                params_dict[param.name] = (param.annotation, param.default)

        return create_model(
            "FunctionToolInputSchema",
            **params_dict,
            __config__=ConfigDict(extra="allow"),
        )

    class FunctionToolFromDecorator(FunctionTool[T]):
        name: str = Field(default=func.__name__)
        description: str = Field(
            default=(func.__doc__ or "") + "\nFunction signature:" + str(inspect.signature(func))
            or f"A tool for executing the {func.__name__} function with signature: {str(inspect.signature(func))}"
        )
        _original_func = staticmethod(func)
        input_schema: ClassVar[type[BaseModel]] = create_input_schema(func)

        def run_func(self, input_data: BaseModel, **kwargs) -> T:
            return func(**input_data.model_dump(), **kwargs)

    FunctionToolFromDecorator.__name__ = func.__name__
    FunctionToolFromDecorator.__qualname__ = (
        f"FunctionToolFromDecorator({func.__name__})"
    )
    FunctionToolFromDecorator.__module__ = func.__module__

    return FunctionToolFromDecorator