feat(web): multi-source fallback chain and search_engine parameter

plugins/web/_360_search/__init__.py

@@ -0,0 +1 @@
+"""360 Search (好搜) web search provider plugin."""

plugins/web/_360_search/plugin.yaml

@@ -0,0 +1,7 @@
+name: web-360-search
+version: 1.0.0
+description: "360 Search (Haosou) — Chinese search engine. NOTE: no public developer API available."
+author: Hermes Community
+kind: backend
+provides_web_providers:
+  - 360-search

plugins/web/_360_search/provider.py

@@ -0,0 +1,45 @@
+"""360 Search (Haosou) — Chinese search engine. NOTE: no public developer API available."""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import Any, Dict
+
+from agent.web_search_provider import WebSearchProvider
+
+logger = logging.getLogger(__name__)
+
+
+class Three60SearchWebSearchProvider(WebSearchProvider):
+    """360 Search (好搜) — unavailable tier search provider."""
+
+    @property
+    def name(self) -> str:
+        return "360-search"
+
+    @property
+    def display_name(self) -> str:
+        return "360 Search (好搜)"
+
+    def is_available(self) -> bool:
+        """No public API available — always unavailable."""
+        return False
+
+    def supports_search(self) -> bool:
+        return True
+
+    def supports_extract(self) -> bool:
+        return False
+
+    def search(self, query: str, limit: int = 5) -> Dict[str, Any]:
+        return {"success": False, "error": "360 Search (好搜) does not offer a public search API. Use Baidu or Bocha for Chinese-language search."}
+
+    def get_setup_schema(self) -> Dict[str, Any]:
+        return {
+            "name": "360 Search (好搜)",
+            "badge": "unavailable",
+            "tag": "No public API available. Listed for registry completeness.",
+            "env_vars": [
+            ],
+        }

plugins/web/baidu/__init__.py

@@ -0,0 +1 @@
+"""Baidu Search web search provider plugin."""

plugins/web/baidu/plugin.yaml

@@ -0,0 +1,7 @@
+name: web-baidu
+version: 1.0.0
+description: "Baidu AI Search API — 100 queries/day free. Requires BAIDU_API_KEY from https://ai.baidu.com."
+author: Hermes Community
+kind: backend
+provides_web_providers:
+  - baidu

plugins/web/baidu/provider.py

@@ -0,0 +1,65 @@
+"""Baidu AI Search API — 100 queries/day free. Requires BAIDU_API_KEY from https://ai.baidu.com."""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import Any, Dict
+
+from agent.web_search_provider import WebSearchProvider
+
+logger = logging.getLogger(__name__)
+
+
+class BaiduWebSearchProvider(WebSearchProvider):
+    """Baidu Search — free tier search provider."""
+
+    @property
+    def name(self) -> str:
+        return "baidu"
+
+    @property
+    def display_name(self) -> str:
+        return "Baidu Search"
+
+    def is_available(self) -> bool:
+        """Return True when BAIDU_API_KEY is set."""
+        return bool(os.getenv("BAIDU_API_KEY", "").strip())
+
+    def supports_search(self) -> bool:
+        return True
+
+    def supports_extract(self) -> bool:
+        return False
+
+    def search(self, query: str, limit: int = 5) -> Dict[str, Any]:
+        import requests
+        api_key = os.getenv("BAIDU_API_KEY", "").strip()
+        if not api_key:
+            return {"success": False, "error": "BAIDU_API_KEY not set. Get a key at https://ai.baidu.com."}
+        try:
+            r = requests.post("https://api.baidu.com/search/v1/websearch", params={"q": query, "topn": min(limit, 50)}, headers={"Authorization": f"Bearer {api_key}"}, timeout=10)
+            r.raise_for_status()
+            data = r.json()
+            raw = data.get("results", [])[:limit]
+            results = [
+                {"title": str(i.get("title", "")), "url": str(i.get("url", "")),
+                 "description": str(i.get("summary", "")), "position": n + 1}
+                for n, i in enumerate(raw)
+            ]
+            return {"success": True, "data": {"web": results}}
+        except Exception as e:
+            return {"success": False, "error": str(e)}
+    def get_setup_schema(self) -> Dict[str, Any]:
+        return {
+            "name": "Baidu Search",
+            "badge": "free",
+            "tag": "100 queries/day free tier, native Chinese content.",
+            "env_vars": [
+                {
+                    "key": "BAIDU_API_KEY",
+                    "prompt": "Baidu Search API key",
+                    "url": "https://ai.baidu.com",
+                },
+            ],
+        }

plugins/web/bocha/__init__.py

@@ -0,0 +1 @@
+"""Bocha (博查) web search provider plugin."""

plugins/web/bocha/plugin.yaml

@@ -0,0 +1,7 @@
+name: web-bocha
+version: 1.0.0
+description: "Bochaa AI Search API — leading Chinese search for AI. Free for personal use. Requires BOCHA_API_KEY from https://open.bochaai.com."
+author: Hermes Community
+kind: backend
+provides_web_providers:
+  - bocha

plugins/web/bocha/provider.py

@@ -0,0 +1,65 @@
+"""Bochaa AI Search API — leading Chinese search for AI. Free for personal use. Requires BOCHA_API_KEY from https://open.bochaai.com."""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import Any, Dict
+
+from agent.web_search_provider import WebSearchProvider
+
+logger = logging.getLogger(__name__)
+
+
+class BochaWebSearchProvider(WebSearchProvider):
+    """Bocha (博查) — free tier search provider."""
+
+    @property
+    def name(self) -> str:
+        return "bocha"
+
+    @property
+    def display_name(self) -> str:
+        return "Bocha (博查)"
+
+    def is_available(self) -> bool:
+        """Return True when BOCHA_API_KEY is set."""
+        return bool(os.getenv("BOCHA_API_KEY", "").strip())
+
+    def supports_search(self) -> bool:
+        return True
+
+    def supports_extract(self) -> bool:
+        return False
+
+    def search(self, query: str, limit: int = 5) -> Dict[str, Any]:
+        import requests
+        api_key = os.getenv("BOCHA_API_KEY", "").strip()
+        if not api_key:
+            return {"success": False, "error": "BOCHA_API_KEY not set. Get a key at https://open.bochaai.com."}
+        try:
+            r = requests.post("https://api.bochaai.com/v1/web-search", json={"query": query, "count": min(limit, 20), "summary": True}, headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}, timeout=10)
+            r.raise_for_status()
+            data = r.json()
+            raw = list(((data or {}).get("webPages", []) or {}).get("value", []))[:limit]
+            results = [
+                {"title": str(i.get("name", "")), "url": str(i.get("url", "")),
+                 "description": str(i.get("summary", i.get("snippet", ""))), "position": n + 1}
+                for n, i in enumerate(raw)
+            ]
+            return {"success": True, "data": {"web": results}}
+        except Exception as e:
+            return {"success": False, "error": str(e)}
+    def get_setup_schema(self) -> Dict[str, Any]:
+        return {
+            "name": "Bocha (博查)",
+            "badge": "free",
+            "tag": "Free for personal use, best Chinese content quality. 1,000 free queries starter pack.",
+            "env_vars": [
+                {
+                    "key": "BOCHA_API_KEY",
+                    "prompt": "Bocha (博查) API key",
+                    "url": "https://open.bochaai.com",
+                },
+            ],
+        }

plugins/web/google-cse/__init__.py

@@ -0,0 +1 @@
+"""Google CSE web search provider plugin."""

plugins/web/google-cse/plugin.yaml

@@ -0,0 +1,7 @@
+name: web-google-cse
+version: 1.0.0
+description: "Google Custom Search JSON API — requires GCP project + API key + SE ID. NOTE: blocked by GFW in China."
+author: Hermes Community
+kind: backend
+provides_web_providers:
+  - google-cse

plugins/web/google-cse/provider.py

@@ -0,0 +1,67 @@
+"""Google Custom Search JSON API — requires GCP project + API key + SE ID. NOTE: blocked by GFW in China."""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import Any, Dict
+
+from agent.web_search_provider import WebSearchProvider
+
+logger = logging.getLogger(__name__)
+
+
+class GoogleCseWebSearchProvider(WebSearchProvider):
+    """Google CSE — free tier search provider."""
+
+    @property
+    def name(self) -> str:
+        return "google-cse"
+
+    @property
+    def display_name(self) -> str:
+        return "Google CSE"
+
+    def is_available(self) -> bool:
+        """Return True when GOOGLE_CSE_API_KEY is set."""
+        return bool(os.getenv("GOOGLE_CSE_API_KEY", "").strip())
+
+    def supports_search(self) -> bool:
+        return True
+
+    def supports_extract(self) -> bool:
+        return False
+
+    def search(self, query: str, limit: int = 5) -> Dict[str, Any]:
+        import requests
+        api_key = os.getenv("GOOGLE_CSE_API_KEY", "").strip()
+        cx = os.getenv("GOOGLE_CSE_CX", "").strip()
+        if not api_key or not cx:
+            return {"success": False, "error": "GOOGLE_CSE_API_KEY or GOOGLE_CSE_CX not set."}
+        try:
+            r = requests.get("https://customsearch.googleapis.com/customsearch/v1",
+                params={"q": query, "num": min(limit, 10), "key": api_key, "cx": cx}, timeout=10)
+            r.raise_for_status()
+            data = r.json()
+            raw = data.get("items", [])[:limit]
+            results = [
+                {"title": str(i.get("title", "")), "url": str(i.get("link", "")),
+                 "description": str(i.get("snippet", "")), "position": n + 1}
+                for n, i in enumerate(raw)
+            ]
+            return {"success": True, "data": {"web": results}}
+        except Exception as e:
+            return {"success": False, "error": str(e)}
+    def get_setup_schema(self) -> Dict[str, Any]:
+        return {
+            "name": "Google CSE",
+            "badge": "free",
+            "tag": "100 queries/day free tier. Requires GCP setup. NOT accessible from mainland China without proxy.",
+            "env_vars": [
+                {
+                    "key": "GOOGLE_CSE_API_KEY",
+                    "prompt": "Google CSE API key",
+                    "url": "https://developers.google.com/custom-search/v1/overview",
+                },
+            ],
+        }

plugins/web/jina/__init__.py

@@ -0,0 +1 @@
+"""Jina AI web search provider plugin."""

plugins/web/jina/plugin.yaml

@@ -0,0 +1,7 @@
+name: web-jina
+version: 1.0.0
+description: "Jina AI Search — full-page content extraction. 10M tokens free for new users. Requires JINA_API_KEY from https://jina.ai. NOTE: blocked by GFW in China."
+author: Hermes Community
+kind: backend
+provides_web_providers:
+  - jina

plugins/web/jina/provider.py

@@ -0,0 +1,68 @@
+"""Jina AI Search — full-page content extraction. 10M tokens free for new users. Requires JINA_API_KEY from https://jina.ai. NOTE: blocked by GFW in China."""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import Any, Dict
+
+from agent.web_search_provider import WebSearchProvider
+
+logger = logging.getLogger(__name__)
+
+
+class JinaWebSearchProvider(WebSearchProvider):
+    """Jina AI — free tier search provider."""
+
+    @property
+    def name(self) -> str:
+        return "jina"
+
+    @property
+    def display_name(self) -> str:
+        return "Jina AI"
+
+    def is_available(self) -> bool:
+        """Return True when JINA_API_KEY is set."""
+        return bool(os.getenv("JINA_API_KEY", "").strip())
+
+    def supports_search(self) -> bool:
+        return True
+
+    def supports_extract(self) -> bool:
+        return False
+
+    def search(self, query: str, limit: int = 5) -> Dict[str, Any]:
+        import requests
+        api_key = os.getenv("JINA_API_KEY", "").strip()
+        if not api_key:
+            return {"success": False, "error": "JINA_API_KEY not set. Get a key at https://jina.ai (10M free tokens for new users)."}
+        try:
+            import urllib.parse
+            r = requests.get(f"https://s.jina.ai/{urllib.parse.quote(query)}",
+                headers={"Authorization": f"Bearer {api_key}", "Accept": "application/json"},
+                timeout=15)
+            r.raise_for_status()
+            data = r.json()
+            raw = data.get("data", [])[:limit]
+            results = [
+                {"title": str(i.get("title", "")), "url": str(i.get("url", "")),
+                 "description": str(i.get("description", i.get("content", "")))[:500], "position": n + 1}
+                for n, i in enumerate(raw)
+            ]
+            return {"success": True, "data": {"web": results}}
+        except Exception as e:
+            return {"success": False, "error": str(e)}
+    def get_setup_schema(self) -> Dict[str, Any]:
+        return {
+            "name": "Jina AI",
+            "badge": "free",
+            "tag": "10M tokens free for new users. NOT accessible from mainland China without proxy.",
+            "env_vars": [
+                {
+                    "key": "JINA_API_KEY",
+                    "prompt": "Jina AI API key",
+                    "url": "https://jina.ai",
+                },
+            ],
+        }

plugins/web/qiniu-baidu/__init__.py

@@ -0,0 +1 @@
+"""Qiniu Baidu Search web search provider plugin."""

plugins/web/qiniu-baidu/plugin.yaml

@@ -0,0 +1,7 @@
+name: web-qiniu-baidu
+version: 1.0.0
+description: "Qiniu Cloud Baidu Search — OpenAI-compatible Baidu Search API. Requires QINIU_API_KEY from https://qiniu.com/ai/models."
+author: Hermes Community
+kind: backend
+provides_web_providers:
+  - qiniu-baidu