deerflow-code/offline-backend-20260512/backend/packages/harness/deerflow/community/configurable_search/tools.py

"""Configurable JSON-over-HTTP implementation of the ``web_search`` tool."""

from __future__ import annotations

import json
import logging
from copy import deepcopy
from dataclasses import dataclass
from typing import Any
from urllib.parse import quote

import httpx
from langchain.tools import tool

from deerflow.config import get_app_config
from deerflow.config.tool_config import ToolConfig

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class ConfigurableSearchSettings:
    """Validated runtime settings read from the ``web_search`` tool entry."""

    enabled: bool
    endpoint: str
    payload: dict[str, Any]
    verify_ssl: bool
    timeout: float
    max_results: int
    result_url_template: str

    @classmethod
    def from_tool_config(cls, config: ToolConfig | None) -> ConfigurableSearchSettings:
        if config is None:
            raise ValueError("web_search is not configured")

        extra = config.model_extra or {}
        endpoint = str(extra.get("endpoint") or "").strip()
        if not endpoint.startswith(("http://", "https://")):
            raise ValueError("web_search.endpoint must use http:// or https://")

        raw_payload = extra.get("payload", {})
        if not isinstance(raw_payload, dict):
            raise ValueError("web_search.payload must be an object")

        try:
            timeout = float(extra.get("timeout", 30))
        except (TypeError, ValueError) as exc:
            raise ValueError("web_search.timeout must be a number") from exc
        if timeout <= 0:
            raise ValueError("web_search.timeout must be greater than zero")

        try:
            max_results = int(extra.get("max_results", 5))
        except (TypeError, ValueError) as exc:
            raise ValueError("web_search.max_results must be an integer") from exc
        if max_results <= 0:
            raise ValueError("web_search.max_results must be greater than zero")

        result_url_template = str(extra.get("result_url_template") or "").strip()
        if result_url_template and "{recUuid}" not in result_url_template:
            raise ValueError("web_search.result_url_template must contain {recUuid}")

        verify_ssl = extra.get("verify_ssl", True)
        if not isinstance(verify_ssl, bool):
            raise ValueError("web_search.verify_ssl must be true or false")

        return cls(
            enabled=config.enabled,
            endpoint=endpoint,
            payload=deepcopy(raw_payload),
            verify_ssl=verify_ssl,
            timeout=timeout,
            max_results=max_results,
            result_url_template=result_url_template,
        )


def _text(value: Any) -> str:
    """Normalize optional scalar result fields without leaking ``None``."""

    if value is None:
        return ""
    if isinstance(value, str):
        return value
    return str(value)


def _error_result(query: str, message: str) -> str:
    return json.dumps(
        {
            "query": query,
            "total_results": 0,
            "results": [],
            "error": message,
        },
        ensure_ascii=False,
    )


def _normalize_results(data: Any, settings: ConfigurableSearchSettings) -> list[dict[str, str]]:
    if not isinstance(data, dict):
        raise ValueError("search API response must be a JSON object")

    raw_results = data.get("results")
    if not isinstance(raw_results, list):
        raise ValueError("search API response field 'results' must be an array")

    normalized: list[dict[str, str]] = []
    for item in raw_results[: settings.max_results]:
        if not isinstance(item, dict):
            logger.warning("Ignoring non-object search result: %r", item)
            continue

        rec_uuid = _text(item.get("recUuid")).strip()
        original_url = _text(item.get("url")).strip()
        result_url = original_url
        if rec_uuid and settings.result_url_template:
            result_url = settings.result_url_template.replace(
                "{recUuid}",
                quote(rec_uuid, safe=""),
            )

        normalized.append(
            {
                "recUuid": rec_uuid,
                "content": _text(item.get("content")),
                "title": _text(item.get("title")),
                "url": result_url,
            }
        )
    return normalized


async def execute_search(
    query: str,
    settings: ConfigurableSearchSettings,
    *,
    transport: httpx.AsyncBaseTransport | None = None,
) -> str:
    """Execute a search request; ``transport`` exists for network-free tests."""

    normalized_query = query.strip()
    if not normalized_query:
        return _error_result(query, "Search query must not be empty")
    if not settings.enabled:
        return _error_result(normalized_query, "web_search is disabled")

    payload = deepcopy(settings.payload)
    payload["query"] = normalized_query

    try:
        async with httpx.AsyncClient(
            timeout=settings.timeout,
            verify=settings.verify_ssl,
            transport=transport,
        ) as client:
            response = await client.post(settings.endpoint, json=payload)
            response.raise_for_status()
            data = response.json()
        results = _normalize_results(data, settings)
    except httpx.TimeoutException:
        logger.warning("Configured web search timed out for endpoint %s", settings.endpoint)
        return _error_result(normalized_query, "Search API request timed out")
    except httpx.HTTPStatusError as exc:
        logger.warning(
            "Configured web search returned HTTP %s for endpoint %s",
            exc.response.status_code,
            settings.endpoint,
        )
        return _error_result(normalized_query, f"Search API returned HTTP {exc.response.status_code}")
    except (httpx.HTTPError, json.JSONDecodeError) as exc:
        logger.warning("Configured web search request failed: %s", exc)
        return _error_result(normalized_query, f"Search API request failed: {type(exc).__name__}")
    except ValueError as exc:
        logger.warning("Configured web search returned an invalid response: %s", exc)
        return _error_result(normalized_query, str(exc))

    return json.dumps(
        {
            "query": normalized_query,
            "total_results": len(results),
            "results": results,
        },
        ensure_ascii=False,
    )


@tool("web_search", parse_docstring=True)
async def web_search_tool(query: str) -> str:
    """Search the configured information service for current facts and source passages.

    Args:
        query: Focused search keywords or a natural-language retrieval question.
    """

    try:
        settings = ConfigurableSearchSettings.from_tool_config(get_app_config().get_tool_config("web_search"))
    except ValueError as exc:
        return _error_result(query, str(exc))
    return await execute_search(query, settings)