simple.py — langchain Source File

Source Code

"""Tool that takes in function or coroutine directly."""

from __future__ import annotations

from collections.abc import Awaitable, Callable
from inspect import signature
from typing import (
    TYPE_CHECKING,
    Any,
)

from typing_extensions import override

# Cannot move to TYPE_CHECKING as _run/_arun parameter annotations are needed at runtime
from langchain_core.callbacks import (
    AsyncCallbackManagerForToolRun,  # noqa: TC001
    CallbackManagerForToolRun,  # noqa: TC001
)
from langchain_core.runnables import RunnableConfig, run_in_executor
from langchain_core.tools.base import (
    ArgsSchema,
    BaseTool,
    ToolException,
    _get_runnable_config_param,
)

if TYPE_CHECKING:
    from langchain_core.messages import ToolCall


class Tool(BaseTool):
    """Tool that takes in function or coroutine directly."""

    description: str = ""

    func: Callable[..., str] | None
    """The function to run when the tool is called."""

    coroutine: Callable[..., Awaitable[str]] | None = None
    """The asynchronous version of the function."""

    # --- Runnable ---

    @override
    async def ainvoke(
        self,
        input: str | dict | ToolCall,
        config: RunnableConfig | None = None,
        **kwargs: Any,
    ) -> Any:
        if not self.coroutine:
            # If the tool does not implement async, fall back to default implementation
            return await run_in_executor(config, self.invoke, input, config, **kwargs)

        return await super().ainvoke(input, config, **kwargs)

    # --- Tool ---

    @property
    def args(self) -> dict:
// ... (145 more lines)