StructuredTool Class — langchain Architecture

Architecture documentation for the StructuredTool class in structured.py from the langchain codebase.

Class python

Entity Profile

Dependency Diagram

graph TD
  5e183a7b_f937_e2fe_e90a_7c6aea33188b["StructuredTool"]
  18e85ff8_9a5d_f800_f722_027398dc89e7["BaseTool"]
  5e183a7b_f937_e2fe_e90a_7c6aea33188b -->|extends| 18e85ff8_9a5d_f800_f722_027398dc89e7
  2812502d_d343_fb5b_dbc3_7ba2d0d9e46b["structured.py"]
  5e183a7b_f937_e2fe_e90a_7c6aea33188b -->|defined in| 2812502d_d343_fb5b_dbc3_7ba2d0d9e46b
  3dd1e452_4a28_462d_fcf8_055d4457894a["ainvoke()"]
  5e183a7b_f937_e2fe_e90a_7c6aea33188b -->|method| 3dd1e452_4a28_462d_fcf8_055d4457894a
  5b12652a_7ec9_da40_d9ab_82ddef9dd864["_run()"]
  5e183a7b_f937_e2fe_e90a_7c6aea33188b -->|method| 5b12652a_7ec9_da40_d9ab_82ddef9dd864
  455aba9b_6f3c_f719_532e_47cf052e2655["_arun()"]
  5e183a7b_f937_e2fe_e90a_7c6aea33188b -->|method| 455aba9b_6f3c_f719_532e_47cf052e2655
  e9a581e7_d840_3674_b506_f8b89ea98202["from_function()"]
  5e183a7b_f937_e2fe_e90a_7c6aea33188b -->|method| e9a581e7_d840_3674_b506_f8b89ea98202
  a109b300_4e6b_2510_d3d1_9f5f11a93081["_injected_args_keys()"]
  5e183a7b_f937_e2fe_e90a_7c6aea33188b -->|method| a109b300_4e6b_2510_d3d1_9f5f11a93081

Relationship Graph

Source Code

libs/core/langchain_core/tools/structured.py lines 40–263

class StructuredTool(BaseTool):
    """Tool that can operate on any number of inputs."""

    description: str = ""

    args_schema: Annotated[ArgsSchema, SkipValidation()] = Field(
        ..., description="The tool schema."
    )
    """The input arguments' schema."""

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

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

    # --- Runnable ---

    # TODO: Is this needed?
    @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 ---

    def _run(
        self,
        *args: Any,
        config: RunnableConfig,
        run_manager: CallbackManagerForToolRun | None = None,
        **kwargs: Any,
    ) -> Any:
        """Use the tool.

        Args:
            *args: Positional arguments to pass to the tool
            config: Configuration for the run
            run_manager: Optional callback manager to use for the run
            **kwargs: Keyword arguments to pass to the tool

        Returns:
            The result of the tool execution
        """
        if self.func:
            if run_manager and signature(self.func).parameters.get("callbacks"):
                kwargs["callbacks"] = run_manager.get_child()
            if config_param := _get_runnable_config_param(self.func):
                kwargs[config_param] = config
            return self.func(*args, **kwargs)
        msg = "StructuredTool does not support sync invocation."
        raise NotImplementedError(msg)

    async def _arun(
        self,
        *args: Any,
        config: RunnableConfig,
        run_manager: AsyncCallbackManagerForToolRun | None = None,
        **kwargs: Any,
    ) -> Any:
        """Use the tool asynchronously.

        Args:
            *args: Positional arguments to pass to the tool
            config: Configuration for the run
            run_manager: Optional callback manager to use for the run
            **kwargs: Keyword arguments to pass to the tool

        Returns:
            The result of the tool execution
        """
        if self.coroutine:
            if run_manager and signature(self.coroutine).parameters.get("callbacks"):