cyper-rag-server

"""
tools/web_tool.py — MCP tool schema + handler for web_search.
"""

from __future__ import annotations

from mcp import types

from src.search.providers import WebSearchEngine
from src.utils.cache import TTLCache
from src.utils.logging import get_logger

log = get_logger("tools.web")


def build_web_tool_schema(provider: str, max_results: int) -> types.Tool:
    return types.Tool(
        name="web_search",
        description=(
            f"Search the live web for current information not available in local documents. "
            f"Provider: {provider}. "
            "Returns title, URL, and a content snippet for each result. "
            "Use this when the question requires recent events, news, or external knowledge."
        ),
        inputSchema={
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "Web search query string.",
                },
                "max_results": {
                    "type": "integer",
                    "description": f"Number of results to return (default: {max_results}, max: 10).",
                    "default": max_results,
                },
            },
            "required": ["query"],
        },
    )


async def handle_web_search(
    arguments: dict,
    engine: WebSearchEngine,
    cache: TTLCache[str],
    cache_ttl: float,
    default_max_results: int,
) -> list[types.TextContent]:
    """Execute a web_search tool call."""
    query       = arguments.get("query", "").strip()
    max_results = min(int(arguments.get("max_results", default_max_results)), 10)

    if not query:
        return [types.TextContent(type="text", text="Error: 'query' must not be empty.")]

    cache_key = f"{query}::{max_results}"
    cached = cache.get(cache_key)
    if cached is not None:
        log.info("Web cache hit: %s…", query[:60])
        return [
            types.TextContent(
                type="text",
                text=f"[Cached result — TTL {cache_ttl:.0f}s]\n\n{cached}",
            )
        ]

    results = await engine.search(query, max_results=max_results)

    if not results:
        return [
            types.TextContent(
                type="text",
                text=f"No web results found for: '{query}'. Try rephrasing.",
            )
        ]

    lines = [f"## Web Search Results — '{query}'\n"]
    for i, r in enumerate(results, start=1):
        lines.append(f"### [{i}] {r['title']}")
        lines.append(f"URL: {r['url']}")
        lines.append(r["snippet"])
        lines.append("\n---\n")

    output = "\n".join(lines)
    cache.set(cache_key, output)
    return [types.TextContent(type="text", text=output)]