Initial commit: Javis project

duckduckgo-mcp/src/duckduckgo_mcp/__init__.py

@@ -0,0 +1 @@
+"""duckduckgo_mcp package."""

duckduckgo-mcp/src/duckduckgo_mcp/__main__.py

@@ -0,0 +1,5 @@
+"""Allow ``python -m duckduckgo_mcp`` to start the server."""
+
+from .server import main
+
+main()

duckduckgo-mcp/src/duckduckgo_mcp/server.py

@@ -0,0 +1,199 @@
+"""DuckDuckGo MCP server — web, news, and instant-answer search."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from mcp.server.fastmcp import FastMCP
+
+log = logging.getLogger("duckduckgo_mcp")
+
+mcp = FastMCP(
+    name="duckduckgo",
+    instructions=(
+        "Use these tools to search the web via DuckDuckGo. "
+        "Prefer `ddg_search` for general research, `ddg_news` for recent events, "
+        "and `ddg_answers` for quick factual lookups (calculations, definitions, etc.)."
+    ),
+)
+
+
+# ── helpers ───────────────────────────────────────────────────────────────────
+
+def _get_ddgs():
+    """Import DDGS lazily so import errors surface at call time.
+
+    Tries the current package name ``ddgs`` first, then falls back to the
+    legacy ``duckduckgo_search`` name for environments that haven't migrated.
+    """
+    try:
+        try:
+            from ddgs import DDGS  # new package name (v1+)
+        except ImportError:
+            from duckduckgo_search import DDGS  # legacy name  # type: ignore[import-untyped]
+        return DDGS()
+    except ImportError as exc:
+        raise RuntimeError(
+            "Neither 'ddgs' nor 'duckduckgo-search' is installed. "
+            "Run: pip install ddgs"
+        ) from exc
+
+
+def _fmt_web(results: list[dict[str, Any]]) -> str:
+    if not results:
+        return "No results found."
+    lines: list[str] = []
+    for i, r in enumerate(results, 1):
+        title = r.get("title", "").strip()
+        url = r.get("href", "")
+        body = (r.get("body") or "").strip()
+        lines.append(f"{i}. **{title}**\n   {url}\n   {body}")
+    return "\n\n".join(lines)
+
+
+def _fmt_news(results: list[dict[str, Any]]) -> str:
+    if not results:
+        return "No news results found."
+    lines: list[str] = []
+    for i, r in enumerate(results, 1):
+        title = r.get("title", "").strip()
+        url = r.get("url", r.get("href", ""))
+        date = r.get("date", "")
+        source = r.get("source", "")
+        body = (r.get("body") or "").strip()
+        meta = " · ".join(filter(None, [source, date]))
+        lines.append(f"{i}. **{title}**{(' — ' + meta) if meta else ''}\n   {url}\n   {body}")
+    return "\n\n".join(lines)
+
+
+def _fmt_answers(results: list[dict[str, Any]]) -> str:
+    if not results:
+        return "No instant answers found."
+    lines: list[str] = []
+    for r in results:
+        text = (r.get("text") or r.get("answer") or "").strip()
+        url = r.get("url") or r.get("href") or ""
+        if text:
+            lines.append(text + (f"\n   Source: {url}" if url else ""))
+    return "\n\n".join(lines) if lines else "No instant answers found."
+
+
+# ── tools ─────────────────────────────────────────────────────────────────────
+
+@mcp.tool(
+    description=(
+        "Search the web using DuckDuckGo. "
+        "Returns titles, URLs, and short descriptions for the top results. "
+        "Use this for general research, how-to questions, finding documentation, "
+        "or any open-ended information lookup."
+    )
+)
+def ddg_search(
+    query: str,
+    max_results: int = 10,
+    region: str = "wt-wt",
+    safesearch: str = "moderate",
+) -> str:
+    """Web search via DuckDuckGo.
+
+    Args:
+        query: The search query.
+        max_results: Number of results to return (1–30).
+        region: DuckDuckGo region code, e.g. 'wt-wt' (global), 'de-de', 'us-en'.
+        safesearch: 'on', 'moderate', or 'off'.
+    """
+    max_results = max(1, min(max_results, 30))
+    ddgs = _get_ddgs()
+    try:
+        results = ddgs.text(
+            query,
+            max_results=max_results,
+            region=region,
+            safesearch=safesearch,
+        )
+        return _fmt_web(list(results))
+    except Exception as exc:
+        log.error(f"ddg_search error: {exc}")
+        return f"Search failed: {exc}"
+
+
+@mcp.tool(
+    description=(
+        "Search DuckDuckGo News for recent articles and breaking stories. "
+        "Use this for current events, news about people or companies, "
+        "or any topic where freshness matters."
+    )
+)
+def ddg_news(
+    query: str,
+    max_results: int = 10,
+    region: str = "wt-wt",
+) -> str:
+    """News search via DuckDuckGo.
+
+    Args:
+        query: The news search query.
+        max_results: Number of articles to return (1–30).
+        region: DuckDuckGo region code, e.g. 'wt-wt' (global), 'de-de', 'us-en'.
+    """
+    max_results = max(1, min(max_results, 30))
+    ddgs = _get_ddgs()
+    try:
+        results = ddgs.news(
+            query,
+            max_results=max_results,
+            region=region,
+        )
+        return _fmt_news(list(results))
+    except Exception as exc:
+        log.error(f"ddg_news error: {exc}")
+        return f"News search failed: {exc}"
+
+
+@mcp.tool(
+    description=(
+        "Look up a quick, focused answer for short factual questions — "
+        "definitions, calculations, unit conversions, 'who is X', 'what is Y', etc. "
+        "Returns the top 3 most relevant web results. "
+        "Prefer ddg_search for open-ended research."
+    )
+)
+def ddg_answers(query: str) -> str:
+    """Quick factual lookup via DuckDuckGo.
+
+    The DDG instant-answers API was removed upstream (duckduckgo-search >= 7 /
+    ddgs >= 1).  This tool now returns the top 3 focused web results as the
+    best available substitute.
+
+    Args:
+        query: A short, direct question or lookup term.
+    """
+    ddgs = _get_ddgs()
+    try:
+        # Use native answers() if the installed library still provides it
+        # (kept for forward-compatibility if DDG ever restores the endpoint).
+        if hasattr(ddgs, "answers"):
+            try:
+                results = list(ddgs.answers(query))
+                if results:
+                    return _fmt_answers(results)
+            except Exception as exc:
+                log.debug(f"ddg native answers() failed ({exc}), falling back to text")
+        # Fallback: top-3 web results as a focused answer proxy
+        results = list(ddgs.text(query, max_results=3))
+        return _fmt_web(results) if results else "No results found."
+    except Exception as exc:
+        log.error(f"ddg_answers error: {exc}")
+        return f"Answer lookup failed: {exc}"
+
+
+# ── entry point ───────────────────────────────────────────────────────────────
+
+def main() -> None:
+    logging.basicConfig(level=logging.INFO)
+    mcp.run(transport="stdio")
+
+
+if __name__ == "__main__":
+    main()