community[minor]: add mojeek search util (#20922)

**Description:** This pull request introduces a new feature to community
tools, enhancing its search capabilities by integrating the Mojeek
search engine
**Dependencies:** None

---------

Co-authored-by: Igor Brai <igor@mojeek.com>
Co-authored-by: Bagatur <22008038+baskaryan@users.noreply.github.com>
Co-authored-by: ccurme <chester.curme@gmail.com>

libs/community/langchain_community/utilities/__init__.py

@@ -82,6 +82,9 @@ if TYPE_CHECKING:
     from langchain_community.utilities.metaphor_search import (
         MetaphorSearchAPIWrapper,  # noqa: F401
     )
+    from langchain_community.utilities.mojeek_search import (
+        MojeekSearchAPIWrapper,  # noqa: F401
+    )
     from langchain_community.utilities.nasa import (
         NasaAPIWrapper,  # noqa: F401
     )
@@ -189,6 +192,7 @@ __all__ = [
     "MaxComputeAPIWrapper",
     "MerriamWebsterAPIWrapper",
     "MetaphorSearchAPIWrapper",
+    "MojeekSearchAPIWrapper",
     "NVIDIARivaASR",
     "NVIDIARivaStream",
     "NVIDIARivaTTS",
@@ -249,6 +253,7 @@ _module_lookup = {
     "MaxComputeAPIWrapper": "langchain_community.utilities.max_compute",
     "MerriamWebsterAPIWrapper": "langchain_community.utilities.merriam_webster",
     "MetaphorSearchAPIWrapper": "langchain_community.utilities.metaphor_search",
+    "MojeekSearchAPIWrapper": "langchain_community.utilities.mojeek_search",
     "NVIDIARivaASR": "langchain_community.utilities.nvidia_riva",
     "NVIDIARivaStream": "langchain_community.utilities.nvidia_riva",
     "NVIDIARivaTTS": "langchain_community.utilities.nvidia_riva",

libs/community/langchain_community/utilities/mojeek_search.py

@@ -0,0 +1,44 @@
+import json
+from typing import List
+
+import requests
+from langchain_core.pydantic_v1 import BaseModel, Field
+
+
+class MojeekSearchAPIWrapper(BaseModel):
+    api_key: str
+    search_kwargs: dict = Field(default_factory=dict)
+    api_url = "https://api.mojeek.com/search"
+
+    def run(self, query: str) -> str:
+        search_results = self._search(query)
+
+        results = []
+
+        for result in search_results:
+            title = result.get("title", "")
+            url = result.get("url", "")
+            desc = result.get("desc", "")
+            results.append({"title": title, "url": url, "desc": desc})
+
+        return json.dumps(results)
+
+    def _search(self, query: str) -> List[dict]:
+        headers = {
+            "Accept": "application/json",
+        }
+
+        req = requests.PreparedRequest()
+        request = {
+            **self.search_kwargs,
+            **{"q": query, "fmt": "json", "api_key": self.api_key},
+        }
+        req.prepare_url(self.api_url, request)
+        if req.url is None:
+            raise ValueError("prepared url is None, this should not happen")
+
+        response = requests.get(req.url, headers=headers)
+        if not response.ok:
+            raise Exception(f"HTTP error {response.status_code}")
+
+        return response.json().get("response", {}).get("results", [])