Documentation to use Flow as Tool about langflow HOT 3 OPEN

from typing import Any, List, Optional, Type
from langchain.tools import BaseTool
from langflow.graph.vertex.base import Vertex
from langflow.helpers.flow import build_schema_from_inputs, get_arg_names, get_flow_inputs, run_flow
from langflow.base.flow_processing.utils import build_records_from_result_data, format_flow_output_records
from langchain_core.tools import ToolException
from pydantic.v1 import BaseModel

class FlowTool(BaseTool):
    name: str
    description: str
    graph: Optional[Graph] = None
    flow_id: Optional[str] = None
    user_id: Optional[str] = None
    inputs: List["Vertex"] = []
    get_final_results_only: bool = True

    @property
    def args(self) -> dict:
        schema = self.get_input_schema()
        return schema.schema()["properties"]

    def get_input_schema(self, config: Optional[RunnableConfig] = None) -> Type[BaseModel]:
        """The tool's input schema."""
        if self.args_schema is not None:
            return self.args_schema
        elif self.graph is not None:
            return build_schema_from_inputs(self.name, get_flow_inputs(self.graph))
        else:
            raise ToolException("No input schema available.")

    def _run(
        self,
        *args: Any,
        **kwargs: Any,
    ) -> str:
        """Use the tool."""
        args_names = get_arg_names(self.inputs)
        if len(args_names) == len(args):
            kwargs = {arg["arg_name"]: arg_value for arg, arg_value in zip(args_names, args)}
        elif len(args_names) != len(args) and len(args) != 0:
            raise ToolException(
                "Number of arguments does not match the number of inputs. Pass keyword arguments instead."
            )
        tweaks = {arg["component_name"]: kwargs[arg["arg_name"]] for arg in args_names}

        run_outputs = syncify(run_flow, raise_sync_error=False)(
            tweaks={key: {"input_value": value} for key, value in tweaks.items()},
            flow_id=self.flow_id,
            user_id=self.user_id,
        )
        if not run_outputs:
            return "No output"
        run_output = run_outputs[0]

        records = []
        if run_output is not None:
            for output in run_output.outputs:
                if output:
                    records.extend(
                        build_records_from_result_data(output, get_final_results_only=self.get_final_results_only)
                    )
        return format_flow_output_records(records)

from typing import Any, List, Optional
from loguru import logger
from langflow.base.tools.flow_tool import FlowTool
from langflow.custom import CustomComponent
from langflow.field_typing import Tool
from langflow.graph.graph.base import Graph
from langflow.helpers.flow import get_flow_inputs
from langflow.schema import Record
from langflow.schema.dotdict import dotdict

class FlowToolComponent(CustomComponent):
    display_name = "Flow as Tool"
    description = "Construct a Tool from a function that runs the loaded Flow."
    field_order = ["flow_name", "name", "description", "return_direct"]

    def get_flow_names(self) -> List[str]:
        flow_records = self.list_flows()
        return [flow_record.data["name"] for flow_record in flow_records]

    def get_flow(self, flow_name: str) -> Optional[Record]:
        flow_records = self.list_flows()
        for flow_record in flow_records:
            if flow_record.data["name"] == flow_name:
                return flow_record
        return None

    def update_build_config(self, build_config: dotdict, field_value: Any, field_name: str | None = None):
        logger.debug(f"Updating build config with field value {field_value} and field name {field_name}")
        if field_name == "flow_name":
            build_config["flow_name"]["options"] = self.get_flow_names()

        return build_config

    def build_config(self):
        return {
            "flow_name": {
                "display_name": "Flow Name",
                "info": "The name of the flow to run.",
                "options": [],
                "real_time_refresh": True,
                "refresh_button": True,
            },
            "name": {
                "display_name": "Name",
                "description": "The name of the tool.",
            },
            "description": {
                "display_name": "Description",
                "description": "The description of the tool.",
            },
            "return_direct": {
                "display_name": "Return Direct",
                "description": "Return the result directly from the Tool.",
                "advanced": True,
            },
        }

    async def build(self, flow_name: str, name: str, description: str, return_direct: bool = False) -> Tool:
        FlowTool.update_forward_refs()
        flow_record = self.get_flow(flow_name)
        if not flow_record:
            raise ValueError("Flow not found.")
        graph = Graph.from_payload(flow_record.data["data"])
        inputs = get_flow_inputs(graph)
        tool = FlowTool(
            name=name,
            description=description,
            graph=graph,
            return_direct=return_direct,
            inputs=inputs,
            flow_id=str(flow_record.id),
            user_id=str(self._user_id),
        )
        description_repr = repr(tool.description).strip("'")
        args_str = "\n".join([f"- {arg_name}: {arg_data['description']}" for arg_name, arg_data in tool.args.items()])
        self.status = f"{description_repr}\nArguments:\n{args_str}"
        return tool  # type: ignore