Add SearchApi tools

Author: SebastjanPrachovskij

api/core/tools/provider/_position.yaml

@@ -1,6 +1,7 @@
 - google
 - bing
 - duckduckgo
+- searchapi
 - searxng
 - dalle
 - azuredalle

api/core/tools/provider/builtin/searchapi/_assets/icon.svg

@@ -0,0 +1,23 @@
+from typing import Any
+
+from core.tools.errors import ToolProviderCredentialValidationError
+from core.tools.provider.builtin.searchapi.tools.google import GoogleTool
+from core.tools.provider.builtin_tool_provider import BuiltinToolProviderController
+
+
+class SearchAPIProvider(BuiltinToolProviderController):
+    def _validate_credentials(self, credentials: dict[str, Any]) -> None:
+        try:
+            GoogleTool().fork_tool_runtime(
+                meta={
+                    "credentials": credentials,
+                }
+            ).invoke(
+                user_id='',
+                tool_parameters={
+                    "query": "SearchApi dify",
+                    "result_type": "link"
+                },
+            )
+        except Exception as e:
+            raise ToolProviderCredentialValidationError(str(e))

api/core/tools/provider/builtin/searchapi/searchapi.py

@@ -0,0 +1,29 @@
+identity:
+  author: SearchApi
+  name: searchapi
+  label:
+    en_US: SearchApi
+    zh_Hans: SearchApi
+    pt_BR: SearchApi
+  description:
+    en_US: SearchApi
+    zh_Hans: SearchApi
+    pt_BR: SearchApi
+  icon: icon.svg
+credentials_for_provider:
+  searchapi_api_key:
+    type: secret-input
+    required: true
+    label:
+      en_US: SearchApi API key
+      zh_Hans: SearchApi API key
+      pt_BR: SearchApi API key
+    placeholder:
+      en_US: Please input your SearchApi API key
+      zh_Hans: 请输入你的 SearchApi API key
+      pt_BR: Please input your SearchApi API key
+    help:
+      en_US: Get your SearchApi API key from SearchApi
+      zh_Hans: 从 SearchApi 获取您的 SearchApi API key
+      pt_BR: Get your SearchApi API key from SearchApi
+    url: https://www.searchapi.io/

api/core/tools/provider/builtin/searchapi/searchapi.yaml

@@ -0,0 +1,102 @@
+from typing import Any, Union
+
+import requests
+
+from core.tools.entities.tool_entities import ToolInvokeMessage
+from core.tools.tool.builtin_tool import BuiltinTool
+
+SEARCH_API_URL = "https://www.searchapi.io/api/v1/search"
+
+class SearchAPI:
+    """
+    SearchAPI tool provider.
+    """
+
+    def __init__(self, api_key: str) -> None:
+        """Initialize SearchAPI tool provider."""
+        self.searchapi_api_key = api_key
+
+    def run(self, query: str, **kwargs: Any) -> str:
+        """Run query through SearchAPI and parse result."""
+        type = kwargs.get("result_type", "text")
+        return self._process_response(self.results(query, **kwargs), type=type)
+
+    def results(self, query: str, **kwargs: Any) -> dict:
+        """Run query through SearchAPI and return the raw result."""
+        params = self.get_params(query, **kwargs)
+        response = requests.get(
+            url=SEARCH_API_URL,
+            params=params,
+            headers={"Authorization": f"Bearer {self.searchapi_api_key}"},
+        )
+        response.raise_for_status()
+        return response.json()
+
+    def get_params(self, query: str, **kwargs: Any) -> dict[str, str]:
+        """Get parameters for SearchAPI."""
+        return {
+            "engine": "google",
+            "q": query,
+            **{key: value for key, value in kwargs.items() if value not in [None, ""]},
+        }
+
+    @staticmethod
+    def _process_response(res: dict, type: str) -> str:
+        """Process response from SearchAPI."""
+        if "error" in res.keys():
+            raise ValueError(f"Got error from SearchApi: {res['error']}")
+        
+        toret = ""
+        if type == "text":
+            if "answer_box" in res.keys() and "answer" in res["answer_box"].keys():
+                toret += res["answer_box"]["answer"] + "\n"
+            if "answer_box" in res.keys() and "snippet" in res["answer_box"].keys():
+                toret += res["answer_box"]["snippet"] + "\n"
+            if "knowledge_graph" in res.keys() and "description" in res["knowledge_graph"].keys():
+                toret += res["knowledge_graph"]["description"] + "\n"
+            if "organic_results" in res.keys() and "snippet" in res["organic_results"][0].keys():
+                for item in res["organic_results"]:
+                    toret += "content: " + item["snippet"] + "\n" + "link: " + item["link"] + "\n"
+            if toret == "":
+                toret = "No good search result found"
+
+        elif type == "link":
+            if "answer_box" in res.keys() and "organic_result" in res["answer_box"].keys():
+                if "title" in res["answer_box"]["organic_result"].keys():
+                    toret = f"[{res['answer_box']['organic_result']['title']}]({res['answer_box']['organic_result']['link']})\n" 
+            elif "organic_results" in res.keys() and "link" in res["organic_results"][0].keys():
+                toret = ""
+                for item in res["organic_results"]:
+                    toret += f"[{item['title']}]({item['link']})\n"
+            elif "related_questions" in res.keys() and "link" in res["related_questions"][0].keys():
+                toret = ""
+                for item in res["related_questions"]:
+                    toret += f"[{item['title']}]({item['link']})\n"
+            elif "related_searches" in res.keys() and "link" in res["related_searches"][0].keys():
+                toret = ""
+                for item in res["related_searches"]:
+                    toret += f"[{item['title']}]({item['link']})\n"
+            else:
+                toret = "No good search result found"
+        return toret
+
+class GoogleTool(BuiltinTool):
+    def _invoke(self, 
+                user_id: str,
+                tool_parameters: dict[str, Any], 
+        ) -> Union[ToolInvokeMessage, list[ToolInvokeMessage]]:
+        """
+        Invoke the SearchApi tool.
+        """
+        query = tool_parameters['query']
+        result_type = tool_parameters['result_type']
+        num = tool_parameters.get("num", 10)
+        google_domain = tool_parameters.get("google_domain", "google.com")
+        gl = tool_parameters.get("gl", "us")
+        hl = tool_parameters.get("hl", "en")
+        location = tool_parameters.get("location", None)
+        
+        api_key = self.runtime.credentials['searchapi_api_key']
+        result = SearchAPI(api_key).run(query, result_type=result_type, num=num, google_domain=google_domain, gl=gl, hl=hl, location=location)
+        
+        return self.create_text_message(text=result)

api/core/tools/provider/builtin/searchapi/tools/google.py

@@ -0,0 +1,99 @@
+identity:
+  name: google_search_api
+  author: SearchApi
+  label:
+    en_US: Google Search API
+    zh_Hans: Google Search API
+description:
+  human:
+    en_US: A tool to retrieve answer boxes, knowledge graphs, snippets, and webpages from Google Search engine.
+    zh_Hans: 一种从 Google 搜索引擎检索答案框、知识图、片段和网页的工具。
+  llm: A tool to retrieve answer boxes, knowledge graphs, snippets, and webpages from Google Search engine.
+parameters:
+  - name: query
+    type: string
+    required: true
+    label:
+      en_US: query
+      zh_Hans: 询问
+    human_description:
+      en_US: Defines the query you want to search.
+      zh_Hans: 定义您要搜索的查询。
+    llm_description: Defines the search query you want to search.
+    form: llm
+  - name: result_type
+    type: select
+    required: true
+    options:
+      - value: text
+        label:
+          en_US: text
+          zh_Hans: 文本
+      - value: link
+        label:
+          en_US: link
+          zh_Hans: 链接
+    default: text
+    label:
+      en_US: Result type
+      zh_Hans: 结果类型
+    human_description:
+      en_US: used for selecting the result type, text or link
+      zh_Hans: 用于选择结果类型，使用文本还是链接进行展示
+    form: form
+  - name: location
+    type: string
+    required: false
+    label:
+      en_US: location
+      zh_Hans: 询问
+    human_description:
+      en_US: Defines from where you want the search to originate. (For example - New York)
+      zh_Hans: 定义您想要搜索的起始位置。 （例如 - 纽约）
+    llm_description: Defines from where you want the search to originate. (For example - New York)
+    form: llm
+  - name: gl
+    type: string
+    required: false
+    label:
+      en_US: gl
+      zh_Hans: gl
+    human_description:
+      en_US: Defines the country of the search. Default is "US".
+      zh_Hans: 定义搜索的国家/地区。默认为“美国”。
+    llm_description: Defines the gl parameter of the Google News search.
+    form: llm
+  - name: hl
+    type: string
+    required: false
+    label:
+      en_US: hl
+      zh_Hans: hl
+    human_description:
+      en_US: Defines the interface language of the search. Default is "en".
+      zh_Hans: 定义搜索的界面语言。默认为“en”。
+    llm_description: Defines the hl parameter of the Google News search.
+    form: llm
+  - name: google_domain
+    type: string
+    required: false
+    label:
+      en_US: google_domain
+      zh_Hans: google_domain
+    human_description:
+      en_US: Defines the Google domain of the search. Default is "google.com".
+      zh_Hans: 定义搜索的 Google 域。默认为“google.com”。
+    llm_description: Defines Google domain in which you want to search.
+    form: llm
+  - name: num
+    type: number
+    required: false
+    label:
+      en_US: num
+      zh_Hans: num
+    human_description:
+      en_US: Specifies the number of results to display per page. Default is 10. Max number - 100, min - 1.
+      zh_Hans: 指定每页显示的结果数。默认值为 10。最大数量 - 100，最小数量 - 1。
+      pt_BR: Specifies the number of results to display per page. Default is 10. Max number - 100, min - 1.
+    llm_description: Specifies the num of results to display per page.
+    form: llm

api/core/tools/provider/builtin/searchapi/tools/google.yaml

@@ -0,0 +1,86 @@
+from typing import Any, Union
+
+import requests
+
+from core.tools.entities.tool_entities import ToolInvokeMessage
+from core.tools.tool.builtin_tool import BuiltinTool
+
+SEARCH_API_URL = "https://www.searchapi.io/api/v1/search"
+
+class SearchAPI:
+    """
+    SearchAPI tool provider.
+    """
+
+    def __init__(self, api_key: str) -> None:
+        """Initialize SearchAPI tool provider."""
+        self.searchapi_api_key = api_key
+
+    def run(self, query: str, **kwargs: Any) -> str:
+        """Run query through SearchAPI and parse result."""
+        type = kwargs.get("result_type", "text")
+        return self._process_response(self.results(query, **kwargs), type=type)
+
+    def results(self, query: str, **kwargs: Any) -> dict:
+        """Run query through SearchAPI and return the raw result."""
+        params = self.get_params(query, **kwargs)
+        response = requests.get(
+            url=SEARCH_API_URL,
+            params=params,
+            headers={"Authorization": f"Bearer {self.searchapi_api_key}"},
+        )
+        response.raise_for_status()
+        return response.json()
+
+    def get_params(self, query: str, **kwargs: Any) -> dict[str, str]:
+        """Get parameters for SearchAPI."""
+        return {
+            "engine": "google_jobs",
+            "q": query,
+            **{key: value for key, value in kwargs.items() if value not in [None, ""]},
+        }
+
+    @staticmethod
+    def _process_response(res: dict, type: str) -> str:
+        """Process response from SearchAPI."""
+        if "error" in res.keys():
+            raise ValueError(f"Got error from SearchApi: {res['error']}")
+
+        toret = ""
+        if type == "text":
+            if "jobs" in res.keys() and "title" in res["jobs"][0].keys():
+                for item in res["jobs"]:
+                    toret += "title: " + item["title"] + "\n" + "company_name: " + item["company_name"] + "content: " + item["description"] + "\n" 
+            if toret == "":
+                toret = "No good search result found"
+
+        elif type == "link":
+            if "jobs" in res.keys() and "apply_link" in res["jobs"][0].keys():
+                for item in res["jobs"]:
+                    toret += f"[{item['title']} - {item['company_name']}]({item['apply_link']})\n"
+            else:
+                toret = "No good search result found"
+        return toret
+
+class GoogleJobsTool(BuiltinTool):
+    def _invoke(self, 
+                user_id: str,
+                tool_parameters: dict[str, Any], 
+        ) -> Union[ToolInvokeMessage, list[ToolInvokeMessage]]:
+        """
+        Invoke the SearchApi tool.
+        """
+        query = tool_parameters['query']
+        result_type = tool_parameters['result_type']
+        is_remote = tool_parameters.get("is_remote", None)
+        google_domain = tool_parameters.get("google_domain", "google.com")
+        gl = tool_parameters.get("gl", "us")
+        hl = tool_parameters.get("hl", "en")
+        location = tool_parameters.get("location", None)
+
+        ltype = 1 if is_remote else None
+
+        api_key = self.runtime.credentials['searchapi_api_key']
+        result = SearchAPI(api_key).run(query, result_type=result_type, google_domain=google_domain, gl=gl, hl=hl, location=location, ltype=ltype)
+        
+        return self.create_text_message(text=result)