fix: Update templates and include global variables (#3755)

* update templates

* update to include global variables

* Refactor code to include global variables

* update PythonREPLTool.py

* [autofix.ci] apply automated fixes

* update pythonREPL and example

* Refactor code to handle decoding chat messages and handle decoding errors

* ✨ (Simple Agent.spec.ts): Add test case to fill textarea with specific text for testing purposes
📝 (Simple Agent.spec.ts): Update test case descriptions for better clarity and accuracy
✅ (Simple Agent.spec.ts): Update test assertions to match the expected behavior of the test case

* [autofix.ci] apply automated fixes

* 🐛 (Dynamic Agent.spec.ts): fix environment variable name from BRAVE_SEARCH_API_KEY to SEARCH_API_KEY for consistency and clarity
💡 (Dynamic Agent.spec.ts): add additional test cases to improve test coverage and ensure specific text is not present in the chat output

* 🔧 (.github/workflows/typescript_test.yml): update environment variable name from BRAVE_SEARCH_API_KEY to SEARCH_API_KEY for consistency
🐛 (Travel Planning Agent.spec.ts): fix test to skip if SEARCH_API_KEY is not available in the environment variables

* ✅ (Simple Agent.spec.ts): update expected count of python words to 3 for accurate test validation

* updating search tools

* [autofix.ci] apply automated fixes

* change examples

* update travel planning to include global variable

* Refactor search API component to include result limiting

* 📝 (Travel Planning Agent.spec.ts): remove unnecessary empty line to improve code readability and consistency

* test: adjusts asserts to make the test pass successfully

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: cristhianzl <cristhian.lousa@gmail.com>
Co-authored-by: italojohnny <italojohnnydosanjos@gmail.com>

.github/workflows/typescript_test.yml

@@ -45,7 +45,7 @@ jobs:
     env:
       OPENAI_API_KEY: ${{ inputs.openai_api_key || secrets.OPENAI_API_KEY }}
       STORE_API_KEY: ${{ inputs.store_api_key || secrets.STORE_API_KEY }}
-      BRAVE_SEARCH_API_KEY: "${{ secrets.BRAVE_SEARCH_API_KEY }}"
+      SEARCH_API_KEY: "${{ secrets.SEARCH_API_KEY }}"
       ASTRA_DB_APPLICATION_TOKEN: "${{ secrets.ASTRA_DB_APPLICATION_TOKEN }}"
       ASTRA_DB_API_ENDPOINT: "${{ secrets.ASTRA_DB_API_ENDPOINT }}"
     outputs:

src/backend/base/langflow/components/tools/PythonREPLTool.py

@@ -1,13 +1,12 @@
 import importlib
-from typing import cast
-
-from langchain_experimental.utilities import PythonREPL
-
+from typing import List, Union
+from pydantic import BaseModel, Field
 from langflow.base.langchain_utilities.model import LCToolComponent
+from langflow.inputs import StrInput
+from langflow.schema import Data
 from langflow.field_typing import Tool
-from langflow.io import MessageTextInput, MultiselectInput
-from langflow.schema.data import Data
-from langflow.template.field.base import Output
+from langchain.tools import StructuredTool
+from langchain_experimental.utilities import PythonREPL
 
 
 class PythonREPLToolComponent(LCToolComponent):
@@ -16,40 +15,45 @@ class PythonREPLToolComponent(LCToolComponent):
     name = "PythonREPLTool"
 
     inputs = [
-        MessageTextInput(name="input_value", display_name="Input", value=""),
-        MessageTextInput(name="name", display_name="Name", value="python_repl"),
-        MessageTextInput(
+        StrInput(
+            name="name",
+            display_name="Tool Name",
+            info="The name of the tool.",
+            value="python_repl",
+        ),
+        StrInput(
             name="description",
-            display_name="Description",
+            display_name="Tool Description",
+            info="A description of the tool.",
             value="A Python shell. Use this to execute python commands. Input should be a valid python command. If you want to see the output of a value, you should print it out with `print(...)`.",
         ),
-        MultiselectInput(
+        StrInput(
             name="global_imports",
             display_name="Global Imports",
-            info="A list of modules to import globally, e.g. ['math', 'numpy'].",
-            value=["math"],
-            combobox=True,
+            info="A comma-separated list of modules to import globally, e.g. 'math,numpy'.",
+            value="math",
+        ),
+        StrInput(
+            name="code",
+            display_name="Python Code",
+            info="The Python code to execute.",
+            value="print('Hello, World!')",
         ),
     ]
 
-    outputs = [
-        Output(name="api_run_model", display_name="Data", method="run_model"),
-        # Keep this for backwards compatibility
-        Output(name="tool", display_name="Tool", method="build_tool"),
-    ]
-
-    def get_globals(self, globals: list[str]) -> dict:
-        """
-        Retrieves the global variables from the specified modules.
+    class PythonREPLSchema(BaseModel):
+        code: str = Field(..., description="The Python code to execute.")
 
-        Args:
-            globals (list[str]): A list of module names.
-
-        Returns:
-            dict: A dictionary containing the global variables from the specified modules.
-        """
+    def get_globals(self, global_imports: Union[str, List[str]]) -> dict:
         global_dict = {}
-        for module in globals:
+        if isinstance(global_imports, str):
+            modules = [module.strip() for module in global_imports.split(",")]
+        elif isinstance(global_imports, list):
+            modules = global_imports
+        else:
+            raise ValueError("global_imports must be either a string or a list")
+
+        for module in modules:
             try:
                 imported_module = importlib.import_module(module)
                 global_dict[imported_module.__name__] = imported_module
@@ -58,24 +62,26 @@ def get_globals(self, globals: list[str]) -> dict:
         return global_dict
 
     def build_tool(self) -> Tool:
-        """
-        Builds a Python REPL tool.
-
-        Returns:
-            Tool: The built Python REPL tool.
-        """
         _globals = self.get_globals(self.global_imports)
         python_repl = PythonREPL(_globals=_globals)
-        return cast(
-            Tool,
-            Tool(
-                name=self.name,
-                description=self.description,
-                func=python_repl.run,
-            ),
+
+        def run_python_code(code: str) -> str:
+            try:
+                return python_repl.run(code)
+            except Exception as e:
+                return f"Error: {str(e)}"
+
+        tool = StructuredTool.from_function(
+            name=self.name,
+            description=self.description,
+            func=run_python_code,
+            args_schema=self.PythonREPLSchema,
         )
 
-    def run_model(self) -> Data:
+        self.status = f"Python REPL Tool created with global imports: {self.global_imports}"
+        return tool
+
+    def run_model(self) -> List[Data]:
         tool = self.build_tool()
-        result = tool.invoke(self.input_value)
-        return Data(text=result)
+        result = tool.run(self.code)
+        return [Data(data={"result": result})]

src/backend/base/langflow/components/tools/SearchAPI.py

@@ -1,16 +1,16 @@
-from typing import Union
-
+from typing import Dict, Any, Optional, List
+from pydantic import BaseModel, Field
 from langchain_community.utilities.searchapi import SearchApiAPIWrapper
-
 from langflow.base.langchain_utilities.model import LCToolComponent
-from langflow.inputs import SecretStrInput, MultilineInput, DictInput, MessageTextInput
+from langflow.inputs import SecretStrInput, MultilineInput, DictInput, MessageTextInput, IntInput
 from langflow.schema import Data
 from langflow.field_typing import Tool
+from langchain.tools import StructuredTool
 
 
 class SearchAPIComponent(LCToolComponent):
     display_name: str = "Search API"
-    description: str = "Call the searchapi.io API"
+    description: str = "Call the searchapi.io API with result limiting"
     name = "SearchAPI"
     documentation: str = "https://www.searchapi.io/docs/google"
 
@@ -22,23 +22,62 @@ class SearchAPIComponent(LCToolComponent):
             display_name="Input",
         ),
         DictInput(name="search_params", display_name="Search parameters", advanced=True, is_list=True),
+        IntInput(name="max_results", display_name="Max Results", value=5, advanced=True),
+        IntInput(name="max_snippet_length", display_name="Max Snippet Length", value=100, advanced=True),
     ]
 
-    def run_model(self) -> Union[Data, list[Data]]:
-        wrapper = self._build_wrapper()
-        results = wrapper.results(query=self.input_value, **(self.search_params or {}))
-        list_results = results.get("organic_results", [])
-        data = [Data(data=result, text=result["snippet"]) for result in list_results]
-        self.status = data
-        return data
+    class SearchAPISchema(BaseModel):
+        query: str = Field(..., description="The search query")
+        params: Optional[Dict[str, Any]] = Field(default_factory=dict, description="Additional search parameters")
+        max_results: int = Field(5, description="Maximum number of results to return")
+        max_snippet_length: int = Field(100, description="Maximum length of each result snippet")
+
+    def _build_wrapper(self):
+        return SearchApiAPIWrapper(engine=self.engine, searchapi_api_key=self.api_key)
 
     def build_tool(self) -> Tool:
         wrapper = self._build_wrapper()
-        return Tool(
+
+        def search_func(
+            query: str, params: Optional[Dict[str, Any]] = None, max_results: int = 5, max_snippet_length: int = 100
+        ) -> List[Dict[str, Any]]:
+            params = params or {}
+            full_results = wrapper.results(query=query, **params)
+            organic_results = full_results.get("organic_results", [])[:max_results]
+
+            limited_results = []
+            for result in organic_results:
+                limited_result = {
+                    "title": result.get("title", "")[:max_snippet_length],
+                    "link": result.get("link", ""),
+                    "snippet": result.get("snippet", "")[:max_snippet_length],
+                }
+                limited_results.append(limited_result)
+
+            return limited_results
+
+        tool = StructuredTool.from_function(
             name="search_api",
-            description="Search for recent results.",
-            func=lambda x: wrapper.run(query=x, **(self.search_params or {})),
+            description="Search for recent results using searchapi.io with result limiting",
+            func=search_func,
+            args_schema=self.SearchAPISchema,
         )
 
-    def _build_wrapper(self):
-        return SearchApiAPIWrapper(engine=self.engine, searchapi_api_key=self.api_key)
+        self.status = f"Search API Tool created with engine: {self.engine}"
+        return tool
+
+    def run_model(self) -> List[Data]:
+        tool = self.build_tool()
+        results = tool.run(
+            {
+                "query": self.input_value,
+                "params": self.search_params or {},
+                "max_results": self.max_results,
+                "max_snippet_length": self.max_snippet_length,
+            }
+        )
+
+        data_list = [Data(data=result, text=result.get("snippet", "")) for result in results]
+
+        self.status = data_list
+        return data_list

src/backend/base/langflow/components/tools/SerpAPI.py

@@ -1,14 +1,16 @@
+from typing import Dict, Any, Optional, List
+from pydantic import BaseModel, Field
 from langchain_community.utilities.serpapi import SerpAPIWrapper
-
 from langflow.base.langchain_utilities.model import LCToolComponent
-from langflow.inputs import SecretStrInput, DictInput, MultilineInput
+from langflow.inputs import SecretStrInput, DictInput, MultilineInput, IntInput
 from langflow.schema import Data
 from langflow.field_typing import Tool
+from langchain.tools import StructuredTool
 
 
 class SerpAPIComponent(LCToolComponent):
     display_name = "Serp Search API"
-    description = "Call Serp Search API"
+    description = "Call Serp Search API with result limiting"
     name = "SerpAPI"
 
     inputs = [
@@ -18,26 +20,71 @@ class SerpAPIComponent(LCToolComponent):
             display_name="Input",
         ),
         DictInput(name="search_params", display_name="Parameters", advanced=True, is_list=True),
+        IntInput(name="max_results", display_name="Max Results", value=5, advanced=True),
+        IntInput(name="max_snippet_length", display_name="Max Snippet Length", value=100, advanced=True),
     ]
 
-    def run_model(self) -> list[Data]:
-        wrapper = self._build_wrapper()
-        results = wrapper.results(self.input_value)
-        list_results = results.get("organic_results", [])
-        data = [Data(data=result, text=result["snippet"]) for result in list_results]
-        self.status = data
-        return data
-
-    def build_tool(self) -> Tool:
-        wrapper = self._build_wrapper()
-        return Tool(name="search_api", description="Search for recent results.", func=wrapper.run)
+    class SerpAPISchema(BaseModel):
+        query: str = Field(..., description="The search query")
+        params: Optional[Dict[str, Any]] = Field(default_factory=dict, description="Additional search parameters")
+        max_results: int = Field(5, description="Maximum number of results to return")
+        max_snippet_length: int = Field(100, description="Maximum length of each result snippet")
 
     def _build_wrapper(self) -> SerpAPIWrapper:
         if self.search_params:
-            return SerpAPIWrapper(  # type: ignore
+            return SerpAPIWrapper(
                 serpapi_api_key=self.serpapi_api_key,
                 params=self.search_params,
             )
-        return SerpAPIWrapper(  # type: ignore
-            serpapi_api_key=self.serpapi_api_key
+        return SerpAPIWrapper(serpapi_api_key=self.serpapi_api_key)
+
+    def build_tool(self) -> Tool:
+        wrapper = self._build_wrapper()
+
+        def search_func(
+            query: str, params: Optional[Dict[str, Any]] = None, max_results: int = 5, max_snippet_length: int = 100
+        ) -> List[Dict[str, Any]]:
+            params = params or {}
+            full_results = wrapper.results(query, **params)
+            organic_results = full_results.get("organic_results", [])[:max_results]
+
+            limited_results = []
+            for result in organic_results:
+                limited_result = {
+                    "title": result.get("title", "")[:max_snippet_length],
+                    "link": result.get("link", ""),
+                    "snippet": result.get("snippet", "")[:max_snippet_length],
+                }
+                limited_results.append(limited_result)
+
+            return limited_results
+
+        tool = StructuredTool.from_function(
+            name="serp_search_api",
+            description="Search for recent results using SerpAPI with result limiting",
+            func=search_func,
+            args_schema=self.SerpAPISchema,
         )
+
+        self.status = "SerpAPI Tool created"
+        return tool
+
+    def run_model(self) -> List[Data]:
+        tool = self.build_tool()
+        try:
+            results = tool.run(
+                {
+                    "query": self.input_value,
+                    "params": self.search_params or {},
+                    "max_results": self.max_results,
+                    "max_snippet_length": self.max_snippet_length,
+                }
+            )
+
+            data_list = [Data(data=result, text=result.get("snippet", "")) for result in results]
+
+            self.status = data_list
+            return data_list
+        except Exception as e:
+            self.status = f"Error: {str(e)}"
+            return [Data(data={"error": str(e)}, text=str(e))]