mirror of
https://github.com/MODSetter/SurfSense.git
synced 2026-07-10 22:32:16 +02:00
feat(agents): consolidate connectors under mcp_discovery; route web search through google_search
MCP consolidation: - Route all MCP-capable connectors (Slack, Jira, Linear, ClickUp, Airtable, Notion, Confluence, interim Gmail/Calendar, custom MCP) through a single `mcp_discovery` subagent. Drive/OneDrive/Dropbox stay native to enrich the KB. - Deprecate Discord/Teams/Luma: no viable official MCP server. Google-only web search: - Remove the main-agent `web_search` tool and the SearXNG platform service; all public web search now flows through the `google_search` subagent via task(). - Deprecate the Tavily/SearXNG/Linkup/Baidu search connectors (HTTP 410 on create, "Deprecated" badge); guide heavy users to the custom MCP connector. - Remove web search from anonymous chat (pure Q&A). - Tear SearXNG out of docker compose + install scripts; drop tavily-python and linkup-sdk deps and their config/env vars. Fix: - metrics._package_version() now swallows any metadata lookup failure. A malformed editable-install distribution with no `Version` field raised KeyError deep in importlib.metadata, and since it runs on every record_subagent_invoke_duration call it was crashing every task() delegation. Verified end-to-end against live GPT-5.4. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
DESKTOP-RTLN3BA\$punk
parent
ff2e5f390f
commit
ab747e7a49
206 changed files with 1704 additions and 7223 deletions
|
|
@ -4,12 +4,9 @@ from datetime import datetime
|
|||
from threading import Lock
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
from linkup import LinkupClient
|
||||
from sqlalchemy import func
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy.future import select
|
||||
from tavily import TavilyClient
|
||||
|
||||
from app.config import config
|
||||
from app.db import (
|
||||
|
|
@ -388,363 +385,6 @@ class ConnectorService:
|
|||
result = await self.session.execute(query)
|
||||
return result.scalars().first()
|
||||
|
||||
async def search_tavily(
|
||||
self, user_query: str, workspace_id: int, top_k: int = 20
|
||||
) -> tuple:
|
||||
"""
|
||||
Search using Tavily API and return both the source information and documents
|
||||
|
||||
Args:
|
||||
user_query: The user's query
|
||||
workspace_id: The workspace ID
|
||||
top_k: Maximum number of results to return
|
||||
|
||||
Returns:
|
||||
tuple: (sources_info, documents)
|
||||
"""
|
||||
# Get Tavily connector configuration
|
||||
tavily_connector = await self.get_connector_by_type(
|
||||
SearchSourceConnectorType.TAVILY_API, workspace_id
|
||||
)
|
||||
|
||||
if not tavily_connector:
|
||||
# Return empty results if no Tavily connector is configured
|
||||
return {
|
||||
"id": 3,
|
||||
"name": "Tavily Search",
|
||||
"type": "TAVILY_API",
|
||||
"sources": [],
|
||||
}, []
|
||||
|
||||
# Initialize Tavily client with API key from connector config
|
||||
tavily_api_key = tavily_connector.config.get("TAVILY_API_KEY")
|
||||
tavily_client = TavilyClient(api_key=tavily_api_key)
|
||||
|
||||
# Perform search with Tavily
|
||||
try:
|
||||
response = tavily_client.search(
|
||||
query=user_query,
|
||||
max_results=top_k,
|
||||
search_depth="advanced", # Use advanced search for better results
|
||||
)
|
||||
|
||||
# Extract results from Tavily response
|
||||
tavily_results = response.get("results", [])
|
||||
|
||||
# Early return if no results
|
||||
if not tavily_results:
|
||||
return {
|
||||
"id": 3,
|
||||
"name": "Tavily Search",
|
||||
"type": "TAVILY_API",
|
||||
"sources": [],
|
||||
}, []
|
||||
|
||||
# Process each result and create sources directly without deduplication
|
||||
sources_list = []
|
||||
documents = []
|
||||
|
||||
async with self.counter_lock:
|
||||
for _i, result in enumerate(tavily_results):
|
||||
# Create a source entry
|
||||
source = {
|
||||
"id": self.source_id_counter,
|
||||
"title": result.get("title", "Tavily Result"),
|
||||
"description": result.get("content", ""),
|
||||
"url": result.get("url", ""),
|
||||
}
|
||||
sources_list.append(source)
|
||||
|
||||
# Create a document entry
|
||||
document = {
|
||||
"chunk_id": self.source_id_counter,
|
||||
"content": result.get("content", ""),
|
||||
"score": result.get("score", 0.0),
|
||||
"document": {
|
||||
"id": self.source_id_counter,
|
||||
"title": result.get("title", "Tavily Result"),
|
||||
"document_type": "TAVILY_API",
|
||||
"metadata": {
|
||||
"url": result.get("url", ""),
|
||||
"published_date": result.get("published_date", ""),
|
||||
"source": "TAVILY_API",
|
||||
},
|
||||
},
|
||||
}
|
||||
documents.append(document)
|
||||
self.source_id_counter += 1
|
||||
|
||||
# Create result object
|
||||
result_object = {
|
||||
"id": 3,
|
||||
"name": "Tavily Search",
|
||||
"type": "TAVILY_API",
|
||||
"sources": sources_list,
|
||||
}
|
||||
|
||||
return result_object, documents
|
||||
|
||||
except Exception as e:
|
||||
# Log the error and return empty results
|
||||
print(f"Error searching with Tavily: {e!s}")
|
||||
return {
|
||||
"id": 3,
|
||||
"name": "Tavily Search",
|
||||
"type": "TAVILY_API",
|
||||
"sources": [],
|
||||
}, []
|
||||
|
||||
async def search_searxng(
|
||||
self,
|
||||
user_query: str,
|
||||
workspace_id: int,
|
||||
top_k: int = 20,
|
||||
) -> tuple:
|
||||
"""Search using the platform SearXNG instance.
|
||||
|
||||
Delegates to ``WebSearchService`` which handles caching, circuit
|
||||
breaking, and retries. SearXNG configuration comes from the
|
||||
docker/searxng/settings.yml file.
|
||||
"""
|
||||
from app.services import web_search_service
|
||||
|
||||
if not web_search_service.is_available():
|
||||
return {
|
||||
"id": 11,
|
||||
"name": "Web Search",
|
||||
"type": "SEARXNG_API",
|
||||
"sources": [],
|
||||
}, []
|
||||
|
||||
return await web_search_service.search(
|
||||
query=user_query,
|
||||
top_k=top_k,
|
||||
)
|
||||
|
||||
async def search_baidu(
|
||||
self,
|
||||
user_query: str,
|
||||
workspace_id: int,
|
||||
top_k: int = 20,
|
||||
) -> tuple:
|
||||
"""
|
||||
Search using Baidu AI Search API and return both sources and documents.
|
||||
|
||||
Baidu AI Search provides intelligent search with automatic summarization.
|
||||
We extract the raw search results (references) from the API response.
|
||||
|
||||
Args:
|
||||
user_query: User's search query
|
||||
workspace_id: Workspace ID
|
||||
top_k: Maximum number of results to return
|
||||
|
||||
Returns:
|
||||
tuple: (sources_info_dict, documents_list)
|
||||
"""
|
||||
# Get Baidu connector configuration
|
||||
baidu_connector = await self.get_connector_by_type(
|
||||
SearchSourceConnectorType.BAIDU_SEARCH_API, workspace_id
|
||||
)
|
||||
|
||||
if not baidu_connector:
|
||||
return {
|
||||
"id": 12,
|
||||
"name": "Baidu Search",
|
||||
"type": "BAIDU_SEARCH_API",
|
||||
"sources": [],
|
||||
}, []
|
||||
|
||||
config = baidu_connector.config or {}
|
||||
api_key = config.get("BAIDU_API_KEY")
|
||||
|
||||
if not api_key:
|
||||
print("ERROR: Baidu connector is missing BAIDU_API_KEY configuration")
|
||||
print(f"Connector config: {config}")
|
||||
return {
|
||||
"id": 12,
|
||||
"name": "Baidu Search",
|
||||
"type": "BAIDU_SEARCH_API",
|
||||
"sources": [],
|
||||
}, []
|
||||
|
||||
# Optional configuration parameters
|
||||
model = config.get("BAIDU_MODEL", "ernie-3.5-8k")
|
||||
search_source = config.get("BAIDU_SEARCH_SOURCE", "baidu_search_v2")
|
||||
enable_deep_search = config.get("BAIDU_ENABLE_DEEP_SEARCH", False)
|
||||
|
||||
# Baidu AI Search API endpoint
|
||||
baidu_endpoint = "https://qianfan.baidubce.com/v2/ai_search/chat/completions"
|
||||
|
||||
# Prepare request headers
|
||||
# Note: Baidu uses X-Appbuilder-Authorization instead of standard Authorization header
|
||||
headers = {
|
||||
"X-Appbuilder-Authorization": f"Bearer {api_key}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
# Prepare request payload
|
||||
# Calculate resource_type_filter top_k values
|
||||
# Baidu v2 supports max 20 per type
|
||||
max_per_type = min(top_k, 20)
|
||||
|
||||
payload = {
|
||||
"messages": [{"role": "user", "content": user_query}],
|
||||
"model": model,
|
||||
"search_source": search_source,
|
||||
"resource_type_filter": [
|
||||
{"type": "web", "top_k": max_per_type},
|
||||
{"type": "video", "top_k": max(1, max_per_type // 4)}, # Fewer videos
|
||||
],
|
||||
"stream": False, # Non-streaming for simpler processing
|
||||
"enable_deep_search": enable_deep_search,
|
||||
"enable_corner_markers": True, # Enable reference markers
|
||||
}
|
||||
|
||||
try:
|
||||
# Baidu AI Search may take longer as it performs search + summarization
|
||||
# Increase timeout to 90 seconds
|
||||
async with httpx.AsyncClient(timeout=90.0) as client:
|
||||
response = await client.post(
|
||||
baidu_endpoint,
|
||||
headers=headers,
|
||||
json=payload,
|
||||
)
|
||||
response.raise_for_status()
|
||||
except httpx.TimeoutException as exc:
|
||||
print(f"ERROR: Baidu API request timeout after 90s: {exc!r}")
|
||||
print(f"Endpoint: {baidu_endpoint}")
|
||||
return {
|
||||
"id": 12,
|
||||
"name": "Baidu Search",
|
||||
"type": "BAIDU_SEARCH_API",
|
||||
"sources": [],
|
||||
}, []
|
||||
except httpx.HTTPStatusError as exc:
|
||||
print(f"ERROR: Baidu API HTTP Status Error: {exc.response.status_code}")
|
||||
print(f"Response text: {exc.response.text[:500]}")
|
||||
print(f"Request URL: {exc.request.url}")
|
||||
return {
|
||||
"id": 12,
|
||||
"name": "Baidu Search",
|
||||
"type": "BAIDU_SEARCH_API",
|
||||
"sources": [],
|
||||
}, []
|
||||
except httpx.RequestError as exc:
|
||||
print(f"ERROR: Baidu API Request Error: {type(exc).__name__}: {exc!r}")
|
||||
print(f"Endpoint: {baidu_endpoint}")
|
||||
return {
|
||||
"id": 12,
|
||||
"name": "Baidu Search",
|
||||
"type": "BAIDU_SEARCH_API",
|
||||
"sources": [],
|
||||
}, []
|
||||
except Exception as exc:
|
||||
print(
|
||||
f"ERROR: Unexpected error calling Baidu API: {type(exc).__name__}: {exc!r}"
|
||||
)
|
||||
print(f"Endpoint: {baidu_endpoint}")
|
||||
print(f"Payload: {payload}")
|
||||
return {
|
||||
"id": 12,
|
||||
"name": "Baidu Search",
|
||||
"type": "BAIDU_SEARCH_API",
|
||||
"sources": [],
|
||||
}, []
|
||||
|
||||
try:
|
||||
data = response.json()
|
||||
except ValueError as e:
|
||||
print(f"ERROR: Failed to decode JSON response from Baidu AI Search: {e}")
|
||||
print(f"Response status: {response.status_code}")
|
||||
print(f"Response text: {response.text[:500]}") # First 500 chars
|
||||
return {
|
||||
"id": 12,
|
||||
"name": "Baidu Search",
|
||||
"type": "BAIDU_SEARCH_API",
|
||||
"sources": [],
|
||||
}, []
|
||||
|
||||
# Extract references (search results) from the response
|
||||
baidu_references = data.get("references", [])
|
||||
|
||||
if "code" in data or "message" in data:
|
||||
print(
|
||||
f"WARNING: Baidu API returned error - Code: {data.get('code')}, Message: {data.get('message')}"
|
||||
)
|
||||
|
||||
if not baidu_references:
|
||||
print("WARNING: No references found in Baidu API response")
|
||||
print(f"Response keys: {list(data.keys())}")
|
||||
return {
|
||||
"id": 12,
|
||||
"name": "Baidu Search",
|
||||
"type": "BAIDU_SEARCH_API",
|
||||
"sources": [],
|
||||
}, []
|
||||
|
||||
sources_list: list[dict[str, Any]] = []
|
||||
documents: list[dict[str, Any]] = []
|
||||
|
||||
async with self.counter_lock:
|
||||
for reference in baidu_references:
|
||||
# Extract basic fields
|
||||
title = reference.get("title", "Baidu Search Result")
|
||||
url = reference.get("url", "")
|
||||
content = reference.get("content", "")
|
||||
date = reference.get("date", "")
|
||||
ref_type = reference.get("type", "web") # web, image, video
|
||||
|
||||
# Create a source entry
|
||||
source = {
|
||||
"id": self.source_id_counter,
|
||||
"title": title,
|
||||
"description": content[:300]
|
||||
if content
|
||||
else "", # Limit description length
|
||||
"url": url,
|
||||
}
|
||||
sources_list.append(source)
|
||||
|
||||
# Prepare metadata
|
||||
metadata = {
|
||||
"url": url,
|
||||
"date": date,
|
||||
"type": ref_type,
|
||||
"source": "BAIDU_SEARCH_API",
|
||||
"web_anchor": reference.get("web_anchor", ""),
|
||||
"website": reference.get("website", ""),
|
||||
}
|
||||
|
||||
# Add type-specific metadata
|
||||
if ref_type == "image" and reference.get("image"):
|
||||
metadata["image"] = reference["image"]
|
||||
elif ref_type == "video" and reference.get("video"):
|
||||
metadata["video"] = reference["video"]
|
||||
|
||||
# Create a document entry
|
||||
document = {
|
||||
"chunk_id": self.source_id_counter,
|
||||
"content": content,
|
||||
"score": 1.0, # Baidu doesn't provide relevance scores
|
||||
"document": {
|
||||
"id": self.source_id_counter,
|
||||
"title": title,
|
||||
"document_type": "BAIDU_SEARCH_API",
|
||||
"metadata": metadata,
|
||||
},
|
||||
}
|
||||
documents.append(document)
|
||||
self.source_id_counter += 1
|
||||
|
||||
result_object = {
|
||||
"id": 12,
|
||||
"name": "Baidu Search",
|
||||
"type": "BAIDU_SEARCH_API",
|
||||
"sources": sources_list,
|
||||
}
|
||||
|
||||
return result_object, documents
|
||||
|
||||
async def search_slack(
|
||||
self,
|
||||
user_query: str,
|
||||
|
|
@ -1875,127 +1515,6 @@ class ConnectorService:
|
|||
|
||||
return result_object, clickup_docs
|
||||
|
||||
async def search_linkup(
|
||||
self,
|
||||
user_query: str,
|
||||
workspace_id: int,
|
||||
mode: str = "standard",
|
||||
) -> tuple:
|
||||
"""
|
||||
Search using Linkup API and return both the source information and documents
|
||||
|
||||
Args:
|
||||
user_query: The user's query
|
||||
workspace_id: The workspace ID
|
||||
mode: Search depth mode, can be "standard" or "deep"
|
||||
|
||||
Returns:
|
||||
tuple: (sources_info, documents)
|
||||
"""
|
||||
# Get Linkup connector configuration
|
||||
linkup_connector = await self.get_connector_by_type(
|
||||
SearchSourceConnectorType.LINKUP_API, workspace_id
|
||||
)
|
||||
|
||||
if not linkup_connector:
|
||||
# Return empty results if no Linkup connector is configured
|
||||
return {
|
||||
"id": 10,
|
||||
"name": "Linkup Search",
|
||||
"type": "LINKUP_API",
|
||||
"sources": [],
|
||||
}, []
|
||||
|
||||
# Initialize Linkup client with API key from connector config
|
||||
linkup_api_key = linkup_connector.config.get("LINKUP_API_KEY")
|
||||
linkup_client = LinkupClient(api_key=linkup_api_key)
|
||||
|
||||
# Perform search with Linkup
|
||||
try:
|
||||
response = linkup_client.search(
|
||||
query=user_query,
|
||||
depth=mode, # Use the provided mode ("standard" or "deep")
|
||||
output_type="searchResults", # Default to search results
|
||||
)
|
||||
|
||||
# Extract results from Linkup response - access as attribute instead of using .get()
|
||||
linkup_results = response.results if hasattr(response, "results") else []
|
||||
|
||||
# Only proceed if we have results
|
||||
if not linkup_results:
|
||||
return {
|
||||
"id": 10,
|
||||
"name": "Linkup Search",
|
||||
"type": "LINKUP_API",
|
||||
"sources": [],
|
||||
}, []
|
||||
|
||||
# Process each result and create sources directly without deduplication
|
||||
sources_list = []
|
||||
documents = []
|
||||
|
||||
async with self.counter_lock:
|
||||
for _i, result in enumerate(linkup_results):
|
||||
# Only process results that have content
|
||||
if not hasattr(result, "content") or not result.content:
|
||||
continue
|
||||
|
||||
# Create a source entry
|
||||
source = {
|
||||
"id": self.source_id_counter,
|
||||
"title": (
|
||||
result.name if hasattr(result, "name") else "Linkup Result"
|
||||
),
|
||||
"description": (
|
||||
result.content if hasattr(result, "content") else ""
|
||||
),
|
||||
"url": result.url if hasattr(result, "url") else "",
|
||||
}
|
||||
sources_list.append(source)
|
||||
|
||||
# Create a document entry
|
||||
document = {
|
||||
"chunk_id": self.source_id_counter,
|
||||
"content": result.content if hasattr(result, "content") else "",
|
||||
"score": 1.0, # Default score since not provided by Linkup
|
||||
"document": {
|
||||
"id": self.source_id_counter,
|
||||
"title": (
|
||||
result.name
|
||||
if hasattr(result, "name")
|
||||
else "Linkup Result"
|
||||
),
|
||||
"document_type": "LINKUP_API",
|
||||
"metadata": {
|
||||
"url": result.url if hasattr(result, "url") else "",
|
||||
"type": result.type if hasattr(result, "type") else "",
|
||||
"source": "LINKUP_API",
|
||||
},
|
||||
},
|
||||
}
|
||||
documents.append(document)
|
||||
self.source_id_counter += 1
|
||||
|
||||
# Create result object
|
||||
result_object = {
|
||||
"id": 10,
|
||||
"name": "Linkup Search",
|
||||
"type": "LINKUP_API",
|
||||
"sources": sources_list,
|
||||
}
|
||||
|
||||
return result_object, documents
|
||||
|
||||
except Exception as e:
|
||||
# Log the error and return empty results
|
||||
print(f"Error searching with Linkup: {e!s}")
|
||||
return {
|
||||
"id": 10,
|
||||
"name": "Linkup Search",
|
||||
"type": "LINKUP_API",
|
||||
"sources": [],
|
||||
}, []
|
||||
|
||||
async def search_discord(
|
||||
self,
|
||||
user_query: str,
|
||||
|
|
|
|||
|
|
@ -185,6 +185,46 @@ MCP_SERVICES: dict[str, MCPServiceConfig] = {
|
|||
),
|
||||
account_metadata_keys=["user_id", "user_email"],
|
||||
),
|
||||
"notion": MCPServiceConfig(
|
||||
name="Notion",
|
||||
mcp_url="https://mcp.notion.com/mcp",
|
||||
connector_type="NOTION_CONNECTOR",
|
||||
# DCR (RFC 7591): Notion issues its own client credentials. It expires
|
||||
# DCR registrations, but refresh reuses the original persisted
|
||||
# ``mcp_oauth.client_id`` (see _refresh_connector_token).
|
||||
allowed_tools=[
|
||||
"search",
|
||||
"fetch",
|
||||
"create-pages",
|
||||
"update-page",
|
||||
],
|
||||
readonly_tools=frozenset({"search", "fetch"}),
|
||||
account_metadata_keys=["workspace_name"],
|
||||
),
|
||||
"confluence": MCPServiceConfig(
|
||||
name="Confluence",
|
||||
# Same Atlassian Rovo server as Jira; tool sets are kept disjoint by
|
||||
# curation so a workspace can connect both as separate connectors.
|
||||
mcp_url="https://mcp.atlassian.com/v1/mcp",
|
||||
connector_type="CONFLUENCE_CONNECTOR",
|
||||
allowed_tools=[
|
||||
"getAccessibleAtlassianResources",
|
||||
"getConfluenceSpaces",
|
||||
"getConfluencePage",
|
||||
"searchConfluenceUsingCql",
|
||||
"createConfluencePage",
|
||||
"updateConfluencePage",
|
||||
],
|
||||
readonly_tools=frozenset(
|
||||
{
|
||||
"getAccessibleAtlassianResources",
|
||||
"getConfluenceSpaces",
|
||||
"getConfluencePage",
|
||||
"searchConfluenceUsingCql",
|
||||
}
|
||||
),
|
||||
account_metadata_keys=["cloud_id", "site_name", "base_url"],
|
||||
),
|
||||
}
|
||||
|
||||
_CONNECTOR_TYPE_TO_SERVICE: dict[str, MCPServiceConfig] = {
|
||||
|
|
@ -205,6 +245,25 @@ LIVE_CONNECTOR_TYPES: frozenset[SearchSourceConnectorType] = frozenset(
|
|||
SearchSourceConnectorType.COMPOSIO_GMAIL_CONNECTOR,
|
||||
SearchSourceConnectorType.DISCORD_CONNECTOR,
|
||||
SearchSourceConnectorType.LUMA_CONNECTOR,
|
||||
# Migrated to hosted MCP — indexing pipelines deprecated (KB is
|
||||
# files/notes/uploads only). LIVE membership blocks the index route
|
||||
# and auto-disables periodic indexing.
|
||||
SearchSourceConnectorType.NOTION_CONNECTOR,
|
||||
SearchSourceConnectorType.CONFLUENCE_CONNECTOR,
|
||||
}
|
||||
)
|
||||
|
||||
# Indexing-only connectors retired with the KB "files, notes, and uploads only"
|
||||
# shift: their ingestion pipelines are deprecated. Like LIVE membership, this
|
||||
# blocks the index route and auto-disables periodic indexing — but the message
|
||||
# frames it as a deprecation, not a real-time-tools swap. Obsidian is
|
||||
# intentionally excluded (file-like vault content still enriches the KB).
|
||||
DEPRECATED_INDEXING_CONNECTOR_TYPES: frozenset[SearchSourceConnectorType] = frozenset(
|
||||
{
|
||||
SearchSourceConnectorType.GITHUB_CONNECTOR,
|
||||
SearchSourceConnectorType.BOOKSTACK_CONNECTOR,
|
||||
SearchSourceConnectorType.ELASTICSEARCH_CONNECTOR,
|
||||
SearchSourceConnectorType.CIRCLEBACK_CONNECTOR,
|
||||
}
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -1,290 +0,0 @@
|
|||
"""
|
||||
Platform-level web search service backed by SearXNG.
|
||||
|
||||
Redis is used only for result caching (graceful degradation if unavailable).
|
||||
The circuit breaker is fully in-process — no external dependency, zero
|
||||
latency overhead.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import threading
|
||||
import time
|
||||
from typing import Any
|
||||
from urllib.parse import urljoin
|
||||
|
||||
import httpx
|
||||
import redis
|
||||
|
||||
from app.config import config
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_EMPTY_RESULT: dict[str, Any] = {
|
||||
"id": 11,
|
||||
"name": "Web Search",
|
||||
"type": "SEARXNG_API",
|
||||
"sources": [],
|
||||
}
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Redis — used only for result caching
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_redis_client: redis.Redis | None = None
|
||||
|
||||
|
||||
def _get_redis() -> redis.Redis:
|
||||
global _redis_client
|
||||
if _redis_client is None:
|
||||
_redis_client = redis.from_url(config.REDIS_APP_URL, decode_responses=True)
|
||||
return _redis_client
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# In-process Circuit Breaker (no Redis dependency)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_CB_FAILURE_THRESHOLD = 5
|
||||
_CB_FAILURE_WINDOW_SECONDS = 60
|
||||
_CB_COOLDOWN_SECONDS = 30
|
||||
|
||||
_cb_lock = threading.Lock()
|
||||
_cb_failure_count: int = 0
|
||||
_cb_last_failure_time: float = 0.0
|
||||
_cb_open_until: float = 0.0
|
||||
|
||||
|
||||
def _circuit_is_open() -> bool:
|
||||
return time.monotonic() < _cb_open_until
|
||||
|
||||
|
||||
def _record_failure() -> None:
|
||||
global _cb_failure_count, _cb_last_failure_time, _cb_open_until
|
||||
now = time.monotonic()
|
||||
with _cb_lock:
|
||||
if now - _cb_last_failure_time > _CB_FAILURE_WINDOW_SECONDS:
|
||||
_cb_failure_count = 0
|
||||
_cb_failure_count += 1
|
||||
_cb_last_failure_time = now
|
||||
if _cb_failure_count >= _CB_FAILURE_THRESHOLD:
|
||||
_cb_open_until = now + _CB_COOLDOWN_SECONDS
|
||||
logger.warning(
|
||||
"Circuit breaker OPENED after %d failures — "
|
||||
"SearXNG calls paused for %ds",
|
||||
_cb_failure_count,
|
||||
_CB_COOLDOWN_SECONDS,
|
||||
)
|
||||
|
||||
|
||||
def _record_success() -> None:
|
||||
global _cb_failure_count, _cb_open_until
|
||||
with _cb_lock:
|
||||
_cb_failure_count = 0
|
||||
_cb_open_until = 0.0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Result Caching (Redis, graceful degradation)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_CACHE_TTL_SECONDS = 300 # 5 minutes
|
||||
_CACHE_PREFIX = "websearch:cache:"
|
||||
|
||||
|
||||
def _cache_key(query: str, engines: str | None, language: str | None) -> str:
|
||||
raw = f"{query}|{engines or ''}|{language or ''}"
|
||||
digest = hashlib.sha256(raw.encode()).hexdigest()[:24]
|
||||
return f"{_CACHE_PREFIX}{digest}"
|
||||
|
||||
|
||||
def _cache_get(key: str) -> dict | None:
|
||||
try:
|
||||
data = _get_redis().get(key)
|
||||
if data:
|
||||
return json.loads(data)
|
||||
except (redis.RedisError, json.JSONDecodeError):
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
def _cache_set(key: str, value: dict) -> None:
|
||||
with contextlib.suppress(redis.RedisError):
|
||||
_get_redis().setex(key, _CACHE_TTL_SECONDS, json.dumps(value))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Public API
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def is_available() -> bool:
|
||||
"""Return ``True`` when the platform SearXNG host is configured."""
|
||||
return bool(config.SEARXNG_DEFAULT_HOST)
|
||||
|
||||
|
||||
async def health_check() -> dict[str, Any]:
|
||||
"""Ping the SearXNG ``/healthz`` endpoint and return status info."""
|
||||
host = config.SEARXNG_DEFAULT_HOST
|
||||
if not host:
|
||||
return {"status": "unavailable", "error": "SEARXNG_DEFAULT_HOST not set"}
|
||||
|
||||
healthz_url = urljoin(host if host.endswith("/") else f"{host}/", "healthz")
|
||||
t0 = time.perf_counter()
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=5.0, verify=False) as client:
|
||||
resp = await client.get(healthz_url)
|
||||
resp.raise_for_status()
|
||||
elapsed_ms = round((time.perf_counter() - t0) * 1000)
|
||||
return {
|
||||
"status": "healthy",
|
||||
"response_time_ms": elapsed_ms,
|
||||
"circuit_breaker": "open" if _circuit_is_open() else "closed",
|
||||
}
|
||||
except Exception as exc:
|
||||
elapsed_ms = round((time.perf_counter() - t0) * 1000)
|
||||
return {
|
||||
"status": "unhealthy",
|
||||
"error": str(exc),
|
||||
"response_time_ms": elapsed_ms,
|
||||
"circuit_breaker": "open" if _circuit_is_open() else "closed",
|
||||
}
|
||||
|
||||
|
||||
async def search(
|
||||
query: str,
|
||||
top_k: int = 20,
|
||||
*,
|
||||
engines: str | None = None,
|
||||
language: str | None = None,
|
||||
safesearch: int | None = None,
|
||||
) -> tuple[dict[str, Any], list[dict[str, Any]]]:
|
||||
"""Execute a web search against the platform SearXNG instance.
|
||||
|
||||
Returns the standard ``(result_object, documents)`` tuple expected by
|
||||
``ConnectorService.search_searxng``.
|
||||
"""
|
||||
host = config.SEARXNG_DEFAULT_HOST
|
||||
if not host:
|
||||
return dict(_EMPTY_RESULT), []
|
||||
|
||||
if _circuit_is_open():
|
||||
logger.info("Web search skipped — circuit breaker is open")
|
||||
result = dict(_EMPTY_RESULT)
|
||||
result["error"] = "Web search temporarily unavailable (circuit open)"
|
||||
result["status"] = "degraded"
|
||||
return result, []
|
||||
|
||||
ck = _cache_key(query, engines, language)
|
||||
cached = _cache_get(ck)
|
||||
if cached is not None:
|
||||
logger.debug("Web search cache HIT for query=%r", query[:60])
|
||||
return cached["result"], cached["documents"]
|
||||
|
||||
params: dict[str, Any] = {
|
||||
"q": query,
|
||||
"format": "json",
|
||||
"limit": max(1, min(top_k, 50)),
|
||||
}
|
||||
if engines:
|
||||
params["engines"] = engines
|
||||
if language:
|
||||
params["language"] = language
|
||||
if safesearch is not None and 0 <= safesearch <= 2:
|
||||
params["safesearch"] = safesearch
|
||||
|
||||
searx_endpoint = urljoin(host if host.endswith("/") else f"{host}/", "search")
|
||||
headers = {"Accept": "application/json"}
|
||||
|
||||
data: dict[str, Any] | None = None
|
||||
last_error: Exception | None = None
|
||||
|
||||
for attempt in range(2):
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=15.0, verify=False) as client:
|
||||
response = await client.get(
|
||||
searx_endpoint,
|
||||
params=params,
|
||||
headers=headers,
|
||||
)
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
break
|
||||
except (httpx.HTTPStatusError, httpx.TimeoutException) as exc:
|
||||
last_error = exc
|
||||
if attempt == 0 and (
|
||||
isinstance(exc, httpx.TimeoutException)
|
||||
or (
|
||||
isinstance(exc, httpx.HTTPStatusError)
|
||||
and exc.response.status_code >= 500
|
||||
)
|
||||
):
|
||||
continue
|
||||
break
|
||||
except httpx.HTTPError as exc:
|
||||
last_error = exc
|
||||
break
|
||||
except ValueError as exc:
|
||||
last_error = exc
|
||||
break
|
||||
|
||||
if data is None:
|
||||
_record_failure()
|
||||
logger.warning("Web search failed after retries: %s", last_error)
|
||||
return dict(_EMPTY_RESULT), []
|
||||
|
||||
_record_success()
|
||||
|
||||
searx_results = data.get("results", [])
|
||||
if not searx_results:
|
||||
return dict(_EMPTY_RESULT), []
|
||||
|
||||
sources_list: list[dict[str, Any]] = []
|
||||
documents: list[dict[str, Any]] = []
|
||||
|
||||
for idx, result in enumerate(searx_results):
|
||||
source_id = 200_000 + idx
|
||||
description = result.get("content") or result.get("snippet") or ""
|
||||
|
||||
sources_list.append(
|
||||
{
|
||||
"id": source_id,
|
||||
"title": result.get("title", "Web Search Result"),
|
||||
"description": description,
|
||||
"url": result.get("url", ""),
|
||||
}
|
||||
)
|
||||
|
||||
documents.append(
|
||||
{
|
||||
"chunk_id": source_id,
|
||||
"content": description or result.get("content", ""),
|
||||
"score": result.get("score", 0.0),
|
||||
"document": {
|
||||
"id": source_id,
|
||||
"title": result.get("title", "Web Search Result"),
|
||||
"document_type": "SEARXNG_API",
|
||||
"metadata": {
|
||||
"url": result.get("url", ""),
|
||||
"engines": result.get("engines", []),
|
||||
"category": result.get("category"),
|
||||
"source": "SEARXNG_API",
|
||||
},
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
result_object: dict[str, Any] = {
|
||||
"id": 11,
|
||||
"name": "Web Search",
|
||||
"type": "SEARXNG_API",
|
||||
"sources": sources_list,
|
||||
}
|
||||
|
||||
_cache_set(ck, {"result": result_object, "documents": documents})
|
||||
|
||||
return result_object, documents
|
||||
Loading…
Add table
Add a link
Reference in a new issue