Add /scrape endpoint and scrape_url MCP tool for direct URL inspection

Implements upstream issue MarcellM01#10: a query-focused, token-budgeted way to inspect
a URL the caller already knows, without first running a web search. Both the
FastAPI POST /scrape endpoint and the MCP scrape_url tool delegate to a
single service function that crawls the page through the existing Crawl4AI
path (with PDF/DOCX support), chunks the markdown, ranks chunks against the
query with BM25, selects chunks under the max_tokens budget, and returns a
URL-grounded answer prompt the caller's LLM can answer from directly.

The grounded answer prompt now lives in services/grounded_prompt_service so
that /research and /scrape share the same builder. URL safety checks (scheme
allowlist, credential rejection, IP literal rejection, DNS rebinding
mitigation by rejecting any non-public resolved address, blocked_domains
enforcement) live in services/url_safety_service and run before fetching as
well as on the final URL after redirects. Stable error codes (invalid_url,
blocked_url, unsupported_document, empty_content, fetch_failed,
fetch_timeout) map to HTTP 4xx/5xx for FastAPI and to a code-prefixed
ValueError for MCP. No new runtime dependencies were added; metadata
extraction reuses the dict from Crawl4AI with a small regex fallback over
the page HTML.

Author: benmaster82

README.md

@@ -60,13 +60,16 @@ Stop and remove the containers later with:
 docker compose -f "https://github.com/MarcellM01/TinySearch.git#main:compose.quickstart.yaml" down
 ```
 
-TinySearch exposes one MCP tool:
+TinySearch exposes two MCP tools:
 
 ```text
 research(query)
+scrape_url(url, query, max_tokens=4000)
 ```
 
-Pass the user's question as-is. TinySearch searches, crawls, reranks, and
+Pass the user's question as-is. `research` searches, crawls, reranks, and
+returns the grounded prompt in `answer`. `scrape_url` inspects a specific URL
+the caller already knows, applies the same ranking and token budget, and
 returns the grounded prompt in `answer`.
 
 ## Community
@@ -279,8 +282,38 @@ Endpoints:
 - `GET /health`
 - `GET /web_search?query=...`
 - `POST /site_crawl`
+- `POST /scrape`
 - `POST /research`
 
+`POST /scrape` accepts a JSON body with `url` (required), `query` (required,
+non-empty), `max_tokens` (optional, default 4000) and `include_metadata`
+(optional, default true). The response includes a `URL-GROUNDED ANSWER PROMPT`
+in `answer`, plus `content_tokens`, `answer_tokens`, `truncated`, `url`,
+`title`, `retrieved_at` (aware UTC) and best-effort `metadata`
+(`description`, `author`, `published_date`).
+
+Errors return `{"detail": {"code", "message"}}` with stable codes:
+`invalid_url` (400), `blocked_url` (403), `unsupported_document` (415),
+`empty_content` (422), `fetch_failed` (502), `fetch_timeout` (504).
+
+### URL safety
+
+`/scrape` and `scrape_url` accept arbitrary user-supplied URLs and enforce
+the following checks before fetching:
+
+- only `http` and `https` schemes
+- URLs with embedded credentials are rejected
+- IP literals and resolved addresses that are loopback, private, link-local,
+  multicast, reserved or unspecified are rejected (DNS rebinding is mitigated
+  by rejecting if **any** resolved address is non-public, not just one)
+- the configured `blocked_domains` list is applied to both the initial URL
+  and the final URL reported by the crawler after redirects
+
+Crawl4AI does not expose intermediate redirect hops, so the safety check runs
+on the initial URL and the final URL. If you need stricter handling for
+redirect chains, run TinySearch behind an egress proxy that enforces your
+policy.
+
 ## Configuration
 
 Tune research defaults in `configs/research_config.json`. Set

pipelines/agentic_research.py

@@ -36,6 +36,7 @@
     resolve_embedding_tokenizer_name,
 )
 from services.chunk_pool_selection_service import select_chunks_with_quota_and_fill
+from services.grounded_prompt_service import format_search_grounded_prompt
 from services.hybrid_embed_search_service import EmbeddingFn, rank_chunks_hybrid
 from services.research_config_service import (
     config_trace_path,
@@ -59,8 +60,6 @@
 _HTTP_URL = re.compile(r"^https?://", re.IGNORECASE)
 DEFAULT_DENSE_QUERY_PREFIX = "task: search result | query: "
 DEFAULT_DENSE_DOCUMENT_PREFIX = "title: none | text: "
-PROMPT_RULE = "=" * 88
-FIELD_RULE = "======"
 
 
 @dataclass(frozen=True)
@@ -130,119 +129,6 @@ def _search_chunk(result: SearchResult) -> dict[str, Any]:
     }
 
 
-def _format_relevant_text(chunks: Sequence[dict[str, Any]]) -> str:
-    blocks: list[str] = []
-    for ordinal, chunk in enumerate(chunks, start=1):
-        text = str(chunk.get("text") or "").strip()
-        if not text:
-            continue
-        blocks.extend(
-            [
-                f"----- RELEVANT CHUNK {ordinal} -----",
-                text,
-            ]
-        )
-    return "\n".join(blocks).strip()
-
-
-def _format_results_prompt(
-    *,
-    question: str,
-    results: Sequence[dict[str, Any]],
-    today: str | None = None,
-) -> str:
-    clean_question = question.strip()
-    today_text = today or datetime.now(UTC).date().isoformat()
-    lines = [
-        PROMPT_RULE,
-        "SEARCH-GROUNDED ANSWER PROMPT",
-        PROMPT_RULE,
-        "",
-        "QUESTION",
-        PROMPT_RULE,
-        clean_question,
-        PROMPT_RULE,
-        "",
-        "TODAY",
-        PROMPT_RULE,
-        today_text,
-        PROMPT_RULE,
-        "",
-        "CRITICAL INSTRUCTIONS",
-        PROMPT_RULE,
-        "You are answering the QUESTION using only the text under RESULTS.",
-        "First resolve any relative date in the QUESTION using TODAY.",
-        f"TODAY is {today_text!r}.",
-        f"For example, 'last year' means calendar year {int(today_text.split('-')[0]) - 1}.",
-        "Use only facts directly supported by RESULTS.",
-        "Do not use your own knowledge.",
-        "Do not add extra historical claims unless directly supported by RESULTS.",
-        "Do not infer 'first ever', 'most recent', 'record', or franchise history unless RESULTS explicitly support it.",
-        "If RESULTS contain conflicting information, prefer the result that directly matches the resolved date and question.",
-        "If the conflict cannot be resolved, say the results conflict.",
-        "Cite the source URL after each factual claim.",
-        "If the answer is not directly supported by RESULTS, say the results are not enough.",
-        PROMPT_RULE,
-        "",
-        PROMPT_RULE,
-        "RESULTS",
-        PROMPT_RULE,
-        "",
-    ]
-
-    for ordinal, result in enumerate(results, start=1):
-        relevant_text = _format_relevant_text(result.get("ranked_chunks") or [])
-        lines.extend(
-            [
-                PROMPT_RULE,
-                f"RESULT {ordinal}",
-                PROMPT_RULE,
-                f"TITLE {ordinal}",
-                FIELD_RULE,
-                str(result["title"]).strip(),
-                FIELD_RULE,
-                f"URL {ordinal}",
-                FIELD_RULE,
-                str(result["url"]).strip(),
-                FIELD_RULE,
-                f"SEARCH PREVIEW {ordinal}",
-                FIELD_RULE,
-                str(result.get("snippet") or "").strip(),
-                FIELD_RULE,
-            ]
-        )
-        if relevant_text:
-            lines.extend(
-                [
-                    f"RELEVANT TEXT {ordinal}",
-                    FIELD_RULE,
-                    relevant_text,
-                    FIELD_RULE,
-                ]
-            )
-        lines.append("")
-
-    lines.extend(
-        [
-            PROMPT_RULE,
-            "QUESTION",
-            PROMPT_RULE,
-            clean_question,
-            PROMPT_RULE,
-            "",
-            "TODAY",
-            PROMPT_RULE,
-            today_text,
-            PROMPT_RULE,
-            "",
-            PROMPT_RULE,
-            "SEARCH-GROUNDED ANSWER PROMPT",
-            PROMPT_RULE,
-        ]
-    )
-    return "\n".join(lines).strip()
-
-
 async def _rank(
     *,
     query: str,
@@ -427,7 +313,7 @@ def finish(status: str, answer: str, crawl_errors: Sequence[str]) -> AgenticResu
             except SearchBackendError as exc:
                 _agentic_log(f"search backend error: {exc}")
                 await emit("search_backend_error", error=str(exc))
-                prompt = _format_results_prompt(question=query, results=[])
+                prompt = format_search_grounded_prompt(question=query, results=[])
                 return finish("search_backend_error", prompt, [])
             results = [result for result in raw_results if _is_http_url(result.url)]
             results = filter_blocked_search_results(results, blocked_domains or [])
@@ -436,7 +322,7 @@ def finish(status: str, answer: str, crawl_errors: Sequence[str]) -> AgenticResu
             await emit("search_results", results_count=len(results))
 
             if not results:
-                prompt = _format_results_prompt(question=query, results=[])
+                prompt = format_search_grounded_prompt(question=query, results=[])
                 return finish("no_search_results", prompt, [])
 
             tokenizer_name = (
@@ -593,13 +479,13 @@ async def crawl_result(search_doc: dict[str, Any]) -> dict[str, Any]:
                 chunks_in_prompt=len(ranked_chunk_pool),
                 crawl_errors_count=len(crawl_errors),
             )
-            prompt = _format_results_prompt(question=query, results=crawled_results)
+            prompt = format_search_grounded_prompt(question=query, results=crawled_results)
             await emit("done", results_count=len(crawled_results), crawl_errors_count=len(crawl_errors))
             _agentic_log(f"done results={len(crawled_results)} crawl_errors={len(crawl_errors)}")
             return finish("ok", prompt, crawl_errors)
     except TimeoutError:
         _agentic_log(f"timeout query={query!r} limit_s={pipeline_timeout_seconds}")
-        prompt = _format_results_prompt(question=query, results=[])
+        prompt = format_search_grounded_prompt(question=query, results=[])
         return finish("timeout", prompt, [])
 
 

servers/fastapi_server.py

@@ -11,7 +11,7 @@
 if str(_PROJECT_ROOT) not in sys.path:
     sys.path.insert(0, str(_PROJECT_ROOT))
 
-from fastapi import FastAPI
+from fastapi import FastAPI, HTTPException
 from pydantic import BaseModel, Field, HttpUrl
 
 from pipelines.agentic_research import agentic_run
@@ -23,7 +23,14 @@
     research_run_kwargs,
     research_tokenizer_name,
 )
+from services.scrape_service import (
+    DEFAULT_SCRAPE_MAX_TOKENS,
+    SCRAPE_ERROR_MAP,
+    ScrapeError,
+    scrape_url,
+)
 from services.site_crawl_service import crawl_search
+from services.url_safety_service import BlockedUrlError, InvalidUrlError
 from services.web_search_service import (
     filter_blocked_search_results,
     search,
@@ -76,6 +83,13 @@ class SiteCrawlRequest(BaseModel):
     encoding_name: str | None = None
 
 
+class ScrapeRequest(BaseModel):
+    url: HttpUrl
+    query: str = Field(..., min_length=1)
+    max_tokens: int = Field(default=DEFAULT_SCRAPE_MAX_TOKENS, ge=1, le=200_000)
+    include_metadata: bool = True
+
+
 class ResearchRequest(BaseModel):
     query: str = Field(..., min_length=1)
     search_top_k: int | None = Field(default=None, ge=1, le=50)
@@ -178,6 +192,56 @@ async def site_crawl_get(
     )
 
 
+def _raise_scrape_http_error(exc: Exception) -> None:
+    mapping = SCRAPE_ERROR_MAP.get(type(exc))
+    if mapping is None:
+        raise HTTPException(
+            status_code=500,
+            detail={"code": "internal_error", "message": "internal error"},
+        ) from exc
+    code, status_code = mapping
+    raise HTTPException(
+        status_code=status_code,
+        detail={"code": code, "message": str(exc)},
+    ) from exc
+
+
+@app.post("/scrape")
+async def scrape_endpoint(request: ScrapeRequest) -> dict[str, Any]:
+    config = load_research_config()
+    await _ensure_local_bundle_for_config(config)
+    tokenizer = research_tokenizer_name(config)
+    try:
+        result = await scrape_url(
+            str(request.url),
+            request.query,
+            max_tokens=request.max_tokens,
+            include_metadata=request.include_metadata,
+            config=config,
+            tokenizer_name=tokenizer,
+        )
+    except (InvalidUrlError, BlockedUrlError, ScrapeError) as exc:
+        _raise_scrape_http_error(exc)
+    return result.to_response(include_metadata=request.include_metadata)
+
+
+@app.get("/scrape")
+async def scrape_get(
+    url: HttpUrl,
+    query: str,
+    max_tokens: int = DEFAULT_SCRAPE_MAX_TOKENS,
+    include_metadata: bool = True,
+) -> dict[str, Any]:
+    return await scrape_endpoint(
+        ScrapeRequest(
+            url=url,
+            query=query,
+            max_tokens=max_tokens,
+            include_metadata=include_metadata,
+        )
+    )
+
+
 @app.post("/research")
 async def research_endpoint(request: ResearchRequest) -> dict[str, Any]:
     config = load_research_config()