Mirror SERPent from Hugging Face Space with bug fixes

Mirrors OrganizedProgrammers/SERPent (https://huggingface.co/spaces/OrganizedProgrammers/SERPent)
with the following fixes applied on top:

- app.py: make /serp/search try all queries concurrently instead of
sequentially, and surface per-query backend failures in the response's
error field instead of silently dropping them when at least one other
query in the batch succeeded.
- app.py: add opt-in SERPENT_API_KEY protection (X-API-Key or Authorization:
Bearer) gating REST and MCP alike via HTTP middleware, closing the
unauthenticated open-scraping-proxy surface. Unset by default, so
behavior is unchanged until explicitly enabled.
- app.py: load docs/docs.md safely (context-managed read, fallback
description) instead of an unguarded open().read() at import time.
- serp.py: raise a clear BrowserUnavailableError instead of an
AttributeError when Playwright failed to start and pw_browser is None.
- serp.py: clamp n_results to [1, 100] in SerpQuery so a negative or huge
value can't reach a scraped page URL/loop bound unclamped.
- scrap.py: fix scrap_patent_bulk_async's patent_ids type hint
(list[int] -> list[str]; patent IDs are strings like "US11930446B2").
- scrap.py: raise a clear error instead of crashing on None.get(...) when
the DC.title meta tag is missing from a scraped page.

.gitattributes

@@ -0,0 +1,35 @@
+*.7z filter=lfs diff=lfs merge=lfs -text
+*.arrow filter=lfs diff=lfs merge=lfs -text
+*.bin filter=lfs diff=lfs merge=lfs -text
+*.bz2 filter=lfs diff=lfs merge=lfs -text
+*.ckpt filter=lfs diff=lfs merge=lfs -text
+*.ftz filter=lfs diff=lfs merge=lfs -text
+*.gz filter=lfs diff=lfs merge=lfs -text
+*.h5 filter=lfs diff=lfs merge=lfs -text
+*.joblib filter=lfs diff=lfs merge=lfs -text
+*.lfs.* filter=lfs diff=lfs merge=lfs -text
+*.mlmodel filter=lfs diff=lfs merge=lfs -text
+*.model filter=lfs diff=lfs merge=lfs -text
+*.msgpack filter=lfs diff=lfs merge=lfs -text
+*.npy filter=lfs diff=lfs merge=lfs -text
+*.npz filter=lfs diff=lfs merge=lfs -text
+*.onnx filter=lfs diff=lfs merge=lfs -text
+*.ot filter=lfs diff=lfs merge=lfs -text
+*.parquet filter=lfs diff=lfs merge=lfs -text
+*.pb filter=lfs diff=lfs merge=lfs -text
+*.pickle filter=lfs diff=lfs merge=lfs -text
+*.pkl filter=lfs diff=lfs merge=lfs -text
+*.pt filter=lfs diff=lfs merge=lfs -text
+*.pth filter=lfs diff=lfs merge=lfs -text
+*.rar filter=lfs diff=lfs merge=lfs -text
+*.safetensors filter=lfs diff=lfs merge=lfs -text
+saved_model/**/* filter=lfs diff=lfs merge=lfs -text
+*.tar.* filter=lfs diff=lfs merge=lfs -text
+*.tar filter=lfs diff=lfs merge=lfs -text
+*.tflite filter=lfs diff=lfs merge=lfs -text
+*.tgz filter=lfs diff=lfs merge=lfs -text
+*.wasm filter=lfs diff=lfs merge=lfs -text
+*.xz filter=lfs diff=lfs merge=lfs -text
+*.zip filter=lfs diff=lfs merge=lfs -text
+*.zst filter=lfs diff=lfs merge=lfs -text
+*tfevents* filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -0,0 +1,4 @@
+__pycache__
+.vscode
+.venv
+.env

Dockerfile

@@ -0,0 +1,18 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+ENV PLAYWRIGHT_BROWSERS_PATH=0
+
+RUN pip install --no-cache-dir playwright && \
+    playwright install-deps chromium && \
+    playwright install chromium
+
+COPY . .
+
+EXPOSE 7860
+
+CMD ["python", "./app.py"]

README.md

@@ -0,0 +1,15 @@
+---
+title: SERPent
+emoji: 🐍
+colorFrom: green
+colorTo: yellow
+sdk: docker
+app_port: 7860
+short_description: A SERP scrapping API for AI projects
+pinned: true
+---
+
+
+# `SERPent`
+
+`SERPent` provides a SERP / scrapping API for use by AI agents / projects.

app.py

@@ -0,0 +1,384 @@
+import asyncio
+import logging
+import os
+import secrets
+from contextlib import asynccontextmanager
+from typing import Optional
+from fastapi import APIRouter, FastAPI, HTTPException, Request
+from fastapi.responses import JSONResponse
+from fastapi.routing import APIRouter
+import httpx
+from httpx import HTTPStatusError
+from pydantic import BaseModel, Field
+from playwright.async_api import async_playwright, Browser, BrowserContext, Page
+import uvicorn
+
+from scrap import PatentScrapBulkResponse, PatentScrapResult, scrap_patent_async, scrap_patent_bulk_async
+from serp import SerpQuery, SerpResults, query_arxiv, query_bing_search, query_brave_search, query_ddg_search, query_google_patents, query_google_scholar
+from ops import OPSBulkResponse, OPSNotConfigured, ops_scrap_patent, ops_scrap_patent_bulk, ops_search, token_manager as ops_token_manager
+from utils import log_gathered_exceptions
+from mcp_server import mount_mcp_server
+
+logging.basicConfig(
+    level=logging.INFO,
+    format='[%(asctime)s][%(levelname)s][%(filename)s:%(lineno)d]: %(message)s',
+    datefmt='%Y-%m-%d %H:%M:%S'
+)
+
+# playwright global context
+playwright = None
+pw_browser: Optional[Browser] = None
+
+# httpx client
+httpx_client = httpx.AsyncClient(timeout=30, limits=httpx.Limits(
+    max_connections=30, max_keepalive_connections=20))
+
+# ===================== Optional API key protection =====================
+# Unset by default, which keeps the API exactly as open as before. Set
+# SERPENT_API_KEY on the deployment to require every request (REST and MCP
+# alike) to carry a matching `X-API-Key` or `Authorization: Bearer <key>`
+# header. This exists to prevent the deployment from being used as an
+# anonymous, unauthenticated scraping proxy by anyone who finds the URL.
+SERPENT_API_KEY = os.environ.get("SERPENT_API_KEY")
+_PUBLIC_PATHS = {"/", "/openapi.json"}
+
+
+def _extract_supplied_key(request: Request) -> str:
+    api_key = request.headers.get("x-api-key")
+    if api_key:
+        return api_key
+    auth = request.headers.get("authorization", "")
+    if auth.lower().startswith("bearer "):
+        return auth[7:]
+    return ""
+
+
+def _load_docs() -> str:
+    try:
+        with open("docs/docs.md", encoding="utf-8") as f:
+            return f.read()
+    except OSError as e:
+        logging.warning(f"Could not read docs/docs.md, using fallback description: {e}")
+        return "SERPent provides a SERP / scrapping API for use by AI agents / projects."
+
+
+@asynccontextmanager
+async def api_lifespan(app: FastAPI):
+    global playwright, pw_browser
+    try:
+        playwright = await async_playwright().start()
+        pw_browser = await playwright.chromium.launch(headless=True)
+        logging.info("Playwright browser started.")
+    except Exception as e:
+        logging.warning(f"Playwright unavailable, browser-based endpoints will fail: {e}")
+    yield
+
+    if pw_browser:
+        await pw_browser.close()
+    if playwright:
+        await playwright.stop()
+
+app = FastAPI(lifespan=api_lifespan, docs_url="/",
+              title="SERPent", description=_load_docs())
+
+
+@app.middleware("http")
+async def api_key_guard(request: Request, call_next):
+    """No-op unless SERPENT_API_KEY is set; then gates every path but the docs."""
+    if SERPENT_API_KEY and request.url.path not in _PUBLIC_PATHS:
+        supplied = _extract_supplied_key(request)
+        if not supplied or not secrets.compare_digest(supplied, SERPENT_API_KEY):
+            return JSONResponse(
+                {"detail": "Missing or invalid API key. Supply it via the X-API-Key or Authorization: Bearer header."},
+                status_code=401,
+            )
+    return await call_next(request)
+
+# Router for scrapping related endpoints
+scrap_router = APIRouter(prefix="/scrap", tags=["scrapping"])
+# Router for SERP-scrapping related endpoints
+serp_router = APIRouter(prefix="/serp", tags=["serp scrapping"])
+# Router for EPO OPS (official patent API) endpoints
+ops_router = APIRouter(prefix="/ops", tags=["EPO OPS"])
+
+# ===================== Search endpoints =====================
+
+
+@serp_router.post("/search_scholar")
+async def search_google_scholar(params: SerpQuery):
+    """Queries google scholar for the specified query"""
+    logging.info(f"Searching Google Scholar for queries: {params.queries}")
+    results = await asyncio.gather(*[query_google_scholar(pw_browser, q, params.n_results) for q in params.queries], return_exceptions=True)
+    log_gathered_exceptions(results, "google scholar search", params)
+
+    # Filter out exceptions and flatten the results
+    filtered_results = [r for r in results if not isinstance(r, Exception)]
+    flattened_results = [
+        item for sublist in filtered_results for item in sublist]
+
+    # all queries failed, return the last exception
+    if len(filtered_results) == 0:
+        return SerpResults(results=[], error=str(results[-1]))
+
+    return SerpResults(results=flattened_results, error=None)
+
+
+@serp_router.post("/search_arxiv")
+async def search_arxiv(params: SerpQuery):
+    """Searches arxiv for the specified queries and returns the found documents."""
+    logging.info(f"Searching Arxiv for queries: {params.queries}")
+    results = await asyncio.gather(*[query_arxiv(httpx_client, q, params.n_results) for q in params.queries], return_exceptions=True)
+    log_gathered_exceptions(results, "arxiv search", params)
+
+    filtered_results = [r for r in results if not isinstance(r, Exception)]
+    flattened_results = [
+        item for sublist in filtered_results for item in sublist]
+
+    if len(filtered_results) == 0:
+        return SerpResults(results=[], error=str(results[-1]))
+
+    return SerpResults(results=flattened_results, error=None)
+
+
+@serp_router.post("/search_patents")
+async def search_patents(params: SerpQuery) -> SerpResults:
+    """Searches google patents for the specified queries and returns the found documents.
+
+    Falls back to the EPO OPS API for any query Google Patents returns nothing
+    for, when OPS credentials are configured.
+    """
+    logging.info(f"Searching Google Patents for queries: {params.queries}")
+    results = await asyncio.gather(*[query_google_patents(pw_browser, q, params.n_results) for q in params.queries], return_exceptions=True)
+    log_gathered_exceptions(results, "google patent search", params)
+
+    # Fall back to OPS for queries that errored or returned no results.
+    if ops_token_manager.configured:
+        for i, res in enumerate(results):
+            if isinstance(res, Exception) or not res:
+                try:
+                    logging.info(
+                        f"Google Patents empty for `{params.queries[i]}`, trying OPS.")
+                    results[i] = await ops_search(httpx_client, params.queries[i], params.n_results)
+                except Exception as e:
+                    logging.warning(f"OPS fallback failed: {e}")
+
+    # Filter out exceptions and flatten the results
+    filtered_results = [r for r in results if not isinstance(r, Exception)]
+    flattened_results = [
+        item for sublist in filtered_results for item in sublist]
+
+    # all queries failed, return the last exception
+    if len(filtered_results) == 0:
+        return SerpResults(results=[], error=str(results[-1]))
+
+    return SerpResults(results=flattened_results, error=None)
+
+
+@serp_router.post("/search_brave")
+async def search_brave(params: SerpQuery) -> SerpResults:
+    """Searches brave search for the specified queries and returns the found documents."""
+    logging.info(f"Searching Brave Search for queries: {params.queries}")
+    results = await asyncio.gather(*[query_brave_search(pw_browser, q, params.n_results) for q in params.queries], return_exceptions=True)
+    log_gathered_exceptions(results, "brave search", params)
+
+    # Filter out exceptions and flatten the results
+    filtered_results = [r for r in results if not isinstance(r, Exception)]
+    flattened_results = [
+        item for sublist in filtered_results for item in sublist]
+
+    # all queries failed, return the last exception
+    if len(filtered_results) == 0:
+        return SerpResults(results=[], error=str(results[-1]))
+
+    return SerpResults(results=flattened_results, error=None)
+
+
+@serp_router.post("/search_bing")
+async def search_bing(params: SerpQuery) -> SerpResults:
+    """Searches Bing search for the specified queries and returns the found documents."""
+    logging.info(f"Searching Bing Search for queries: {params.queries}")
+    results = await asyncio.gather(*[query_bing_search(pw_browser, q, params.n_results) for q in params.queries], return_exceptions=True)
+    log_gathered_exceptions(results, "bing search", params)
+
+    # Filter out exceptions and flatten the results
+    filtered_results = [r for r in results if not isinstance(r, Exception)]
+    flattened_results = [
+        item for sublist in filtered_results for item in sublist]
+
+    # all queries failed, return the last exception
+    if len(filtered_results) == 0:
+        return SerpResults(results=[], error=str(results[-1]))
+
+    return SerpResults(results=flattened_results, error=None)
+
+
+@serp_router.post("/search_duck")
+async def search_duck(params: SerpQuery) -> SerpResults:
+    """Searches duckduckgo for the specified queries and returns the found documents"""
+    logging.info(f"Searching DuckDuckGo for queries: {params.queries}")
+    results = await asyncio.gather(*[query_ddg_search(q, params.n_results) for q in params.queries], return_exceptions=True)
+    log_gathered_exceptions(results, "duckduckgo search", params)
+
+    # Filter out exceptions and flatten the results
+    filtered_results = [r for r in results if not isinstance(r, Exception)]
+    flattened_results = [
+        item for sublist in filtered_results for item in sublist]
+
+    # all queries failed, return the last exception
+    if len(filtered_results) == 0:
+        return SerpResults(results=[], error=str(results[-1]))
+
+    return SerpResults(results=flattened_results, error=None)
+
+
+async def _search_one(q: str, n_results: int) -> tuple[str, list[dict], Optional[str]]:
+    """Try DDG, then Brave, then Bing for a single query; stop at the first success."""
+    backends = [
+        ("DuckDuckGo", lambda: query_ddg_search(q, n_results)),
+        ("Brave Search", lambda: query_brave_search(pw_browser, q, n_results)),
+        ("Bing", lambda: query_bing_search(pw_browser, q, n_results)),
+    ]
+    last_error: Optional[Exception] = None
+    for name, call in backends:
+        try:
+            logging.info(f"Querying {name} with query: `{q}`")
+            return q, await call(), None
+        except Exception as e:
+            logging.error(f"Failed to query {name} with query `{q}`: {e}")
+            last_error = e
+
+    return q, [], f"All backends failed for query '{q}': {last_error}"
+
+
+@serp_router.post("/search")
+async def search(params: SerpQuery):
+    """Attempts to search the specified queries using ALL backends"""
+    outcomes = await asyncio.gather(*[_search_one(q, params.n_results) for q in params.queries])
+
+    results: list[dict] = []
+    errors: list[str] = []
+    for _q, res, err in outcomes:
+        results.extend(res)
+        if err:
+            errors.append(err)
+
+    if len(results) == 0:
+        return SerpResults(results=[], error="; ".join(errors) if errors else "All backends are rate-limited.")
+
+    return SerpResults(results=results, error="; ".join(errors) if errors else None)
+
+# =========================== Scrapping endpoints ===========================
+
+
+@scrap_router.get("/scrap_patent/{patent_id}")
+async def scrap_patent(patent_id: str) -> PatentScrapResult:
+    """Scraps the specified patent from Google Patents.
+
+    Falls back to the EPO OPS API (which covers patents missing from Google
+    Patents) when the scrape fails and OPS credentials are configured.
+    """
+    try:
+        return await scrap_patent_async(httpx_client, f"https://patents.google.com/patent/{patent_id}/en")
+    except HTTPStatusError as e:
+        logging.warning(
+            f"Google Patents returned {e.response.status_code} for {patent_id}.")
+    except Exception as e:
+        logging.warning(f"Failed to scrap patent {patent_id}: {e}")
+
+    if not ops_token_manager.configured:
+        raise HTTPException(
+            status_code=404,
+            detail=f"Patent '{patent_id}' not found on Google Patents (EPO OPS fallback not configured).")
+
+    try:
+        logging.info(f"Trying OPS for patent {patent_id}.")
+        return await ops_scrap_patent(httpx_client, patent_id)
+    except HTTPStatusError as e:
+        if e.response.status_code == 404:
+            raise HTTPException(
+                status_code=404,
+                detail=f"Patent '{patent_id}' not found on Google Patents or EPO OPS.")
+        raise HTTPException(
+            status_code=502,
+            detail=f"EPO OPS returned {e.response.status_code} for '{patent_id}'.")
+    except Exception as e:
+        logging.warning(f"OPS fallback failed for {patent_id}: {e}")
+        raise HTTPException(
+            status_code=502,
+            detail=f"EPO OPS request failed for '{patent_id}': {e}")
+
+
+class ScrapPatentsRequest(BaseModel):
+    """Request model for scrapping multiple patents."""
+    patent_ids: list[str] = Field(...,
+                                  description="List of patent IDs to scrap")
+
+
+@scrap_router.post("/scrap_patents_bulk", response_model=PatentScrapBulkResponse)
+async def scrap_patents(params: ScrapPatentsRequest) -> PatentScrapBulkResponse:
+    """Scraps multiple patents from Google Patents."""
+    patents = await scrap_patent_bulk_async(httpx_client, params.patent_ids)
+    return patents
+
+# =========================== EPO OPS endpoints ===========================
+
+
+@ops_router.post("/search")
+async def ops_keyword_search(params: SerpQuery) -> SerpResults:
+    """Keyword-searches patents via the official EPO OPS API."""
+    logging.info(f"Searching EPO OPS for queries: {params.queries}")
+    results = await asyncio.gather(*[ops_search(httpx_client, q, params.n_results) for q in params.queries], return_exceptions=True)
+    log_gathered_exceptions(results, "OPS search", params)
+
+    filtered_results = [r for r in results if not isinstance(r, Exception)]
+    flattened_results = [
+        item for sublist in filtered_results for item in sublist]
+
+    if len(filtered_results) == 0:
+        return SerpResults(results=[], error=str(results[-1]))
+
+    return SerpResults(results=flattened_results, error=None)
+
+
+@ops_router.get("/scrap_patent/{patent_id}")
+async def ops_get_patent(patent_id: str) -> PatentScrapResult:
+    """Retrieves a patent (biblio + abstract + claims + description) via EPO OPS."""
+    if not ops_token_manager.configured:
+        raise HTTPException(
+            status_code=503,
+            detail="EPO OPS is not configured (OPS_CONSUMER_KEY / OPS_CONSUMER_SECRET missing).")
+    try:
+        return await ops_scrap_patent(httpx_client, patent_id)
+    except OPSNotConfigured:
+        raise HTTPException(status_code=503, detail="EPO OPS is not configured.")
+    except HTTPStatusError as e:
+        if e.response.status_code == 404:
+            raise HTTPException(
+                status_code=404, detail=f"Patent '{patent_id}' not found in EPO OPS.")
+        raise HTTPException(
+            status_code=502,
+            detail=f"EPO OPS returned {e.response.status_code} for '{patent_id}'.")
+    except Exception as e:
+        logging.warning(f"Failed to retrieve patent {patent_id} from OPS: {e}")
+        raise HTTPException(
+            status_code=502, detail=f"EPO OPS request failed for '{patent_id}': {e}")
+
+
+@ops_router.post("/scrap_patents_bulk", response_model=OPSBulkResponse)
+async def ops_get_patents_bulk(params: ScrapPatentsRequest) -> OPSBulkResponse:
+    """Retrieves multiple patents via EPO OPS."""
+    return await ops_scrap_patent_bulk(httpx_client, params.patent_ids)
+
+# ===============================================================================
+
+app.include_router(serp_router)
+app.include_router(scrap_router)
+app.include_router(ops_router)
+
+# =============================== MCP server ===================================
+# Re-exposes every endpoint above as an MCP tool over streamable HTTP at /mcp.
+# Must stay below the include_router() calls, or the routes are not picked up.
+mcp = mount_mcp_server(app)
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=7860)

docs/README_MCP.md

@@ -0,0 +1,145 @@
+# MCP server support
+
+SERPent now speaks two protocols from the same deployment:
+
+| Audience | Endpoint |
+|---|---|
+| REST clients / Swagger UI | `https://<host>/` (docs), `POST /serp/...`, `/scrap/...`, `/ops/...` |
+| MCP clients (Claude, Cursor, agents) | `https://<host>/mcp` — streamable HTTP transport |
+
+## What changed
+
+| File | Change |
+|---|---|
+| `mcp_server.py` | **New.** Builds the MCP server from the FastAPI app and mounts it. |
+| `app.py` | Two lines: `from mcp_server import mount_mcp_server` and `mcp = mount_mcp_server(app)` after the `include_router` calls. |
+| `requirements.txt` | Added `fastmcp>=3.0,<4`. |
+| `Dockerfile` | Unchanged — same image, same port 7860, same `CMD`. |
+
+## How it works
+
+`FastMCP.from_fastapi()` reads the app's OpenAPI schema and turns every route
+into an MCP tool. When a tool is called, the request is dispatched **back into
+the same FastAPI app in-process** over an ASGI transport — no network hop, no
+second server, no duplicated business logic. Add an endpoint, get a tool.
+
+The 12 tools generated today:
+
+```
+search                 search_arxiv          search_google_scholar
+search_patents         search_brave          search_bing
+search_duck            scrap_patent          scrap_patents
+ops_keyword_search     ops_get_patent        ops_get_patents_bulk
+```
+
+Tool names come from the Python handler names rather than FastAPI's generated
+operation ids, so agents see `search_arxiv` instead of
+`search_arxiv_serp_search_arxiv_post`. Descriptions and JSON schemas come from
+your docstrings and Pydantic models — improving a docstring improves the tool.
+To pin a specific tool name, set `operation_id="..."` on the route decorator.
+
+## Connecting a client
+
+**Claude Code / any streamable-HTTP client:**
+
+```bash
+claude mcp add --transport http serpent https://<your-space>.hf.space/mcp
+```
+
+**`.mcp.json` / `claude_desktop_config.json`:**
+
+```json
+{
+  "mcpServers": {
+    "serpent": {
+      "type": "http",
+      "url": "https://<your-space>.hf.space/mcp"
+    }
+  }
+}
+```
+
+**Private HF Space** — pass your token:
+
+```json
+{
+  "mcpServers": {
+    "serpent": {
+      "type": "http",
+      "url": "https://<your-space>.hf.space/mcp",
+      "headers": { "Authorization": "Bearer hf_..." }
+    }
+  }
+}
+```
+
+**If `SERPENT_API_KEY` is set on the deployment** (see Configuration below),
+also pass it, either as `X-API-Key` or reusing the `Authorization: Bearer`
+header:
+
+```json
+{
+  "mcpServers": {
+    "serpent": {
+      "type": "http",
+      "url": "https://<your-space>.hf.space/mcp",
+      "headers": { "X-API-Key": "..." }
+    }
+  }
+}
+```
+
+**Quick check without a client:**
+
+```bash
+curl -X POST http://localhost:7860/mcp \
+  -H 'Content-Type: application/json' \
+  -H 'Accept: application/json, text/event-stream' \
+  -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{
+        "protocolVersion":"2025-06-18","capabilities":{},
+        "clientInfo":{"name":"curl","version":"1"}}}'
+```
+
+Or with the FastMCP CLI: `fastmcp inspect http://localhost:7860/mcp`
+
+## Configuration
+
+All optional, all environment variables:
+
+| Variable | Default | Purpose |
+|---|---|---|
+| `MCP_ENABLED` | `1` | Set to `0` to run REST-only. |
+| `MCP_PATH` | `/mcp` | Where the MCP endpoint lives. |
+| `MCP_STATELESS` | `1` | Stateless HTTP — no session affinity needed behind the HF Spaces proxy. Set `0` for resumable SSE streams. |
+| `MCP_ALLOWED_HOSTS` | `*` | Comma-separated Host allow-list (DNS-rebinding protection). |
+| `MCP_ALLOWED_ORIGINS` | `*` | Comma-separated Origin allow-list (browser clients). |
+| `SERPENT_API_KEY` | unset | When set, every REST and MCP request must carry a matching `X-API-Key` or `Authorization: Bearer <key>` header (except `/` and `/openapi.json`). Unset = fully open, same as before. Set this to stop the deployment being usable as an anonymous scraping proxy by anyone who finds the URL. |
+
+## Notes and caveats
+
+- **Ordering matters.** `mount_mcp_server(app)` must come after every
+  `include_router()` call. Routes registered after it are not exposed as tools,
+  and are not reachable at all — the MCP app is mounted at the root so that a
+  bare `POST /mcp` answers directly instead of 307-redirecting to `/mcp/`
+  (a redirect some MCP clients mishandle).
+- **Lifespans are combined**, so Playwright still starts up and shuts down
+  exactly as before, alongside the MCP session manager.
+- **The `ops_*` tools** are exposed but return 503 without
+  `OPS_CONSUMER_KEY` / `OPS_CONSUMER_SECRET`. The server instructions tell
+  agents to prefer the regular tools, which already fall back to OPS.
+- **Auth.** `SERPENT_API_KEY` (see Configuration) gates both REST and MCP
+  uniformly via an HTTP-level middleware, so it applies before FastMCP's
+  in-process dispatch and needs no per-tool wiring. It's separate from
+  whatever HF Space visibility (public/private) or token you also use.
+- **Tool count.** 12 tools with six overlapping search backends is a lot of
+  surface for a model to choose from. If tool selection gets noisy, pass
+  `route_maps=[...]` to `FastMCP.from_fastapi()` in `mcp_server.py` to exclude
+  the per-backend tools and keep `search`, `search_arxiv`, `search_patents`,
+  `search_google_scholar` and the scrapers.
+
+## Verified
+
+Tested against `fastmcp 3.4.7` / `fastapi 0.141.1`: server boots, all 12 tools
+list over HTTP with correct schemas, tool calls round-trip through the real
+endpoints, and the existing REST routes (`/`, `/openapi.json`, `/redoc`,
+`/serp/*`) are unaffected.

docs/docs.md

@@ -0,0 +1,38 @@
+# `SERPent`
+
+## SERP results scrapping
+
+SERPent exposes an unified API to query SERP (Search Engine Result Pages) for a few common search engines, namely:
+
+- DuckDuckGo
+- Brave
+- Bing
+- Google Patents
+- arXiv
+- Google
+
+The application uses the `playwright` library to control a headless web browser, to simulate normal user activity, to fool the anti-bot measures often present on those sites. See the `/serp/` endpoints for search results scrapping.
+
+
+## Website sources scrapping
+
+SERPent also exposes a few endpoints to scrap the contents of certain sources (patents, scholar). See the `/scrap/` endpoints for supported website sources scrapping.
+
+
+## EPO OPS (official patent API)
+
+SERPent integrates the European Patent Office's Open Patent Services (OPS) v3.2 REST API — a legitimate source for patents that are missing from, or unscrapable via, Google Patents. See the `/ops/` endpoints for keyword search and by-number retrieval (bibliographic data, abstract, claims, description, CPC classifications).
+
+OPS requires OAuth2 client credentials. Set them via the `OPS_CONSUMER_KEY` and `OPS_CONSUMER_SECRET` environment variables (register a free app at the EPO Developer Portal). When configured, the Google Patents endpoints (`/serp/search_patents` and `/scrap/scrap_patent/{id}`) automatically fall back to OPS when they return nothing.
+
+
+## Access control
+
+By default every endpoint is open (no authentication), matching how this API
+has always run. To restrict access, set the `SERPENT_API_KEY` environment
+variable — every request (REST and MCP alike) then needs a matching
+`X-API-Key` or `Authorization: Bearer <key>` header, except this docs page and
+`/openapi.json`. This is recommended once the deployment is public, since it
+otherwise doubles as an anonymous, unauthenticated scraping proxy for anyone
+who finds the URL.
+

mcp_server.py

@@ -0,0 +1,203 @@
+"""MCP (Model Context Protocol) server for SERPent.
+
+Every route registered on the FastAPI app is automatically re-exposed as an MCP
+tool, so a single deployment serves both audiences:
+
+    REST  ->  POST https://<host>/serp/search_arxiv
+    MCP   ->  POST https://<host>/mcp        (streamable HTTP transport)
+
+Nothing here duplicates business logic. FastMCP reads the app's OpenAPI schema
+and, when a tool is called, dispatches the request back into the same FastAPI
+app in-process (via an ASGI transport, no network round-trip), so tools always
+stay in sync with the endpoints.
+
+Usage (see the bottom of `app.py`):
+
+    from mcp_server import mount_mcp_server
+    mcp = mount_mcp_server(app)
+
+Configuration (all optional, read from the environment):
+
+    MCP_ENABLED        "0"/"false" to disable the MCP server entirely.
+    MCP_PATH           Mount path. Default "/mcp".
+    MCP_STATELESS      "0" to keep per-session state. Default stateless, which
+                       is what you want behind the HF Spaces proxy.
+    MCP_ALLOWED_HOSTS  Comma-separated Host allow-list. Default "*".
+    MCP_ALLOWED_ORIGINS Comma-separated Origin allow-list. Default "*".
+    MCP_EXPOSE_ALL     "1" to expose every endpoint as a tool instead of the
+                       curated set (see EXCLUDED_ROUTES below).
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from typing import Optional
+
+from fastapi import FastAPI
+from fastapi.routing import APIRoute
+from fastapi.utils import generate_unique_id
+
+from fastmcp import FastMCP
+from fastmcp.server.providers.openapi import MCPType, RouteMap
+from fastmcp.utilities.lifespan import combine_lifespans
+
+logger = logging.getLogger(__name__)
+
+MCP_SERVER_NAME = "SERPent"
+
+# Endpoints kept out of the MCP tool surface. They stay fully available over
+# REST — this only controls what an LLM sees.
+#
+# Fewer, well-differentiated tools measurably improve tool selection, and both
+# groups below are redundant *for an agent*:
+#
+#   /ops/*        The regular `scrap_patent` / `search_patents` tools already
+#                 fall back to EPO OPS on their own. Calling the OPS tools
+#                 directly skips that fallback, so a patent missing from OPS
+#                 becomes a hard error instead of a Google Patents hit.
+#   search_brave  These are `search` with the fallback chain removed. `search`
+#   search_bing   already tries DuckDuckGo, then Brave, then Bing, so exposing
+#   search_duck   them separately only invites the model to pick a worse path.
+#
+# Set MCP_EXPOSE_ALL=1 to expose everything again, or edit this list.
+EXCLUDED_ROUTES = [
+    r"^/ops/.*",
+    r"^/serp/search_(brave|bing|duck)$",
+]
+
+MCP_INSTRUCTIONS = """\
+SERPent gives you web, academic and patent search plus full-text patent retrieval.
+
+Typical workflow:
+  1. Search with the backend that matches the question:
+     - `search` for general web results (tries DuckDuckGo, then Brave, then Bing)
+     - `search_arxiv` for preprints, `search_google_scholar` for academic papers
+     - `search_patents` for prior art (falls back to the EPO OPS API)
+  2. Feed the patent ids you found to `scrap_patent` (one) or `scrap_patents`
+     (many, and much faster than looping) to get title, abstract, claims and
+     description.
+
+Every search tool accepts a LIST of queries and runs them concurrently, so send
+all your query variations in one call rather than making several calls. Set
+`n_results` to control how many hits come back per query.
+
+`search` and `search_patents` already fall back across several backends
+internally, including the official EPO OPS patent API, so a single call is the
+most thorough option available — there is no lower-level tool to reach for when
+one comes back empty. A patent that returns "not found" is genuinely absent
+from every backend; move on to the next id rather than retrying.
+"""
+
+
+def _env_flag(name: str, default: bool) -> bool:
+    raw = os.getenv(name)
+    if raw is None:
+        return default
+    return raw.strip().lower() not in {"0", "false", "no", "off", ""}
+
+
+def _env_list(name: str, default: list[str]) -> list[str]:
+    raw = os.getenv(name)
+    if not raw:
+        return default
+    return [item.strip() for item in raw.split(",") if item.strip()]
+
+
+def _tool_names(app: FastAPI) -> dict[str, str]:
+    """Map each route's OpenAPI operationId to a clean MCP tool name.
+
+    FastAPI's generated operationIds look like
+    `search_arxiv_serp_search_arxiv_post`, which is what an LLM would otherwise
+    see. The Python handler name (`search_arxiv`) is a far better tool name, so
+    use that instead. Adding an explicit `operation_id=` to a route decorator
+    overrides this.
+    """
+    names: dict[str, str] = {}
+
+    def walk(routes) -> None:
+        for route in routes:
+            # Depending on the FastAPI version, included routers either flatten
+            # into the parent or stay nested, so recurse.
+            if isinstance(route, APIRoute):
+                generate = getattr(route, "generate_unique_id_function", None)
+                # FastAPI stores an unset callable as a DefaultPlaceholder.
+                generate = getattr(generate, "value", generate)
+                if not callable(generate):
+                    generate = generate_unique_id
+                operation_id = route.operation_id or generate(route)
+                names[operation_id] = route.name
+            # Starlette Mounts nest under `.routes`; FastAPI >= 0.141 wraps
+            # included routers in `_IncludedRouter`, which nests under
+            # `.original_router`.
+            sub = getattr(route, "routes", None)
+            if sub:
+                walk(sub)
+            included = getattr(route, "original_router", None)
+            if included is not None and getattr(included, "routes", None):
+                walk(included.routes)
+
+    walk(app.routes)
+    return names
+
+
+def build_mcp_server(app: FastAPI) -> FastMCP:
+    """Build an MCP server exposing the app's routes as tools."""
+    if _env_flag("MCP_EXPOSE_ALL", False):
+        route_maps = None
+        logger.info("MCP exposing all endpoints (MCP_EXPOSE_ALL=1).")
+    else:
+        route_maps = [
+            RouteMap(pattern=pattern, mcp_type=MCPType.EXCLUDE)
+            for pattern in EXCLUDED_ROUTES
+        ]
+
+    mcp = FastMCP.from_fastapi(
+        app=app,
+        name=MCP_SERVER_NAME,
+        mcp_names=_tool_names(app),
+        route_maps=route_maps,
+        # Requests are dispatched straight back into this same app in-process.
+        httpx_client_kwargs={"timeout": 300.0},
+    )
+    mcp.instructions = MCP_INSTRUCTIONS
+    return mcp
+
+
+def mount_mcp_server(app: FastAPI, path: Optional[str] = None) -> Optional[FastMCP]:
+    """Mount the MCP server onto `app` (default: `/mcp`).
+
+    Must be called after every router has been included, otherwise the missing
+    routes will not show up as tools. Returns None when MCP is disabled.
+    """
+    if not _env_flag("MCP_ENABLED", True):
+        logger.info("MCP server disabled (MCP_ENABLED=0).")
+        return None
+
+    path = path or os.getenv("MCP_PATH", "/mcp")
+    path = "/" + path.strip("/")
+
+    mcp = build_mcp_server(app)
+
+    mcp_app = mcp.http_app(
+        path=path,
+        stateless_http=_env_flag("MCP_STATELESS", True),
+        allowed_hosts=_env_list("MCP_ALLOWED_HOSTS", ["*"]),
+        allowed_origins=_env_list("MCP_ALLOWED_ORIGINS", ["*"]),
+    )
+
+    # The MCP session manager needs its own lifespan to run. The app already has
+    # one (Playwright), so run both.
+    app.router.lifespan_context = combine_lifespans(
+        app.router.lifespan_context,
+        mcp_app.lifespan,
+    )
+
+    # The MCP endpoint is `path` *inside* mcp_app, and mcp_app is mounted at the
+    # root. Mounting it at `path` instead would make Starlette answer a bare
+    # `POST /mcp` with a 307 to `/mcp/`, which several MCP clients choke on.
+    # This mount is added last, so all the API routes above still match first.
+    app.mount("/", mcp_app)
+
+    logger.info("MCP server mounted at %s (streamable HTTP).", path)
+    return mcp

ops.py

@@ -0,0 +1,321 @@
+"""EPO Open Patent Services (OPS) v3.2 backend.
+
+Legitimate REST alternative to scraping Espacenet (which sits behind an
+interactive Cloudflare CAPTCHA). Provides keyword search and by-number
+retrieval of the same bibliographic / full-text data.
+
+Auth is OAuth2 client-credentials: the consumer key/secret are exchanged for a
+short-lived (~20 min) bearer token at the token endpoint. Credentials are read
+from the OPS_CONSUMER_KEY / OPS_CONSUMER_SECRET environment variables.
+"""
+
+import asyncio
+import base64
+import logging
+import os
+import re
+import time
+from typing import Optional
+
+from dotenv import load_dotenv
+from httpx import AsyncClient, HTTPStatusError
+from pydantic import BaseModel
+
+from scrap import ClassificationCode, PatentScrapResult
+
+load_dotenv()
+
+OPS_BASE = "https://ops.epo.org/3.2/rest-services"
+OPS_TOKEN_URL = "https://ops.epo.org/3.2/auth/accesstoken"
+
+
+class OPSNotConfigured(Exception):
+    """Raised when OPS credentials are not set in the environment."""
+
+
+class OPSError(Exception):
+    """Raised when OPS returns an error that isn't a plain 'not found'."""
+
+
+class _OPSTokenManager:
+    """Caches and refreshes the OPS OAuth2 bearer token."""
+
+    def __init__(self) -> None:
+        self._key = os.environ.get("OPS_CONSUMER_KEY")
+        self._secret = os.environ.get("OPS_CONSUMER_SECRET")
+        self._token: Optional[str] = None
+        self._expiry: float = 0.0
+        self._lock = asyncio.Lock()
+
+    @property
+    def configured(self) -> bool:
+        return bool(self._key and self._secret)
+
+    async def get_token(self, client: AsyncClient, force: bool = False) -> str:
+        if not self.configured:
+            raise OPSNotConfigured(
+                "OPS_CONSUMER_KEY / OPS_CONSUMER_SECRET are not set.")
+
+        async with self._lock:
+            # 30s safety margin before expiry.
+            if not force and self._token and time.monotonic() < self._expiry - 30:
+                return self._token
+
+            basic = base64.b64encode(
+                f"{self._key}:{self._secret}".encode()).decode()
+            resp = await client.post(
+                OPS_TOKEN_URL,
+                headers={"Authorization": f"Basic {basic}"},
+                data={"grant_type": "client_credentials"},
+            )
+            resp.raise_for_status()
+            data = resp.json()
+            self._token = data["access_token"]
+            self._expiry = time.monotonic() + int(data.get("expires_in", 1199))
+            logging.info("Obtained new OPS access token.")
+            return self._token
+
+
+token_manager = _OPSTokenManager()
+
+
+_KIND_SUFFIX_RE = re.compile(r"[A-Z]\d{0,2}$")
+
+
+def _normalize_epodoc(number: str) -> str:
+    """Strip whitespace and a trailing kind code for the epodoc format.
+
+    OPS epodoc lookups reject the kind suffix (e.g. 'US11930446B2' -> 404), so
+    'US11930446B2' becomes 'US11930446' and 'EP4760514A1' becomes 'EP4760514'.
+    """
+    number = number.strip().replace(" ", "")
+    return _KIND_SUFFIX_RE.sub("", number)
+
+
+async def _ops_get(client: AsyncClient, path: str, params: Optional[dict] = None) -> dict:
+    """GET an OPS endpoint as JSON, refreshing the token once on auth failure."""
+    token = await token_manager.get_token(client)
+    url = f"{OPS_BASE}{path}"
+    headers = {"Authorization": f"Bearer {token}",
+               "Accept": "application/json"}
+    resp = await client.get(url, params=params, headers=headers)
+
+    # A stale/invalid token surfaces as 400/401/403 mentioning the token; refresh once.
+    if resp.status_code in (400, 401, 403) and "token" in resp.text.lower():
+        token = await token_manager.get_token(client, force=True)
+        headers["Authorization"] = f"Bearer {token}"
+        resp = await client.get(url, params=params, headers=headers)
+
+    resp.raise_for_status()
+    return resp.json()
+
+
+# --------------------------- JSON walking helpers ---------------------------
+# OPS returns XML-to-JSON: text nodes are {"$": "..."}, attributes are "@name",
+# and repeated elements become lists (but a single element stays a bare dict).
+
+def _as_list(node) -> list:
+    if node is None:
+        return []
+    return node if isinstance(node, list) else [node]
+
+
+def _text(node) -> Optional[str]:
+    if isinstance(node, dict):
+        return node.get("$")
+    if isinstance(node, str):
+        return node
+    return None
+
+
+def _pick_lang(nodes: list, lang: str = "en"):
+    """From a list of {@lang, ...} nodes, prefer the requested language."""
+    nodes = _as_list(nodes)
+    for n in nodes:
+        if isinstance(n, dict) and n.get("@lang", "").lower() == lang:
+            return n
+    return nodes[0] if nodes else None
+
+
+def _epodoc_number(exchange_doc: dict) -> Optional[str]:
+    """Best-effort human patent number (e.g. 'EP4760514')."""
+    bd = exchange_doc.get("bibliographic-data", {})
+    ref = bd.get("publication-reference", {})
+    for doc_id in _as_list(ref.get("document-id")):
+        if doc_id.get("@document-id-type") == "epodoc":
+            return _text(doc_id.get("doc-number"))
+    country = exchange_doc.get("@country", "")
+    number = exchange_doc.get("@doc-number", "")
+    return f"{country}{number}" if (country or number) else None
+
+
+def _extract_title(bibliographic_data: dict) -> Optional[str]:
+    title = _pick_lang(bibliographic_data.get("invention-title"))
+    return _text(title) if title else None
+
+
+def _extract_abstract(exchange_doc: dict) -> Optional[str]:
+    abstract = _pick_lang(exchange_doc.get("abstract"))
+    if not abstract:
+        return None
+    paras = [_text(p) for p in _as_list(abstract.get("p"))]
+    joined = " ".join(p for p in paras if p)
+    return joined or None
+
+
+def _extract_classifications(bibliographic_data: dict) -> Optional[list[ClassificationCode]]:
+    """Assemble CPC symbols from patent-classifications (no descriptions in OPS)."""
+    codes: list[ClassificationCode] = []
+    seen: set[str] = set()
+    container = bibliographic_data.get("patent-classifications", {})
+    for pc in _as_list(container.get("patent-classification")):
+        section = _text(pc.get("section")) or ""
+        klass = _text(pc.get("class")) or ""
+        subclass = _text(pc.get("subclass")) or ""
+        main_group = _text(pc.get("main-group")) or ""
+        subgroup = _text(pc.get("subgroup")) or ""
+        if not (section and klass and subclass and main_group):
+            continue
+        code = f"{section}{klass}{subclass}{main_group}/{subgroup}".rstrip("/")
+        if code not in seen:
+            seen.add(code)
+            codes.append(ClassificationCode(code=code, description=""))
+    return codes or None
+
+
+def _fulltext_paragraphs(data: dict, section: str) -> Optional[str]:
+    """Extract text from a fulltext service response (claims / description)."""
+    wpd = data.get("ops:world-patent-data", {})
+    ftxt_docs = wpd.get("ftxt:fulltext-documents", {})
+    doc = ftxt_docs.get("ftxt:fulltext-document")
+    doc = doc[0] if isinstance(doc, list) else doc
+    if not doc:
+        return None
+    node = _pick_lang(doc.get(section)) if isinstance(
+        doc.get(section), list) else doc.get(section)
+    if not node:
+        return None
+
+    if section == "claims":
+        lines: list[str] = []
+        for claim in _as_list(node.get("claim")):
+            for ct in _as_list(claim.get("claim-text")):
+                t = _text(ct)
+                if t:
+                    lines.append(t)
+        return "\n".join(lines) or None
+
+    # description
+    paras = [_text(p) for p in _as_list(node.get("p"))]
+    return "\n".join(p for p in paras if p) or None
+
+
+# ------------------------------- Public API --------------------------------
+
+def _to_cql(query: str) -> str:
+    """Turn a free-text query into a CQL expression OPS accepts.
+
+    OPS only auto-wraps a single bare word; multi-word queries like
+    'agentic ai' are invalid CQL and 404. Queries already containing a
+    relational operator ('=') are assumed to be hand-written CQL and passed
+    through unchanged.
+    """
+    q = query.strip()
+    if not q or "=" in q:
+        return q
+    escaped = q.replace('"', " ").strip()
+    return f'txt all "{escaped}"'
+
+
+async def ops_search(client: AsyncClient, query: str, n_results: int) -> list[dict]:
+    """Keyword search against OPS published-data, normalized to SERP dicts."""
+    n_results = max(1, min(n_results, 100))  # OPS caps a page at 100.
+    try:
+        data = await _ops_get(
+            client,
+            "/published-data/search/biblio",
+            params={"q": _to_cql(query), "Range": f"1-{n_results}"},
+        )
+    except HTTPStatusError as e:
+        # OPS returns 404 (SERVER.EntityNotFound) when a search has no hits.
+        if e.response.status_code == 404:
+            return []
+        raise
+    search = data.get("ops:world-patent-data", {}).get("ops:biblio-search", {})
+    result = search.get("ops:search-result", {})
+    results: list[dict] = []
+    for entry in _as_list(result.get("exchange-documents")):
+        ed = entry.get("exchange-document") if isinstance(entry, dict) else None
+        if not ed:
+            continue
+        number = _epodoc_number(ed)
+        bd = ed.get("bibliographic-data", {})
+        results.append({
+            "title": _extract_title(bd) or number or "Untitled",
+            "body": _extract_abstract(ed) or "",
+            "href": f"https://worldwide.espacenet.com/patent/search?q={number}" if number else "",
+            "id": number,
+        })
+    return results
+
+
+async def ops_biblio(client: AsyncClient, number: str, doc_type: str = "publication",
+                     fmt: str = "epodoc") -> dict:
+    if fmt == "epodoc":
+        number = _normalize_epodoc(number)
+    data = await _ops_get(
+        client, f"/published-data/{doc_type}/{fmt}/{number}/biblio")
+    return data["ops:world-patent-data"]["exchange-documents"]["exchange-document"]
+
+
+async def _ops_fulltext(client: AsyncClient, number: str, section: str,
+                        doc_type: str = "publication", fmt: str = "epodoc") -> Optional[str]:
+    """Fetch claims/description full text; returns None when unavailable (404)."""
+    if fmt == "epodoc":
+        number = _normalize_epodoc(number)
+    try:
+        data = await _ops_get(client, f"/published-data/{doc_type}/{fmt}/{number}/{section}")
+    except HTTPStatusError as e:
+        if e.response.status_code == 404:
+            return None
+        raise
+    return _fulltext_paragraphs(data, section)
+
+
+async def ops_scrap_patent(client: AsyncClient, number: str, doc_type: str = "publication",
+                           fmt: str = "epodoc") -> PatentScrapResult:
+    """Retrieve a patent via OPS and shape it like the Google Patents scraper."""
+    exchange_doc = await ops_biblio(client, number, doc_type, fmt)
+    bd = exchange_doc.get("bibliographic-data", {})
+
+    claims, description = await asyncio.gather(
+        _ops_fulltext(client, number, "claims", doc_type, fmt),
+        _ops_fulltext(client, number, "description", doc_type, fmt),
+        return_exceptions=True,
+    )
+    claims = None if isinstance(claims, Exception) else claims
+    description = None if isinstance(description, Exception) else description
+
+    return PatentScrapResult(
+        title=_extract_title(bd) or (_epodoc_number(exchange_doc) or number),
+        abstract=_extract_abstract(exchange_doc),
+        description=description,
+        claims=claims,
+        field_of_invention=None,
+        background=None,
+        classifications=_extract_classifications(bd),
+    )
+
+
+class OPSBulkResponse(BaseModel):
+    patents: list[PatentScrapResult]
+    failed_ids: list[str]
+
+
+async def ops_scrap_patent_bulk(client: AsyncClient, numbers: list[str]) -> OPSBulkResponse:
+    results = await asyncio.gather(
+        *[ops_scrap_patent(client, n) for n in numbers], return_exceptions=True)
+    patents = [r for r in results if not isinstance(r, Exception)]
+    failed = [numbers[i] for i, r in enumerate(
+        results) if isinstance(r, Exception)]
+    return OPSBulkResponse(patents=patents, failed_ids=failed)

requirements.txt

@@ -0,0 +1,10 @@
+fastapi
+uvicorn
+pydantic
+playwright
+duckduckgo_search
+beautifulsoup4
+httpx
+lxml
+python-dotenv
+fastmcp>=3.0,<4

scrap.py

@@ -0,0 +1,139 @@
+import asyncio
+import logging
+import re
+from typing import Optional
+from httpx import AsyncClient
+from bs4 import BeautifulSoup
+from pydantic import BaseModel
+
+
+class ClassificationCode(BaseModel):
+    """A single CPC or IPC classification code with its description."""
+    code: str
+    description: str
+
+
+class PatentScrapResult(BaseModel):
+    """Schema for the result of scraping a google patents page."""
+    # The title of the patent.
+    title: str
+    # The abstract of the patent, if available.
+    abstract: Optional[str] = None
+    # The full description of the patent containing the field of the invention, background, summary, etc.
+    description: Optional[str] = None
+    # The full claims of the patent.
+    claims: Optional[str] = None
+    # The field of the invention, if available.
+    field_of_invention: Optional[str] = None
+    # The background of the invention, if available.
+    background: Optional[str] = None
+    # CPC and IPC classification codes with descriptions.
+    classifications: Optional[list[ClassificationCode]] = None
+
+
+async def scrap_patent_async(client: AsyncClient, patent_url: str) -> PatentScrapResult:
+    headers = {
+        "User-Agent": "Mozilla/5.0 (compatible; GPTBot/1.0; +https://openai.com/gptbot)"
+    }
+    response = await client.get(patent_url, headers=headers)
+    response.raise_for_status()
+
+    soup = BeautifulSoup(response.text, "html.parser")
+
+    # Abstract
+    abstract_div = soup.find("div", {"class": "abstract"})
+    abstract = abstract_div.get_text(
+        strip=True) if abstract_div else None
+
+    # Description
+    description_section = soup.find("section", itemprop="description")
+    description = description_section.get_text(
+        separator="\n", strip=True) if description_section else None
+
+    # Field of the Invention
+    invention_field_match = re.findall(
+        r"(FIELD OF THE INVENTION|TECHNICAL FIELD)(.*?)(?:(BACKGROUND|BACKGROUND OF THE INVENTION|SUMMARY|BRIEF SUMMARY|DETAILED DESCRIPTION|DESCRIPTION OF THE RELATED ART))", description, re.IGNORECASE | re.DOTALL) if description_section else None
+    invention_field = invention_field_match[0][1].strip(
+    ) if invention_field_match else None
+
+    # Background of the Invention
+    invention_background_match = re.findall(
+        r"(BACKGROUND OF THE INVENTION|BACKGROUND)(.*?)(?:(SUMMARY|BRIEF SUMMARY|DETAILED DESCRIPTION|DESCRIPTION OF THE PREFERRED EMBODIMENTS|DESCRIPTION))", description, re.IGNORECASE | re.DOTALL) if description_section else None
+    invention_background = invention_background_match[0][1].strip(
+    ) if invention_background_match else None
+
+    # Claims
+    claims_section = soup.find("section", itemprop="claims")
+    claims = claims_section.get_text(
+        separator="\n", strip=True) if claims_section else None
+
+    # Patent Title
+    meta_title_tag = soup.find("meta", {"name": "DC.title"})
+    if meta_title_tag is None or not meta_title_tag.get("content"):
+        # Unexpected page layout (interstitial, non-patent page, markup
+        # change) rather than a normal 4xx/5xx - raise so callers (the OPS
+        # fallback in app.py, or scrap_patent_bulk_async's gather) treat it
+        # as a failed scrape instead of crashing on `None.get(...)`.
+        raise ValueError(
+            f"Could not find a patent title (meta[name=DC.title]) at {patent_url}; "
+            "the page may not be a valid Google Patents patent page.")
+    meta_title = meta_title_tag.get("content").strip()
+
+    # Patent publication number
+    # pub_num = soup.select_one("h2#pubnum").get_text(strip=True)
+    # get the h2 with id ="pubnum" and extract the text
+
+    # Classification codes (CPC + IPC, flat list, deduplicated by code).
+    # Google Patents renders each code in a <span> inside a <li>. Leaf entries
+    # (no nested <ul>) carry exactly one code; parent entries are breadcrumbs.
+    leaf_code_re = re.compile(r'^[A-Z]\d{2}[A-Z]\d+/\d+$')
+    classifications = []
+    seen_codes: set[str] = set()
+    for li in soup.find_all("li"):
+        if li.find("ul"):
+            continue
+        span = li.find("span")
+        if not span:
+            continue
+        code = span.get_text(strip=True)
+        if leaf_code_re.match(code) and code not in seen_codes:
+            seen_codes.add(code)
+            full_text = li.get_text(separator=" ", strip=True)
+            desc = full_text.replace(code, "", 1).strip().lstrip("—").lstrip("-").strip()
+            classifications.append(ClassificationCode(code=code, description=desc))
+
+    return PatentScrapResult(
+        # publication_number=pub_num,
+        abstract=abstract,
+        description=description,
+        claims=claims,
+        title=meta_title,
+        field_of_invention=invention_field,
+        background=invention_background,
+        classifications=classifications or None
+    )
+
+
+class PatentScrapBulkResponse(BaseModel):
+    """Response model for bulk patent scraping."""
+    patents: list[PatentScrapResult]
+    failed_ids: list[str]
+
+
+async def scrap_patent_bulk_async(client: AsyncClient, patent_ids: list[str]) -> PatentScrapBulkResponse:
+    """Scrape multiple patents asynchronously."""
+    urls = [
+        f"https://patents.google.com/patent/{pid}/en" for pid in patent_ids]
+    results = await asyncio.gather(*[scrap_patent_async(client, url) for url in urls], return_exceptions=True)
+
+    filtered_results = [
+        res for res in results if not isinstance(res, Exception)]
+
+    failed_ids = [
+        patent_ids[i] for i, res in enumerate(results) if isinstance(res, Exception)
+    ]
+
+    return PatentScrapBulkResponse(
+        patents=filtered_results,
+        failed_ids=failed_ids
+    )

serp.py

@@ -0,0 +1,296 @@
+from contextlib import asynccontextmanager
+from typing import Optional
+from duckduckgo_search import DDGS
+import httpx
+from pydantic import BaseModel, Field, field_validator
+from playwright.async_api import Browser, BrowserContext, Page, TimeoutError
+from urllib.parse import quote_plus
+import logging
+import re
+from lxml import etree
+from asyncio import Semaphore
+
+# Concurrency limit for Playwright browser contexts.
+# This is to prevent too many concurrent browser contexts from being created,
+PLAYWRIGHT_CONCURRENCY_LIMIT = 10
+
+
+class SerpQuery(BaseModel):
+    queries: list[str] = Field(...,
+                               description="The list of queries to search for")
+    n_results: int = Field(
+        10, description="Number of results to return for each query. Valid values are 10, 25, 50 and 100")
+
+    @field_validator("n_results")
+    @classmethod
+    def _clamp_n_results(cls, v: int) -> int:
+        """Keep callers from passing a negative or absurdly large value straight into
+        a page URL or loop bound (every backend but OPS did this unclamped)."""
+        return max(1, min(v, 100))
+
+
+class SerpResults(BaseModel):
+    """Model for SERP scrapping results"""
+    error: Optional[str]
+    results: Optional[list[dict]]
+
+
+class BraveSearchBlockedException(Exception):
+    """Dummy exception to detect when the headless browser is flagged as suspicious."""
+
+    def __init__(self, *args):
+        super().__init__("Brave Search blocked the request, likely due to flagging browser as suspicious")
+    pass
+
+
+class BrowserUnavailableError(Exception):
+    """Raised when a Playwright-backed query is attempted but the browser failed to start."""
+
+    def __init__(self, *args):
+        super().__init__("Playwright browser is not available (it failed to start at startup).")
+
+
+_PLAYWRIGHT_CONCURRENCY_SEMAPHORE = Semaphore(PLAYWRIGHT_CONCURRENCY_LIMIT)
+
+
+@asynccontextmanager
+async def playwright_open_page(browser: Optional[Browser]):
+    """Context manager for playwright pages"""
+    if browser is None:
+        raise BrowserUnavailableError()
+    async with _PLAYWRIGHT_CONCURRENCY_SEMAPHORE:
+        context: BrowserContext = await browser.new_context()
+        page: Page = await context.new_page()
+        try:
+            yield page
+        finally:
+            await page.close()
+            await context.close()
+
+
+async def query_google_scholar(browser: Browser, q: str, n_results: int = 10):
+    """Queries google scholar for the specified query and number of results. Returns relevant papers"""
+
+    async with playwright_open_page(browser) as page:
+
+        async def _block_resources(route, request):
+            if request.resource_type in ["stylesheet", "image"]:
+                await route.abort()
+            else:
+                await route.continue_()
+
+        await page.route("**/*", _block_resources)
+
+        url = f"https://scholar.google.com/scholar?q={quote_plus(q)}&num={n_results}"
+        await page.goto(url)
+
+        await page.wait_for_selector("div.gs_ri")
+
+        items = await page.locator("div.gs_ri").all()
+        results = []
+        for item in items[:n_results]:
+            title = await item.locator("h3").inner_text(timeout=1000)
+            body = await item.locator("div.gs_rs").inner_text(timeout=1000)
+            href = await item.locator("h3 > a").get_attribute("href")
+
+            results.append({
+                "title": title,
+                "body": body,
+                "href": href
+            })
+
+    return results
+
+
+async def query_google_patents(browser: Browser, q: str, n_results: int = 10):
+    """Queries google patents for the specified query and number of results. Returns relevant patents"""
+
+    # regex to locate a patent id
+    PATENT_ID_REGEX = r"\b[A-Z]{2}\d{6,}(?:[A-Z]\d?)?\b"
+
+    async with playwright_open_page(browser) as page:
+
+        async def _block_resources(route, request):
+            if request.resource_type in ["stylesheet", "image"]:
+                await route.abort()
+            else:
+                await route.continue_()
+
+        await page.route("**/*", _block_resources)
+
+        url = f"https://patents.google.com/?q={quote_plus(q)}&num={n_results}"
+        await page.goto(url)
+
+        # Wait for at least one search result item to appear
+        # This ensures the page has loaded enough to start scraping
+        await page.wait_for_function(
+            f"""() => document.querySelectorAll('search-result-item').length >= 1""",
+            timeout=30_000
+        )
+
+        items = await page.locator("search-result-item").all()
+        results = []
+        for item in items:
+            text = " ".join(await item.locator("span").all_inner_texts())
+            match = re.search(PATENT_ID_REGEX, text)
+            if not match:
+                continue
+
+            patent_id = match.group()
+
+            try:
+                title = await item.locator("h3, h4").first.inner_text(timeout=1000)
+                body = await item.locator("div.abstract, div.result-snippet, .snippet, .result-text").first.inner_text(timeout=1000)
+            except:
+                continue  # If we can't get title or body, skip this item
+
+            results.append({
+                "id": patent_id,
+                "href": f"https://patents.google.com/patent/{patent_id}/en",
+                "title": title,
+                "body": body
+            })
+
+    return results[:n_results]
+
+
+async def query_brave_search(browser: Browser, q: str, n_results: int = 10):
+    """Queries Brave Search for the specified query."""
+
+    async with playwright_open_page(browser) as page:
+
+        async def _block_resources(route, request):
+            if request.resource_type in ["stylesheet", "image"]:
+                await route.abort()
+            else:
+                await route.continue_()
+
+        await page.route("**/*", _block_resources)
+
+        url = f"https://search.brave.com/search?q={quote_plus(q)}"
+        await page.goto(url)
+
+        results_cards = await page.locator('.snippet').all()
+
+        if len(results_cards) == 0:
+            page_content = await page.content()
+
+            if "suspicious" in page_content:
+                raise BraveSearchBlockedException()
+
+        results = []
+
+        for result in results_cards:
+            title = await result.locator('.title').all_inner_texts()
+            description = await result.locator('.snippet-description').all_inner_texts()
+            url = await result.locator('a').nth(0).get_attribute('href')
+
+            # Filter out results with no URL or brave-specific URLs
+            if url is None or url.startswith('/'):
+                continue
+
+            results.append({
+                "title": title[0] if title else "",
+                "body": description[0] if description else "",
+                "href": url
+            })
+
+            if len(results) >= n_results:
+                break
+
+        return results
+
+
+async def query_bing_search(browser: Browser, q: str, n_results: int = 10):
+    """Queries bing search for the specified query"""
+    async with playwright_open_page(browser) as page:
+        async def _block_resources(route, request):
+            if request.resource_type in ["stylesheet", "image"]:
+                await route.abort()
+            else:
+                await route.continue_()
+
+        await page.route("**/*", _block_resources)
+
+        url = f"https://www.bing.com/search?q={quote_plus(q)}"
+        await page.goto(url)
+
+        await page.wait_for_selector("li.b_algo")
+
+        results = []
+
+        items = await page.query_selector_all("li.b_algo")
+        for item in items[:n_results]:
+            title_el = await item.query_selector("h2 > a")
+            url = await title_el.get_attribute("href") if title_el else None
+            title = await title_el.inner_text() if title_el else ""
+
+            snippet = ""
+
+            # Try several fallback selectors
+            for selector in [
+                "div.b_caption p",  # typical snippet
+                "div.b_caption",    # sometimes snippet is here
+                "div.b_snippet",    # used in some result types
+                "div.b_text",       # used in some panels
+                "p"                 # fallback to any paragraph
+            ]:
+                snippet_el = await item.query_selector(selector)
+                if snippet_el:
+                    snippet = await snippet_el.inner_text()
+                    if snippet.strip():
+                        break
+
+            if title and url:
+                results.append({
+                    "title": title.strip(),
+                    "href": url.strip(),
+                    "body": snippet.strip()
+                })
+
+        return results
+
+
+async def query_ddg_search(q: str, n_results: int = 10):
+    """Queries duckduckgo search for the specified query"""
+    ddgs = DDGS()
+    results = []
+
+    for result in ddgs.text(q, max_results=n_results):
+        results.append(
+            {"title": result["title"], "body": result["body"], "href": result["href"]})
+
+    return results
+
+
+async def query_arxiv(client: httpx.AsyncClient, query: str, max_results: int = 3):
+    """Searches arXiv for the specified query and returns a list of results with titles and PDF URLs."""
+    ATOM_NAMESPACE = {'atom': 'http://www.w3.org/2005/Atom'}
+    ARXIV_API_URL = 'https://export.arxiv.org/api/query?'
+
+    search_params = {
+        'search_query': query,
+        'start': 0,
+        'max_results': max_results
+    }
+    query_url = ARXIV_API_URL
+
+    response = await client.get(query_url, params=search_params)
+    response.raise_for_status()
+
+    root = etree.fromstring(response.content)
+    entries = root.findall('atom:entry', ATOM_NAMESPACE)
+
+    results = []
+    for entry in entries:
+        title = entry.find(
+            'atom:title', ATOM_NAMESPACE).text.strip().replace('\n', ' ')
+        id = entry.find('atom:id', ATOM_NAMESPACE).text.strip()
+        pdf_url = entry.find(
+            'atom:id', ATOM_NAMESPACE).text.replace('/abs/', '/pdf/')
+        summary = entry.find(
+            'atom:summary', ATOM_NAMESPACE).text.strip()
+        results.append({'title': title, 'href': pdf_url,
+                       'body': summary, 'id': id})
+
+    return results

utils.py

@@ -0,0 +1,9 @@
+from asyncio.log import logger
+from serp import SerpQuery
+
+
+def log_gathered_exceptions(results: list, context: str, params: SerpQuery):
+    """Logs gathered exceptions with context and parameters."""
+    for exc, q in zip(results, params.queries):
+        if isinstance(exc, Exception):
+            logger.warning(f"Error during {context} for query '{q}': {exc}")