eigent/backend/camel/toolkits/function_tool.py

# ========= Copyright 2023-2026 @ CAMEL-AI.org. All Rights Reserved. =========
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ========= Copyright 2023-2026 @ CAMEL-AI.org. All Rights Reserved. =========
import ast
import asyncio
import functools
import inspect
import logging
import textwrap
import threading
import warnings
from concurrent.futures import ThreadPoolExecutor
from inspect import Parameter, getsource, signature
from typing import Any, Callable, Dict, Mapping, Optional, Tuple, Type

from docstring_parser import parse
from jsonschema.exceptions import SchemaError
from jsonschema.validators import Draft202012Validator as JSONValidator
from pydantic import BaseModel, create_model
from pydantic.fields import FieldInfo

from camel.models import BaseModelBackend, ModelFactory
from camel.types import ModelPlatformType, ModelType
from camel.utils import get_pydantic_object_schema, to_pascal

logger = logging.getLogger(__name__)

# Shared thread pool for running sync tools without blocking the event loop
_SYNC_TOOL_EXECUTOR = ThreadPoolExecutor(max_workers=64)

# Persistent event loop to avoid httpx connection pool issues
_PERSISTENT_LOOP: Optional[asyncio.AbstractEventLoop] = None
_PERSISTENT_LOOP_LOCK = threading.Lock()


def _remove_a_key(d: Dict, remove_key: Any) -> None:
    r"""Remove a key from a dictionary recursively."""
    if isinstance(d, dict):
        for key in list(d.keys()):
            if key == remove_key:
                del d[key]
            else:
                _remove_a_key(d[key], remove_key)


def _remove_title_recursively(data, parent_key=None):
    r"""Recursively removes the 'title' key from all levels of a nested
    dictionary, except when 'title' is an argument name in the schema.
    """
    if isinstance(data, dict):
        # Only remove 'title' if it's not an argument name
        if parent_key not in [
            "properties",
            "$defs",
            "items",
            "allOf",
            "oneOf",
            "anyOf",
        ]:
            data.pop("title", None)

        # Recursively process each key-value pair
        for key, value in data.items():
            _remove_title_recursively(value, parent_key=key)
    elif isinstance(data, list):
        # Recursively process each element in the list
        for item in data:
            _remove_title_recursively(item, parent_key=parent_key)


def get_openai_function_schema(func: Callable) -> Dict[str, Any]:
    r"""Generates a schema dict for an OpenAI function based on its signature.

    This function is deprecated and will be replaced by
    :obj:`get_openai_tool_schema()` in future versions. It parses the
    function's parameters and docstring to construct a JSON schema-like
    dictionary.

    Args:
        func (Callable): The OpenAI function to generate the schema for.

    Returns:
        Dict[str, Any]: A dictionary representing the JSON schema of the
            function, including its name, description, and parameter
            specifications.
    """
    openai_function_schema = get_openai_tool_schema(func)["function"]
    return openai_function_schema


def get_openai_tool_schema(func: Callable) -> Dict[str, Any]:
    r"""Generates an OpenAI JSON schema from a given Python function.

    This function creates a schema compatible with OpenAI's API specifications,
    based on the provided Python function. It processes the function's
    parameters, types, and docstrings, and constructs a schema accordingly.

    Note:
        - Each parameter in `func` must have a type annotation; otherwise, it's
          treated as 'Any'.
        - Variable arguments (*args) and keyword arguments (**kwargs) are not
          supported and will be ignored.
        - A functional description including a brief and detailed explanation
          should be provided in the docstring of `func`.
        - All parameters of `func` must be described in its docstring.
        - Supported docstring styles: ReST, Google, Numpydoc, and Epydoc.

    Args:
        func (Callable): The Python function to be converted into an OpenAI
                         JSON schema.

    Returns:
        Dict[str, Any]: A dictionary representing the OpenAI JSON schema of
                        the provided function.

    See Also:
        `OpenAI API Reference
            <https://platform.openai.com/docs/api-reference/assistants/object>`_
    """
    params: Mapping[str, Parameter] = signature(func).parameters
    fields: Dict[str, Tuple[type, FieldInfo]] = {}
    for param_name, p in params.items():
        param_type = p.annotation
        param_default = p.default
        param_kind = p.kind
        param_annotation = p.annotation
        # Variable parameters are not supported
        if (
            param_kind == Parameter.VAR_POSITIONAL
            or param_kind == Parameter.VAR_KEYWORD
        ):
            continue
        # If the parameter type is not specified, it defaults to typing.Any
        if param_annotation is Parameter.empty:
            param_type = Any
        # Check if the parameter has a default value
        if param_default is Parameter.empty:
            fields[param_name] = (param_type, FieldInfo())
        else:
            fields[param_name] = (param_type, FieldInfo(default=param_default))

    # Applying `create_model()` directly will result in a mypy error,
    # create an alias to avoid this.
    def _create_mol(name, field):
        return create_model(name, **field)

    model = _create_mol(to_pascal(func.__name__), fields)
    parameters_dict = get_pydantic_object_schema(model)

    # The `"title"` is generated by `model.model_json_schema()`
    # but is useless for openai json schema, remove generated 'title' from
    # parameters_dict
    _remove_title_recursively(parameters_dict)

    docstring = parse(func.__doc__ or "")
    for param in docstring.params:
        if (name := param.arg_name) in parameters_dict["properties"] and (
            description := param.description
        ):
            # OpenAI does not allow descriptions on properties that use $ref.
            # To avoid schema errors, we only add the description if "$ref" is
            # not present.
            prop = parameters_dict["properties"][name]
            if "$ref" not in prop:
                prop["description"] = description

    short_description = docstring.short_description or ""
    long_description = docstring.long_description or ""
    if long_description:
        func_description = f"{short_description}\n{long_description}"
    else:
        func_description = short_description

    # OpenAI client.beta.chat.completions.parse for structured output has
    # additional requirements for the schema, refer:
    # https://platform.openai.com/docs/guides/structured-outputs/some-type-specific-keywords-are-not-yet-supported#supported-schemas
    parameters_dict["additionalProperties"] = False

    openai_function_schema = {
        "name": func.__name__,
        "description": func_description,
        "strict": True,
        "parameters": parameters_dict,
    }

    openai_tool_schema = {
        "type": "function",
        "function": openai_function_schema,
    }

    openai_tool_schema = sanitize_and_enforce_required(openai_tool_schema)
    return openai_tool_schema


def sanitize_and_enforce_required(parameters_dict):
    r"""Cleans and updates the function schema to conform with OpenAI's
    requirements:
    - Removes invalid 'default' fields from the parameters schema.
    - Ensures all fields are marked as required or have null type for optional
    fields.
    - Recursively adds additionalProperties: false to all nested objects.

    Args:
        parameters_dict (dict): The dictionary representing the function
            schema.

    Returns:
        dict: The updated dictionary with invalid defaults removed and all
            fields properly configured for strict mode.
    """

    def _add_additional_properties_false(obj):
        r"""Recursively add additionalProperties: false to all objects."""
        if isinstance(obj, dict):
            if (
                obj.get("type") == "object"
                and "additionalProperties" not in obj
            ):
                obj["additionalProperties"] = False

            # Process nested structures
            for key, value in obj.items():
                if key == "properties" and isinstance(value, dict):
                    for prop_value in value.values():
                        _add_additional_properties_false(prop_value)
                elif key in [
                    "items",
                    "allOf",
                    "oneOf",
                    "anyOf",
                ] and isinstance(value, (dict, list)):
                    if isinstance(value, dict):
                        _add_additional_properties_false(value)
                    elif isinstance(value, list):
                        for item in value:
                            _add_additional_properties_false(item)
                elif key == "$defs" and isinstance(value, dict):
                    for def_value in value.values():
                        _add_additional_properties_false(def_value)

    # Check if 'function' and 'parameters' exist
    if (
        'function' in parameters_dict
        and 'parameters' in parameters_dict['function']
    ):
        # Access the 'parameters' section
        parameters = parameters_dict['function']['parameters']
        properties = parameters.get('properties', {})

        # Track which fields should be required vs optional
        required_fields = []

        # Process each property
        for field_name, field_schema in properties.items():
            # Check if this field had a default value (making it optional)
            had_default = 'default' in field_schema

            # Remove 'default' key from field schema as required by OpenAI
            field_schema.pop('default', None)

            if had_default:
                # This field is optional - add null to its type
                current_type = field_schema.get('type')
                has_ref = '$ref' in field_schema
                has_any_of = 'anyOf' in field_schema

                if has_ref:
                    # Fields with $ref shouldn't have additional type field
                    # The $ref itself defines the type structure
                    pass
                elif has_any_of:
                    # Field already has anyOf
                    any_of_types = field_schema['anyOf']
                    has_null_type = any(
                        item.get('type') == 'null' for item in any_of_types
                    )
                    if not has_null_type:
                        # Add null type to anyOf
                        field_schema['anyOf'].append({'type': 'null'})
                    # Remove conflicting type field if it exists
                    if 'type' in field_schema:
                        del field_schema['type']
                elif current_type:
                    if isinstance(current_type, str):
                        # Single type - convert to array with null
                        field_schema['type'] = [current_type, 'null']
                    elif (
                        isinstance(current_type, list)
                        and 'null' not in current_type
                    ):
                        # Array of types - add null if not present
                        field_schema['type'] = [*current_type, 'null']
                else:
                    # No type specified, add null type
                    field_schema['type'] = ['null']

                # Optional fields are still marked as required in strict mode
                # but with null type to indicate they can be omitted
                required_fields.append(field_name)
            else:
                # This field is required
                required_fields.append(field_name)

        # Set all fields as required (strict mode requirement)
        parameters['required'] = required_fields

        # Recursively add additionalProperties: false to all objects
        _add_additional_properties_false(parameters)

    return parameters_dict


def generate_docstring(
    code: str,
    model: Optional[BaseModelBackend] = None,
) -> str:
    r"""Generates a docstring for a given function code using LLM.

    This function leverages a language model to generate a
    PEP 8/PEP 257-compliant docstring for a provided Python function.
    If no model is supplied, a default gpt-4o-mini is used.

    Args:
        code (str): The source code of the function.
        model (Optional[BaseModelBackend]): An optional language model backend
            instance. If not provided, a default gpt-4o-mini is used.

    Returns:
        str: The generated docstring.
    """

    from camel.agents import ChatAgent

    # Create the docstring prompt
    docstring_prompt = textwrap.dedent(
        """\
        **Role**: Generate professional Python docstrings conforming to PEP 8/PEP 257.

        **Requirements**:
        - Use appropriate format: reST, Google, or NumPy, as needed.
        - Include parameters, return values, and exceptions.
        - Reference any existing docstring in the function and retain useful information.

        **Input**: Python function.

        **Output**: Docstring content (plain text, no code markers).

        **Example:**

        Input:
        ```python
        def add(a: int, b: int) -> int:
            return a + b
        ```

        Output:
        Adds two numbers.
        Args:
            a (int): The first number.
            b (int): The second number.

        Returns:
            int: The sum of the two numbers.

        **Task**: Generate a docstring for the function below.
        """  # noqa: E501
    )
    # Initialize assistant with system message and model
    assistant_sys_msg = "You are a helpful assistant."
    docstring_assistant = ChatAgent(assistant_sys_msg, model=model)

    # Create user message to prompt the assistant
    user_msg = docstring_prompt + code

    # Get the response containing the generated docstring
    response = docstring_assistant.step(user_msg)
    return response.msg.content


class FunctionTool:
    r"""An abstraction of a function that OpenAI chat models can call. See
    https://platform.openai.com/docs/api-reference/chat/create.

    By default, the tool schema will be parsed from the func, or you can
    provide a user-defined tool schema to override.

    Args:
        func (Callable): The function to call. The tool schema is parsed from
            the function signature and docstring by default.
        openai_tool_schema (Optional[Dict[str, Any]], optional): A
            user-defined OpenAI tool schema to override the default result.
            (default: :obj:`None`)
        synthesize_schema (Optional[bool], optional): Whether to enable the
            use of a schema assistant model to automatically synthesize the
            schema if validation fails or no valid schema is provided.
            (default: :obj:`False`)
        synthesize_schema_model (Optional[BaseModelBackend], optional): An
            assistant model (e.g., an LLM model) used to synthesize the schema
            if `synthesize_schema` is enabled and no valid schema is
            provided. (default: :obj:`None`)
        synthesize_schema_max_retries (int, optional): The maximum
            number of attempts to retry schema synthesis using the schema
            assistant model if the previous attempts fail. (default: 2)
        synthesize_output (Optional[bool], optional): Flag for enabling
            synthesis output mode, where output is synthesized based on the
            function's execution. (default: :obj:`False`)
        synthesize_output_model (Optional[BaseModelBackend], optional):
            Model used for output synthesis in synthesis mode.
            (default: :obj:`None`)
        synthesize_output_format (Optional[Type[BaseModel]], optional): Format
            for the response when synthesizing output. (default: :obj:`None`)
    """

    def __init__(
        self,
        func: Callable,
        openai_tool_schema: Optional[Dict[str, Any]] = None,
        synthesize_schema: Optional[bool] = False,
        synthesize_schema_model: Optional[BaseModelBackend] = None,
        synthesize_schema_max_retries: int = 2,
        synthesize_output: Optional[bool] = False,
        synthesize_output_model: Optional[BaseModelBackend] = None,
        synthesize_output_format: Optional[Type[BaseModel]] = None,
    ) -> None:
        self.func = func
        self.openai_tool_schema = openai_tool_schema or get_openai_tool_schema(
            func
        )
        self.synthesize_output = synthesize_output
        self.synthesize_output_model = synthesize_output_model
        if synthesize_output and synthesize_output_model is None:
            self.synthesize_output_model = ModelFactory.create(
                model_platform=ModelPlatformType.DEFAULT,
                model_type=ModelType.DEFAULT,
            )
            logger.warning(
                "Warning: No synthesize_output_model provided. "
                f"Use `{self.synthesize_output_model.model_type}` to "
                "synthesize the output."
            )
        self.synthesize_output_format: Optional[type[BaseModel]] = None
        return_annotation = inspect.signature(self.func).return_annotation
        if synthesize_output_format is not None:
            self.synthesize_output_format = synthesize_output_format
        elif isinstance(return_annotation, type) and issubclass(
            return_annotation, BaseModel
        ):
            self.synthesize_output_format = return_annotation

        self.synthesize_schema_model = synthesize_schema_model
        if synthesize_schema:
            if openai_tool_schema:
                logger.warning("""The user-defined OpenAI tool schema will be
                              overridden by the schema assistant model.""")
            if self.synthesize_schema_model is None:
                self.synthesize_schema_model = ModelFactory.create(
                    model_platform=ModelPlatformType.DEFAULT,
                    model_type=ModelType.DEFAULT,
                )
                logger.warning(
                    "Warning: No synthesize_schema_model provided. "
                    f"Use `{self.synthesize_schema_model.model_type}` to "
                    "synthesize the schema."
                )
            schema = self.synthesize_openai_tool_schema(
                synthesize_schema_max_retries
            )
            if schema:
                self.openai_tool_schema = schema
            else:
                raise ValueError(
                    f"Failed to synthesize a valid schema for "
                    f"{self.func.__name__}."
                )

    def __call__(self, *args: Any, **kwargs: Any) -> Any:
        if self.synthesize_output:
            result = self.synthesize_execution_output(args, kwargs)
            return result

        # Call the function first
        try:
            result = self.func(*args, **kwargs)
        except Exception as e:
            parts = []
            if args:
                parts.append(f"args={args}")
            if kwargs:
                parts.append(f"kwargs={kwargs}")
            args_str = ", ".join(parts) if parts else "no arguments"
            raise ValueError(
                f"Execution of function {self.func.__name__} failed with "
                f"{args_str}. Error: {e}"
            )

        # Handle coroutine result (from async function or sync wrapper
        # returning coroutine)
        if inspect.iscoroutine(result):
            # Check if there's already a running event loop
            try:
                asyncio.get_running_loop()
                has_running_loop = True
            except RuntimeError:
                has_running_loop = False

            if has_running_loop:
                # Already in an async context
                warnings.warn(
                    f"Async tool '{self.func.__name__}' is being called "
                    f"synchronously within an async context. Consider using "
                    f"'await tool.async_call()' or 'await agent.astep()' for "
                    f"better performance.",
                    RuntimeWarning,
                    stacklevel=2,
                )
                # Must run in separate thread to avoid blocking current loop
                future = _SYNC_TOOL_EXECUTOR.submit(
                    self._run_async_in_persistent_loop, result
                )
                return future.result()
            else:
                warnings.warn(
                    f"Async tool '{self.func.__name__}' is being called "
                    f"synchronously. Consider using 'await tool.async_call()' "
                    f"or 'await agent.astep()' for better performance.",
                    RuntimeWarning,
                    stacklevel=2,
                )
                return self._run_async_in_persistent_loop(result)

        return result

    @staticmethod
    def _run_async_in_persistent_loop(coro):
        r"""Run coroutine in persistent loop to preserve httpx connections."""
        global _PERSISTENT_LOOP
        with _PERSISTENT_LOOP_LOCK:
            need_new_loop = (
                _PERSISTENT_LOOP is None
                or _PERSISTENT_LOOP.is_closed()
                or not _PERSISTENT_LOOP.is_running()
            )
            if need_new_loop:
                _PERSISTENT_LOOP = asyncio.new_event_loop()
                t = threading.Thread(
                    target=_PERSISTENT_LOOP.run_forever, daemon=True
                )
                t.start()
                while not _PERSISTENT_LOOP.is_running():
                    pass  # Wait for loop to start
        future = asyncio.run_coroutine_threadsafe(coro, _PERSISTENT_LOOP)
        return future.result()

    async def async_call(self, *args: Any, **kwargs: Any) -> Any:
        if self.synthesize_output:
            result = self.synthesize_execution_output(args, kwargs)
            return result

        # Check if the function itself (not unwrapped) is a coroutine function
        if inspect.iscoroutinefunction(self.func):
            return await self.func(*args, **kwargs)

        # For sync functions (including sync wrappers around async functions),
        # run in executor to avoid blocking
        loop = asyncio.get_running_loop()
        result = await loop.run_in_executor(
            _SYNC_TOOL_EXECUTOR,
            functools.partial(self.func, *args, **kwargs),
        )

        # If the sync wrapper returned a coroutine, await it
        if inspect.iscoroutine(result):
            return await result

        return result

    @property
    def is_async(self) -> bool:
        return inspect.iscoroutinefunction(inspect.unwrap(self.func))

    @staticmethod
    def validate_openai_tool_schema(
        openai_tool_schema: Dict[str, Any],
    ) -> None:
        r"""Validates the OpenAI tool schema against
        :obj:`ToolAssistantToolsFunction`.
        This function checks if the provided :obj:`openai_tool_schema` adheres
        to the specifications required by OpenAI's
        :obj:`ToolAssistantToolsFunction`. It ensures that the function
        description and parameters are correctly formatted according to JSON
        Schema specifications.
        Args:
            openai_tool_schema (Dict[str, Any]): The OpenAI tool schema to
                validate.
        Raises:
            ValidationError: If the schema does not comply with the
                specifications.
            SchemaError: If the parameters do not meet JSON Schema reference
                specifications.
        """
        # Check the type
        if not openai_tool_schema["type"]:
            raise ValueError("miss `type` in tool schema.")

        # Check the function description, if no description then raise warming
        if not openai_tool_schema["function"].get("description"):
            warnings.warn(f"""Function description is missing for
                          {openai_tool_schema['function']['name']}. This may
                          affect the quality of tool calling.""")

        # Validate whether parameters
        # meet the JSON Schema reference specifications.
        # See https://platform.openai.com/docs/guides/gpt/function-calling
        # for examples, and the
        # https://json-schema.org/understanding-json-schema/ for
        # documentation about the format.
        parameters = openai_tool_schema["function"]["parameters"]
        try:
            JSONValidator.check_schema(parameters)
        except SchemaError as e:
            raise e

        # Check the parameter description, if no description then raise warming
        properties: Dict[str, Any] = parameters["properties"]
        for param_name in properties.keys():
            param_dict = properties[param_name]
            if "description" not in param_dict:
                warnings.warn(
                    f"Parameter description is missing for the "
                    f"function '{openai_tool_schema['function']['name']}'. "
                    f"The parameter definition is {param_dict}. "
                    f"This may affect the quality of tool calling."
                )

    def get_openai_tool_schema(self) -> Dict[str, Any]:
        r"""Gets the OpenAI tool schema for this function.

        This method returns the OpenAI tool schema associated with this
        function, after validating it to ensure it meets OpenAI's
        specifications.

        Returns:
            Dict[str, Any]: The OpenAI tool schema for this function.
        """
        self.validate_openai_tool_schema(self.openai_tool_schema)
        return self.openai_tool_schema

    def set_openai_tool_schema(self, schema: Dict[str, Any]) -> None:
        r"""Sets the OpenAI tool schema for this function.

        Allows setting a custom OpenAI tool schema for this function.

        Args:
            schema (Dict[str, Any]): The OpenAI tool schema to set.
        """
        self.openai_tool_schema = schema

    def get_openai_function_schema(self) -> Dict[str, Any]:
        r"""Gets the schema of the function from the OpenAI tool schema.

        This method extracts and returns the function-specific part of the
        OpenAI tool schema associated with this function.

        Returns:
            Dict[str, Any]: The schema of the function within the OpenAI tool
                schema.
        """
        self.validate_openai_tool_schema(self.openai_tool_schema)
        return self.openai_tool_schema["function"]

    def set_openai_function_schema(
        self,
        openai_function_schema: Dict[str, Any],
    ) -> None:
        r"""Sets the schema of the function within the OpenAI tool schema.

        Args:
            openai_function_schema (Dict[str, Any]): The function schema to
                set within the OpenAI tool schema.
        """
        self.openai_tool_schema["function"] = openai_function_schema

    def get_function_name(self) -> str:
        r"""Gets the name of the function from the OpenAI tool schema.

        Returns:
            str: The name of the function.
        """
        self.validate_openai_tool_schema(self.openai_tool_schema)
        return self.openai_tool_schema["function"]["name"]

    def set_function_name(self, name: str) -> None:
        r"""Sets the name of the function in the OpenAI tool schema.

        Args:
            name (str): The name of the function to set.
        """
        self.openai_tool_schema["function"]["name"] = name

    def get_function_description(self) -> str:
        r"""Gets the description of the function from the OpenAI tool
        schema.

        Returns:
            str: The description of the function.
        """
        self.validate_openai_tool_schema(self.openai_tool_schema)
        return self.openai_tool_schema["function"]["description"]

    def set_function_description(self, description: str) -> None:
        r"""Sets the description of the function in the OpenAI tool schema.

        Args:
            description (str): The description for the function.
        """
        self.openai_tool_schema["function"]["description"] = description

    def get_parameter_description(self, param_name: str) -> str:
        r"""Gets the description of a specific parameter from the function
        schema.

        Args:
            param_name (str): The name of the parameter to get the
                description.

        Returns:
            str: The description of the specified parameter.
        """
        self.validate_openai_tool_schema(self.openai_tool_schema)
        return self.openai_tool_schema["function"]["parameters"]["properties"][
            param_name
        ]["description"]

    def set_parameter_description(
        self,
        param_name: str,
        description: str,
    ) -> None:
        r"""Sets the description for a specific parameter in the function
        schema.

        Args:
            param_name (str): The name of the parameter to set the description
                for.
            description (str): The description for the parameter.
        """
        self.openai_tool_schema["function"]["parameters"]["properties"][
            param_name
        ]["description"] = description

    def get_parameter(self, param_name: str) -> Dict[str, Any]:
        r"""Gets the schema for a specific parameter from the function schema.

        Args:
            param_name (str): The name of the parameter to get the schema.

        Returns:
            Dict[str, Any]: The schema of the specified parameter.
        """
        self.validate_openai_tool_schema(self.openai_tool_schema)
        return self.openai_tool_schema["function"]["parameters"]["properties"][
            param_name
        ]

    def set_parameter(self, param_name: str, value: Dict[str, Any]):
        r"""Sets the schema for a specific parameter in the function schema.

        Args:
            param_name (str): The name of the parameter to set the schema for.
            value (Dict[str, Any]): The schema to set for the parameter.
        """
        try:
            JSONValidator.check_schema(value)
        except SchemaError as e:
            raise e
        self.openai_tool_schema["function"]["parameters"]["properties"][
            param_name
        ] = value

    def synthesize_openai_tool_schema(
        self,
        max_retries: Optional[int] = None,
    ) -> Dict[str, Any]:
        r"""Synthesizes an OpenAI tool schema for the specified function.

        This method uses a language model (LLM) to synthesize the OpenAI tool
        schema for the specified function by first generating a docstring and
        then creating a schema based on the function's source code. The
        schema synthesis and validation process is retried up to
        `max_retries` times in case of failure.

        Args:
            max_retries (Optional[int], optional): The maximum number of
                retries for schema synthesis and validation if the process
                fails. (default: :obj:`None`)

        Returns:
            Dict[str, Any]: The synthesis OpenAI tool schema for the function.

        Raises:
            ValueError: If schema synthesis or validation fails after the
                maximum number of retries, a ValueError is raised, prompting
                manual schema setting.
        """
        code = getsource(self.func)
        retries = 0
        if max_retries is None:
            max_retries = 0
        # Retry loop to handle schema synthesis and validation
        while retries <= max_retries:
            try:
                # Generate the docstring and the schema
                docstring = generate_docstring(
                    code, self.synthesize_schema_model
                )
                self.func.__doc__ = docstring
                schema = get_openai_tool_schema(self.func)
                # Validate the schema
                self.validate_openai_tool_schema(schema)
                return schema

            except Exception as e:
                retries += 1
                if retries == max_retries:
                    raise ValueError(
                        f"Failed to synthesize the OpenAI tool Schema after "
                        f"{max_retries} retries. "
                        f"Please set the OpenAI tool schema for "
                        f"function {self.func.__name__} manually."
                    ) from e
                logger.warning("Schema validation failed. Retrying...")

        return {}

    def synthesize_execution_output(
        self,
        args: Optional[tuple[Any, ...]] = None,
        kwargs: Optional[Dict[str, Any]] = None,
    ) -> Any:
        r"""Synthesizes the output of the function based on the provided
        positional arguments and keyword arguments.

        Args:
            args (Optional[tuple]): Positional arguments to pass to the
                function during synthesis. (default: :obj:`None`)
            kwargs (Optional[Dict[str, Any]]): Keyword arguments to pass to the
                function during synthesis. (default: :obj:`None`)

        Returns:
            Any: Synthesized output from the function execution. If no
                synthesis model is provided, a warning is logged.
        """
        from camel.agents import ChatAgent

        # Retrieve the function source code
        function_string = inspect.getsource(self.func)

        # Check and update docstring if necessary
        if self.func.__doc__ is not None:
            function_string = textwrap.dedent(function_string)
            tree = ast.parse(function_string)
            func_node = (
                tree.body[0]
                if isinstance(tree.body[0], ast.FunctionDef)
                else None
            )
            if func_node:
                existing_docstring = ast.get_docstring(func_node)
                if existing_docstring != self.func.__doc__:
                    func_node.body[0] = ast.Expr(
                        value=ast.Constant(value=self.func.__doc__, kind=None)
                    )
                    function_string = ast.unparse(tree)

        # Append the args and kwargs information to the function string
        if args:
            function_string += f"\nargs:\n{list(args)}"
        if kwargs:
            function_string += f"\nkwargs:\n{kwargs}"

        # Define the assistant system message
        assistant_sys_msg = textwrap.dedent(
            '''\
            **Role:** AI Assistant specialized in synthesizing tool execution outputs without actual execution.

            **Capabilities:**
            - Analyzes function to understand their purpose and expected outputs.
            - Generates synthetic outputs based on the function logic.
            - Ensures the synthesized output is contextually accurate and aligns with the function's intended behavior.

            **Instructions:**
            1. **Input:** Provide the function code, function docstring, args, and kwargs.
            2. **Output:** Synthesize the expected output of the function based on the provided args and kwargs.

            **Example:**
            - **User Input:**
            def sum(a, b, c=0):
                """Adds three numbers together."""
                return a + b + c

            - **Input Arguments:**
            args: (1, 2)
            kwargs: {"c": 3}

            - **Output:**
            6

            **Note:**
            - Just return the synthesized output of the function without any explanation.
            - The output should be in plain text without any formatting.
            '''  # noqa: E501
        )

        # Initialize the synthesis agent
        synthesis_agent = ChatAgent(
            assistant_sys_msg,
            model=self.synthesize_output_model,
        )

        # User message combining function string and additional context
        user_msg = function_string
        response = synthesis_agent.step(
            user_msg,
            response_format=self.synthesize_output_format,
        )

        return response.msg.content

    @property
    def parameters(self) -> Dict[str, Any]:
        r"""Getter method for the property :obj:`parameters`.

        Returns:
            Dict[str, Any]: the dictionary containing information of
                parameters of this function.
        """
        self.validate_openai_tool_schema(self.openai_tool_schema)
        return self.openai_tool_schema["function"]["parameters"]["properties"]

    @parameters.setter
    def parameters(self, value: Dict[str, Any]) -> None:
        r"""Setter method for the property :obj:`parameters`. It will
        firstly check if the input parameters schema is valid. If invalid,
        the method will raise :obj:`jsonschema.exceptions.SchemaError`.

        Args:
            value (Dict[str, Any]): the new dictionary value for the
                function's parameters.
        """
        try:
            JSONValidator.check_schema(value)
        except SchemaError as e:
            raise e
        self.openai_tool_schema["function"]["parameters"]["properties"] = value


def tool(
    func: Optional[Callable] = None,
    *,
    openai_tool_schema: Optional[Dict[str, Any]] = None,
    synthesize_schema: bool = False,
    synthesize_schema_model: Optional[BaseModelBackend] = None,
    synthesize_schema_max_retries: int = 2,
    synthesize_output: bool = False,
    synthesize_output_model: Optional[BaseModelBackend] = None,
    synthesize_output_format: Optional[Type[BaseModel]] = None,
):
    r"""A decorator that converts a Python function into a FunctionTool
    instance.

    This decorator can be used with or without parentheses:
        - @tool - without parentheses, uses default settings
        - @tool() - with parentheses, uses default settings
        - @tool(synthesize_output=True) - with custom settings

    Args:
        func (Optional[Callable], optional): The function to be decorated.
            This is automatically passed when using @tool without parentheses.
            (default: :obj:`None`)
        openai_tool_schema (Optional[Dict[str, Any]], optional): A
            user-defined OpenAI tool schema to override the default result.
            (default: :obj:`None`)
        synthesize_schema (bool, optional): Whether to enable schema synthesis.
            (default: :obj:`False`)
        synthesize_schema_model (Optional[BaseModelBackend], optional):
            Model to use for schema synthesis. (default: :obj:`None`)
        synthesize_schema_max_retries (int, optional): Maximum number of
            retries for schema synthesis. (default: :obj:`2`)
        synthesize_output (bool, optional): Whether to enable output synthesis.
            (default: :obj:`False`)
        synthesize_output_model (Optional[BaseModelBackend], optional):
            Model to use for output synthesis. (default: :obj:`None`)
        synthesize_output_format (Optional[Type[BaseModel]], optional):
            Format for synthesized output. (default: :obj:`None`)

    Returns:
        Callable[[Callable], FunctionTool]: A decorator function that converts
            the decorated function into a FunctionTool instance.

    Example:
        Using @tool without parentheses::

            @tool
            def add(a: int, b: int = 0) -> int:
                '''Add two numbers.'''
                return a + b

        Using @tool() with parentheses::

            @tool()
            def multiply(a: int, b: int) -> int:
                '''Multiply two numbers.'''
                return a * b
    """

    def decorator(f: Callable) -> FunctionTool:
        r"""The actual decorator function.

        Args:
            f (Callable): The function to be converted into a FunctionTool.

        Returns:
            FunctionTool: The function wrapped as a FunctionTool instance.
        """
        return FunctionTool(
            func=f,
            openai_tool_schema=openai_tool_schema,
            synthesize_schema=synthesize_schema,
            synthesize_schema_model=synthesize_schema_model,
            synthesize_schema_max_retries=synthesize_schema_max_retries,
            synthesize_output=synthesize_output,
            synthesize_output_model=synthesize_output_model,
            synthesize_output_format=synthesize_output_format,
        )

    # Support both @tool and @tool() usage patterns
    if func is not None:
        return decorator(func)
    else:
        return decorator