diff --git a/qiskit-docs-mcp-server/README.md b/qiskit-docs-mcp-server/README.md index e358052..f8c7e1f 100644 --- a/qiskit-docs-mcp-server/README.md +++ b/qiskit-docs-mcp-server/README.md @@ -28,7 +28,7 @@ The server implements three tools for documentation access: | Tool | Description | Parameters | |------|-------------|------------| -| `search_docs_tool` | Search across the entire Qiskit documentation for relevant content | `query`: Search query string
`scope`: Search scope filter — `all` (default), `documentation`, `api`, `learning`, `tutorials` | +| `search_docs_tool` | Search the Qiskit documentation; returns short query-centered snippets by default so the response stays compact for repeated agent use | `query`: Search query string
`scope`: Search scope filter — `all` (default), `documentation`, `api`, `learning`, `tutorials`
`top_k`: Max results to return (default: 5, capped at 10)
`detail`: `snippet` (default, short excerpt per result) or `full` (full page body — prefer `get_page_tool` for a single page) | | `get_page_tool` | Fetch any Qiskit documentation page and return as markdown | `url`: Full URL or relative path (e.g., `guides/transpile`, `api/qiskit/circuit`)
`max_length`: Max characters to return (default: 20000, 0 for unlimited)
`offset`: Character offset for pagination (default: 0) | | `lookup_error_code_tool` | Look up a Qiskit/IBM Quantum error code | `code`: 4-digit error code (e.g., 1002, 7001, 8004) | @@ -209,13 +209,26 @@ asyncio.run(main()) ### Search Documentation ```python -# Search for transpiler information +# Search returns short snippets (not full pages) — cheap to call inside an agent result = await search_docs_tool("transpiler optimization") -print(f"Found {result['total_results']} results") +print(f"Showing {result['returned_results']} of {result['total_results']} matches") for item in result["results"]: print(f"- {item['title']}: {item['url']}") + print(f" {item['snippet']}") + +# Then fetch the full content of the one page you want +page = await get_page_tool(result["results"][0]["url"]) +print(page["documentation"]) ``` +> **Note (behavior change):** `search_docs_tool` now returns short, query-centered +> **snippets** by default and caps results to `top_k` (default 5). In default +> (`detail="snippet"`) mode each result carries a `snippet` field — not the full +> `text` body — and is limited to `id, url, title, pageTitle, module, section, +> snippet`. For full content, pass `detail="full"` or fetch the page with +> `get_page_tool`. `total_results` is the grand total of matches; `returned_results` +> is the count after the `top_k` cap. + ### Fetch a Documentation Page ```python diff --git a/qiskit-docs-mcp-server/src/qiskit_docs_mcp_server/constants.py b/qiskit-docs-mcp-server/src/qiskit_docs_mcp_server/constants.py index 635f411..f9ddf55 100644 --- a/qiskit-docs-mcp-server/src/qiskit_docs_mcp_server/constants.py +++ b/qiskit-docs-mcp-server/src/qiskit_docs_mcp_server/constants.py @@ -38,6 +38,27 @@ def _get_env_float(name: str, default: float) -> float: return default +def _get_env_int(name: str, default: int) -> int: + """ + Get environment variable as int with fallback to default. + + Args: + name: Environment variable name + default: Default value if not set or invalid + + Returns: + Int value from environment or default + """ + try: + value = os.getenv(name) + if value is None: + return default + return int(value) + except (ValueError, TypeError): + logger.warning(f"Invalid {name} value: {os.getenv(name)}, using default {default}") + return default + + # Qiskit documentation base URL (configurable via environment variable) QISKIT_DOCS_BASE = os.getenv("QISKIT_DOCS_BASE", "https://quantum.cloud.ibm.com/docs/") BASE_URL = os.getenv("QISKIT_SEARCH_BASE_URL", "https://quantum.cloud.ibm.com/") @@ -66,6 +87,23 @@ def _get_env_float(name: str, default: float) -> float: CACHE_TTL = _get_env_float("QISKIT_DOCS_CACHE_TTL", 3600.0) SEARCH_CACHE_TTL = _get_env_float("QISKIT_SEARCH_CACHE_TTL", 300.0) # 5 min default +# --------------------------------------------------------------------------- +# search_docs response budget +# --------------------------------------------------------------------------- +# search_docs is used inside LLM agent loops, where the whole conversation +# (including every past tool result) is re-sent to the model on every turn. +# Returning full page bodies for many results makes a single call cost +# ~12k tokens and stacks per call, overflowing the context window. To stay +# token-economical, search returns short query-centered snippets by default +# and caps the number of results; full page content is fetched separately via +# get_page. These knobs bound the default response to roughly ~2k tokens. +# Floored at sensible minimums so an out-of-range env override (e.g. a negative +# value) can't produce a negative list slice or snippet window at runtime. +DEFAULT_SEARCH_TOP_K = max(1, _get_env_int("QISKIT_DOCS_SEARCH_TOP_K", 5)) +MAX_SEARCH_TOP_K = max(1, _get_env_int("QISKIT_DOCS_MAX_SEARCH_TOP_K", 10)) +SNIPPET_MAX_CHARS = max(50, _get_env_int("QISKIT_DOCS_SNIPPET_CHARS", 320)) +VALID_SEARCH_DETAIL = ("snippet", "full") + # --------------------------------------------------------------------------- # Fallback lists — used when sitemap discovery is unavailable # --------------------------------------------------------------------------- diff --git a/qiskit-docs-mcp-server/src/qiskit_docs_mcp_server/data_fetcher.py b/qiskit-docs-mcp-server/src/qiskit_docs_mcp_server/data_fetcher.py index 9f1c72f..d8ba69a 100644 --- a/qiskit-docs-mcp-server/src/qiskit_docs_mcp_server/data_fetcher.py +++ b/qiskit-docs-mcp-server/src/qiskit_docs_mcp_server/data_fetcher.py @@ -27,9 +27,13 @@ AVAILABLE_MODULES, AVAILABLE_TUTORIALS, BASE_URL, + DEFAULT_SEARCH_TOP_K, ERROR_CODE_CATEGORIES, + MAX_SEARCH_TOP_K, QISKIT_DOCS_BASE, SEARCH_PATH, + SNIPPET_MAX_CHARS, + VALID_SEARCH_DETAIL, ) from qiskit_docs_mcp_server.html_processing import ( _strip_html_tags, @@ -195,17 +199,178 @@ async def get_page_docs(url: str, max_length: int = 20000, offset: int = 0) -> d _VALID_SCOPES = {"all", "documentation", "api", "learning", "tutorials"} +# Metadata fields kept from each upstream search result. The full page body +# (the upstream ``text`` field) is deliberately excluded and replaced with a +# short ``snippet``; whitelisting also keeps the response small and stable if +# the upstream API later adds new (potentially large) fields. +_RESULT_META_FIELDS = ("id", "url", "title", "pageTitle", "module", "section") -async def search_qiskit_docs(query: str, scope: str = "all") -> dict[str, Any]: +# Words this short are ignored when locating the query match in a body, so that +# stop-word-like fragments don't anchor the snippet on a noisy position. +_MIN_TERM_LEN = 2 + +_ELLIPSIS = "…" # single-character "..." + + +def _densest_window_center(positions: list[int], width: int) -> int: + """Return the center of the densest cluster of match positions. + + Given sorted character offsets where query terms occur, find the window of + ``width`` characters that contains the most matches and return the midpoint + of that cluster (a good place to center a snippet). Returns -1 if there are + no positions. + """ + if not positions: + return -1 + best_count = 0 + best_center = positions[0] + left = 0 + for right, pos in enumerate(positions): + while pos - positions[left] > width: + left += 1 + count = right - left + 1 + if count > best_count: + best_count = count + best_center = (positions[left] + pos) // 2 + return best_center + + +def _trim_partial_words(window: str, *, head: bool, tail: bool) -> str: + """Trim partial words at clipped edges and add ellipsis markers. + + Args: + window: The raw character window sliced out of the body. + head: True if the window was clipped at the start (text precedes it). + tail: True if the window was clipped at the end (text follows it). + """ + if head: + # Drop a leading partial word (text before the first space). + space = window.find(" ") + if space != -1: + window = window[space + 1 :] + if tail: + # Drop a trailing partial word (text after the last space). + space = window.rfind(" ") + if space != -1: + window = window[:space] + window = window.strip() + if head: + window = f"{_ELLIPSIS} {window}" + if tail: + window = f"{window} {_ELLIPSIS}" + return window + + +def _make_snippet(text: str, query: str, max_chars: int = SNIPPET_MAX_CHARS) -> str: + """Build a compact, query-centered excerpt from a documentation body. + + Whitespace is collapsed so the character budget reflects real content. When + the query (or its terms) appears in the body, the snippet is a window + centered on the densest cluster of matches; otherwise it falls back to the + start of the body. Partial words at clipped edges are trimmed and marked + with an ellipsis. + + Args: + text: Plain-text body (HTML already stripped). + query: The search query, used to locate the most relevant window. + max_chars: Maximum length of the returned snippet (excluding ellipses). + + Returns: + A snippet no longer than roughly ``max_chars`` characters. + """ + if not text: + return "" + + normalized = " ".join(text.split()) + if len(normalized) <= max_chars: + return normalized + + haystack = normalized.lower() + + # Prefer the full query phrase; fall back to the densest cluster of terms. + # Guard against an empty phrase: ``"".find`` is 0, which would otherwise + # short-circuit to a head-anchored window and skip term clustering. + phrase = query.strip().lower() + anchor = haystack.find(phrase) if phrase else -1 + if anchor != -1: + # Center on the phrase's middle so a long phrase isn't clipped. + anchor += len(phrase) // 2 + else: + terms = {t for t in re.findall(r"\w+", query.lower()) if len(t) >= _MIN_TERM_LEN} + positions = sorted( + match.start() for term in terms for match in re.finditer(re.escape(term), haystack) + ) + anchor = _densest_window_center(positions, max_chars) + + if anchor == -1: + # No query term present (e.g. a semantic match) — use the body's head. + return _trim_partial_words(normalized[:max_chars], head=False, tail=True) + + # Center a window of max_chars on the anchor, clamped to the body bounds. + start = max(0, anchor - max_chars // 2) + end = min(len(normalized), start + max_chars) + start = max(0, end - max_chars) # pull the start back if we hit the tail + return _trim_partial_words(normalized[start:end], head=start > 0, tail=end < len(normalized)) + + +def _normalize_result_url(url_val: Any, base: str) -> Any: + """Resolve a relative search-result URL to a full documentation URL.""" + if not url_val or not isinstance(url_val, str): + return url_val + parsed = urlparse(url_val) + if not parsed.scheme and not parsed.netloc: + return f"{base}/{url_val.lstrip('/')}" + return url_val + + +def _build_result_entry(item: dict[str, Any], query: str, detail: str, base: str) -> dict[str, Any]: + """Build a single search-result entry with a snippet (or full text).""" + entry: dict[str, Any] = { + key: item[key] for key in _RESULT_META_FIELDS if item.get(key) is not None + } + if isinstance(entry.get("title"), str): + entry["title"] = _strip_html_tags(entry["title"]) + if "url" in entry: + entry["url"] = _normalize_result_url(entry["url"], base) + + raw_text = item.get("text") + clean_text = _strip_html_tags(raw_text) if isinstance(raw_text, str) else "" + if detail == "full": + entry["text"] = clean_text + else: + entry["snippet"] = _make_snippet(clean_text, query) + return entry + + +async def search_qiskit_docs( + query: str, + scope: str = "all", + top_k: int = DEFAULT_SEARCH_TOP_K, + detail: str = "snippet", +) -> dict[str, Any]: """Search Qiskit documentation for relevant results. + Returns concise, query-centered snippets by default so the response stays + small enough for repeated use inside an LLM agent loop. Use ``get_page_docs`` + with a result's ``url`` to read a page in full. + Args: query: Search query string scope: Search scope filter. Valid values (case-sensitive): 'all', 'documentation', 'api', 'learning', 'tutorials' + top_k: Maximum number of results to return (clamped to + [1, MAX_SEARCH_TOP_K]). Defaults to DEFAULT_SEARCH_TOP_K. + detail: 'snippet' (default) returns a short excerpt per result; + 'full' returns each result's full page body (heavier — prefer + get_page for full content). Returns: - Search results with matching entries, total count, and metadata + Search results with matching entries, counts, and metadata. + 'total_results' is the grand total of matches found; 'returned_results' + is the (possibly smaller) number included after the top_k cap, with a + 'truncated' flag. Each entry carries 'id', 'url', 'title', 'pageTitle', + 'module', 'section' (when available) and either a 'snippet' or, for + detail='full', a 'text' body. """ query = query.strip() if not query: @@ -220,6 +385,19 @@ async def search_qiskit_docs(query: str, scope: str = "all") -> dict[str, Any]: f"Invalid scope '{scope}'. Valid values: {', '.join(sorted(_VALID_SCOPES))}." ), } + if detail not in VALID_SEARCH_DETAIL: + return { + "status": "error", + "message": ( + f"Invalid detail '{detail}'. Valid values: {', '.join(VALID_SEARCH_DETAIL)}." + ), + } + + # Clamp top_k: non-positive falls back to the default, and an upper bound + # guards against a single call ballooning the response. The final max(1, ...) + # keeps the slice valid even if MAX_SEARCH_TOP_K were misconfigured low. + effective_top_k = top_k if top_k > 0 else DEFAULT_SEARCH_TOP_K + effective_top_k = max(1, min(effective_top_k, MAX_SEARCH_TOP_K)) url = f"{BASE_URL}{SEARCH_PATH}?query={quote(query)}&module={quote(scope)}" logger.info("Searching docs for '%s' in scope '%s'", query, scope) @@ -236,29 +414,34 @@ async def search_qiskit_docs(query: str, scope: str = "all") -> dict[str, Any]: }, } - # Strip HTML tags from search result fields (shallow copy to avoid mutating cached data) - cleaned = [] + # `total_results` keeps its pre-PR meaning: the grand total of matches found. + # `returned_results` is the (possibly smaller) count actually included here. + total_results = len(results) base = QISKIT_DOCS_BASE.rstrip("/") - for item in results: - entry = dict(item) - if "title" in entry: - entry["title"] = _strip_html_tags(entry["title"]) - if "text" in entry: - entry["text"] = _strip_html_tags(entry["text"]) - # Normalize URL to full URL if relative - url_val = entry.get("url") - if url_val: - parsed = urlparse(url_val) - if not parsed.scheme and not parsed.netloc: - entry["url"] = f"{base}/{url_val.lstrip('/')}" - cleaned.append(entry) + selected = results[:effective_top_k] + cleaned = [_build_result_entry(item, query, detail, base) for item in selected] + truncated = total_results > len(cleaned) + + if detail == "snippet": + note = "Showing snippets; call get_page_tool with a result's url for full page content." + else: + note = "Showing full page bodies; for a single page, get_page_tool is more economical." + if truncated: + note = ( + f"Showing top {len(cleaned)} of {total_results} matches. " + "Refine the query for fewer, more relevant results. " + note + ) return { "status": "success", "query": query, "scope": scope, + "detail": detail, "results": cleaned, - "total_results": len(cleaned), + "total_results": total_results, + "returned_results": len(cleaned), + "truncated": truncated, + "note": note, "metadata": { "url": url, "timestamp": datetime.now(timezone.utc).isoformat(), diff --git a/qiskit-docs-mcp-server/src/qiskit_docs_mcp_server/server.py b/qiskit-docs-mcp-server/src/qiskit_docs_mcp_server/server.py index b888905..94c75dd 100644 --- a/qiskit-docs-mcp-server/src/qiskit_docs_mcp_server/server.py +++ b/qiskit-docs-mcp-server/src/qiskit_docs_mcp_server/server.py @@ -25,7 +25,7 @@ import httpx from fastmcp import FastMCP -from qiskit_docs_mcp_server.constants import HTTP_TIMEOUT +from qiskit_docs_mcp_server.constants import DEFAULT_SEARCH_TOP_K, HTTP_TIMEOUT from qiskit_docs_mcp_server.data_fetcher import ( get_list_of_addons, get_list_of_api_packages, @@ -64,9 +64,10 @@ async def lifespan(server: FastMCP) -> AsyncIterator[None]: This server provides access to the Qiskit and IBM Quantum documentation. Recommended workflow: -1. Use search_docs_tool to find relevant pages. Specific queries yield \ -better results than broad ones. -2. Use get_page_tool to fetch the full content of pages found by search. \ +1. Use search_docs_tool to find relevant pages. It returns short snippets \ +(not full pages), so it is cheap to call. Specific queries yield better \ +results than broad ones. +2. Use get_page_tool to fetch the full content of a page found by search. \ For large pages, use the offset parameter to paginate through content. 3. Browse qiskit-docs:// resources (modules, addons, guides, error-codes) \ to discover what documentation is available. @@ -88,12 +89,18 @@ async def lifespan(server: FastMCP) -> AsyncIterator[None]: @mcp.tool() -async def search_docs_tool(query: str, scope: str = "all") -> dict[str, Any]: +async def search_docs_tool( + query: str, + scope: str = "all", + top_k: int = DEFAULT_SEARCH_TOP_K, + detail: str = "snippet", +) -> dict[str, Any]: """Search across the entire Qiskit documentation for relevant content. - Use this as the primary entry point to discover documentation pages. - Returns ranked results with titles, URLs, sections, and text snippets. - Use get_page_tool to fetch the full content of any result URL. + Use this as the primary entry point to discover documentation pages. Returns + a small set of ranked results, each with a short query-centered snippet plus + its title, URL, and section. This keeps the response compact for repeated use + inside an agent. To read a page in full, pass a result's URL to get_page_tool. Args: query: Search query string (e.g., 'error mitigation', 'QuantumCircuit', @@ -104,12 +111,18 @@ async def search_docs_tool(query: str, scope: str = "all") -> dict[str, Any]: 'api' — API reference pages only 'learning' — Learning resources and tutorials 'tutorials' — Tutorial content only + top_k: Maximum number of results to return (default: 5, capped at 10). + detail: Per-result content level. 'snippet' (default) returns a short + excerpt; 'full' returns each result's full page body — heavier, so + prefer get_page_tool when you only need one page in full. Returns: - List of matching documentation entries with URLs, titles, sections, - and text snippets. Use the URLs with get_page_tool to fetch full content. + Matching documentation entries (id, url, title, pageTitle, module, + section, and a snippet) plus 'total_results' (grand total of matches), + 'returned_results' (count after the top_k cap), and a 'truncated' flag. + Use a result's URL with get_page_tool to fetch the full page content. """ - return await search_qiskit_docs(query, scope) + return await search_qiskit_docs(query, scope, top_k=top_k, detail=detail) @mcp.tool() diff --git a/qiskit-docs-mcp-server/tests/test_data_fetcher.py b/qiskit-docs-mcp-server/tests/test_data_fetcher.py index 78683ae..c576231 100644 --- a/qiskit-docs-mcp-server/tests/test_data_fetcher.py +++ b/qiskit-docs-mcp-server/tests/test_data_fetcher.py @@ -12,6 +12,7 @@ """Tests for data_fetcher module.""" +import json from unittest.mock import AsyncMock, MagicMock, patch import httpx @@ -23,11 +24,17 @@ AVAILABLE_MODULES, AVAILABLE_TUTORIALS, CACHE_TTL, + DEFAULT_SEARCH_TOP_K, HTTP_TIMEOUT, + MAX_SEARCH_TOP_K, SEARCH_CACHE_TTL, + SNIPPET_MAX_CHARS, _get_env_float, + _get_env_int, ) from qiskit_docs_mcp_server.data_fetcher import ( + _make_snippet, + _normalize_result_url, _resolve_url, _truncate_content, get_list_of_addons, @@ -576,7 +583,7 @@ async def test_search_fetch_fails(self, mock_fetch): @patch("qiskit_docs_mcp_server.data_fetcher.fetch_text_json") async def test_search_strips_html_tags(self, mock_fetch): - """Test that HTML tags are stripped from search results.""" + """Test that HTML tags are stripped from search result titles and snippets.""" mock_fetch.return_value = [ { "title": "Transpiler options", @@ -585,7 +592,9 @@ async def test_search_strips_html_tags(self, mock_fetch): ] result = await search_qiskit_docs("transpiler") assert result["results"][0]["title"] == "Transpiler options" - assert result["results"][0]["text"] == "Transpiler passes" + # Default detail is 'snippet'; short bodies pass through whole, HTML-free. + assert result["results"][0]["snippet"] == "Transpiler passes" + assert "text" not in result["results"][0] @patch("qiskit_docs_mcp_server.data_fetcher.fetch_text_json") async def test_search_uses_scope_param(self, mock_fetch): @@ -680,6 +689,262 @@ async def test_search_long_query_accepted(self, mock_fetch): assert result["status"] == "success" +def _make_search_payload(n: int, body_chars: int = 3500) -> list[dict]: + """Build a realistic upstream search payload of ``n`` full-body results.""" + body = ( + "The QuantumCircuit class represents a quantum circuit. " + "Use load to read a QASM3 program into a circuit. " + ("lorem ipsum " * 400) + )[:body_chars] + return [ + { + "id": f"qiskit_current_section-en-{i}", + "url": f"https://quantum.cloud.ibm.com/docs/api/qiskit/qasm3#load-{i}", + "pageTitle": "qasm3 (latest version)", + "module": "api", + "section": "Qiskit SDK", + "language": "en", + "title": f"load {i}", + "text": body, + "package": "qiskit", + } + for i in range(n) + ] + + +class TestMakeSnippet: + """Test the _make_snippet helper.""" + + def test_short_text_passthrough(self): + """Short bodies are returned whole (whitespace-normalized).""" + assert _make_snippet("A short body.", "body") == "A short body." + + def test_collapses_whitespace(self): + """Runs of whitespace/newlines are collapsed to single spaces.""" + assert _make_snippet("a\n\n b\t c", "a") == "a b c" + + def test_empty_text(self): + """Empty body yields an empty snippet.""" + assert _make_snippet("", "anything") == "" + + def test_long_text_is_capped(self): + """A long body is truncated to roughly max_chars (plus ellipses).""" + text = "alpha " * 2000 + snippet = _make_snippet(text, "alpha", max_chars=100) + # Allow a small margin for the ellipsis markers and word-boundary trim. + assert len(snippet) <= 100 + 8 + + def test_centers_on_query_match(self): + """The snippet window is centered on the query match deep in the body.""" + filler = "x" * 4000 + text = filler + " UNIQUEMATCHTOKEN " + filler + snippet = _make_snippet(text, "UNIQUEMATCHTOKEN", max_chars=200) + assert "UNIQUEMATCHTOKEN" in snippet + assert len(snippet) <= 200 + 8 + # Window is interior, so both edges are clipped with ellipses. + assert snippet.startswith("…") + assert snippet.endswith("…") + + def test_head_fallback_when_no_match(self): + """When no query term appears, the snippet falls back to the head.""" + text = "zzz " * 2000 + snippet = _make_snippet(text, "nonexistentterm", max_chars=100) + assert snippet.startswith("zzz") + assert snippet.endswith("…") + assert not snippet.startswith("…") + + def test_phrase_match_preferred(self): + """A contiguous phrase anchors the snippet even with scattered terms.""" + text = "load " + ("noise " * 500) + "load a QASM3 circuit here " + ("noise " * 500) + snippet = _make_snippet(text, "load a QASM3 circuit", max_chars=120) + assert "load a QASM3 circuit" in snippet + + def test_centers_on_densest_term_cluster(self): + """With no contiguous phrase, the window lands on the densest term cluster. + + 'alpha' appears alone early and 'beta' alone in the middle, but all three + query terms cluster together at the end (in a different order, so the exact + phrase never matches). The snippet must center on that dense cluster — this + exercises the two-pointer narrowing in _densest_window_center. + """ + text = "alpha " + ("x " * 600) + "beta " + ("y " * 600) + "gamma beta alpha tail" + snippet = _make_snippet(text, "alpha beta gamma", max_chars=80) + # 'gamma' only occurs in the end cluster, so its presence proves the window + # centered there rather than on the lone early 'alpha'. + assert "gamma" in snippet + assert len(snippet) <= 80 + 8 + + def test_blank_query_falls_back_to_head(self): + """A whitespace-only query must not anchor at offset 0 via "".find(). + + An empty phrase has no terms either, so the snippet should fall back to + the head of the body rather than short-circuiting on the empty match. + """ + text = "first part here " + ("filler " * 400) + snippet = _make_snippet(text, " ", max_chars=100) + assert snippet.startswith("first") + assert not snippet.startswith("…") + assert snippet.endswith("…") + + +class TestNormalizeResultUrl: + """Test the _normalize_result_url helper.""" + + def test_relative_url_resolved(self): + """A relative URL is resolved against the docs base.""" + out = _normalize_result_url("/docs/api/qiskit/circuit", "https://quantum.cloud.ibm.com") + assert out == "https://quantum.cloud.ibm.com/docs/api/qiskit/circuit" + + def test_absolute_url_passthrough(self): + """An absolute URL is returned unchanged.""" + url = "https://quantum.cloud.ibm.com/docs/api/qiskit/circuit" + assert _normalize_result_url(url, "https://quantum.cloud.ibm.com") == url + + def test_empty_or_non_string_passthrough(self): + """Falsy or non-string URL values are returned as-is (defensive guard).""" + assert _normalize_result_url("", "https://base") == "" + assert _normalize_result_url(None, "https://base") is None + assert _normalize_result_url(123, "https://base") == 123 + + +class TestSearchSnippetsAndPaging: + """Test snippet/full detail modes, top_k, and response-size control.""" + + @patch("qiskit_docs_mcp_server.data_fetcher.fetch_text_json") + async def test_default_caps_results_to_top_k(self, mock_fetch): + """Default search returns at most DEFAULT_SEARCH_TOP_K results.""" + mock_fetch.return_value = _make_search_payload(20) + result = await search_qiskit_docs("load QASM3 circuit", scope="api") + assert result["status"] == "success" + assert len(result["results"]) == DEFAULT_SEARCH_TOP_K + # total_results is the grand total of matches; returned_results is capped. + assert result["total_results"] == 20 + assert result["returned_results"] == DEFAULT_SEARCH_TOP_K + assert result["truncated"] is True + + @patch("qiskit_docs_mcp_server.data_fetcher.fetch_text_json") + async def test_results_carry_snippet_not_full_text(self, mock_fetch): + """Each default result has a bounded snippet and no full text body.""" + mock_fetch.return_value = _make_search_payload(3) + result = await search_qiskit_docs("load QASM3 circuit", scope="api") + for entry in result["results"]: + assert "snippet" in entry + assert "text" not in entry + assert len(entry["snippet"]) <= SNIPPET_MAX_CHARS + 8 + # Navigation metadata is preserved. + assert entry["url"].startswith("https://") + assert "id" in entry + assert "title" in entry + + @patch("qiskit_docs_mcp_server.data_fetcher.fetch_text_json") + async def test_default_response_under_size_cap(self, mock_fetch): + """Regression: a large upstream payload yields a compact response. + + Reproduces the reported case (full bodies for many results, ~43 KB) + and asserts the default response stays well under ~2000 tokens. + """ + mock_fetch.return_value = _make_search_payload(15) + result = await search_qiskit_docs("load QASM3 circuit", scope="api") + serialized = json.dumps(result) + assert len(serialized) < 8000 # ~2000 tokens at ~4 chars/token + + @patch("qiskit_docs_mcp_server.data_fetcher.fetch_text_json") + async def test_custom_top_k(self, mock_fetch): + """An explicit top_k limits the number of results.""" + mock_fetch.return_value = _make_search_payload(20) + result = await search_qiskit_docs("circuit", top_k=2) + assert len(result["results"]) == 2 + + @patch("qiskit_docs_mcp_server.data_fetcher.fetch_text_json") + async def test_top_k_clamped_to_max(self, mock_fetch): + """top_k above the ceiling is clamped to MAX_SEARCH_TOP_K.""" + mock_fetch.return_value = _make_search_payload(50) + result = await search_qiskit_docs("circuit", top_k=999) + assert len(result["results"]) == MAX_SEARCH_TOP_K + + @patch("qiskit_docs_mcp_server.data_fetcher.fetch_text_json") + async def test_non_positive_top_k_uses_default(self, mock_fetch): + """A non-positive top_k falls back to the default.""" + mock_fetch.return_value = _make_search_payload(20) + result = await search_qiskit_docs("circuit", top_k=0) + assert len(result["results"]) == DEFAULT_SEARCH_TOP_K + + @patch("qiskit_docs_mcp_server.data_fetcher.MAX_SEARCH_TOP_K", -1) + @patch("qiskit_docs_mcp_server.data_fetcher.fetch_text_json") + async def test_misconfigured_max_does_not_negative_slice(self, mock_fetch): + """A misconfigured (negative) cap must not silently drop the last result. + + Guards the effective_top_k = max(1, min(...)) floor: with a negative cap, + a naive min() would yield results[:-1]; we must still return >= 1 result + and keep returned_results consistent with the list length. + """ + mock_fetch.return_value = _make_search_payload(5) + result = await search_qiskit_docs("circuit") + assert result["status"] == "success" + assert result["returned_results"] >= 1 + assert len(result["results"]) == result["returned_results"] + + @patch("qiskit_docs_mcp_server.data_fetcher.fetch_text_json") + async def test_not_truncated_when_few_results(self, mock_fetch): + """truncated is False when the API returns fewer than top_k results.""" + mock_fetch.return_value = _make_search_payload(2) + result = await search_qiskit_docs("circuit") + assert result["truncated"] is False + assert result["total_results"] == 2 + assert result["returned_results"] == 2 + + @patch("qiskit_docs_mcp_server.data_fetcher.fetch_text_json") + async def test_detail_full_returns_full_body(self, mock_fetch): + """detail='full' returns each result's full (HTML-stripped) text.""" + mock_fetch.return_value = _make_search_payload(2) + result = await search_qiskit_docs("circuit", detail="full") + assert result["detail"] == "full" + for entry in result["results"]: + assert "text" in entry + assert "snippet" not in entry + assert len(entry["text"]) > SNIPPET_MAX_CHARS + + @patch("qiskit_docs_mcp_server.data_fetcher.fetch_text_json") + async def test_detail_full_strips_html(self, mock_fetch): + """detail='full' still strips HTML tags from the body.""" + mock_fetch.return_value = [ + {"url": "/docs/api/qiskit/circuit", "text": "Circuit module body"}, + ] + result = await search_qiskit_docs("circuit", detail="full") + assert result["results"][0]["text"] == "Circuit module body" + + async def test_invalid_detail_returns_error(self): + """An invalid detail value returns an error without hitting the API.""" + result = await search_qiskit_docs("circuit", detail="verbose") + assert result["status"] == "error" + assert "Invalid detail" in result["message"] + + @patch("qiskit_docs_mcp_server.data_fetcher.fetch_text_json") + async def test_note_present_in_snippet_mode(self, mock_fetch): + """The response includes a note nudging toward get_page_tool.""" + mock_fetch.return_value = _make_search_payload(1) + result = await search_qiskit_docs("circuit") + assert "get_page_tool" in result["note"] + + @patch("qiskit_docs_mcp_server.data_fetcher.fetch_text_json") + async def test_extra_upstream_fields_are_dropped(self, mock_fetch): + """Unknown/large upstream fields are not echoed back.""" + mock_fetch.return_value = [ + { + "url": "/docs/api/qiskit/circuit", + "title": "circuit", + "text": "body", + "language": "en", + "package": "qiskit", + "huge_unknown_field": "x" * 10000, + }, + ] + result = await search_qiskit_docs("circuit") + entry = result["results"][0] + assert "huge_unknown_field" not in entry + assert "language" not in entry + assert "package" not in entry + + class TestLookupErrorCode: """Test lookup_error_code function.""" @@ -1297,6 +1562,44 @@ def test_search_cache_ttl_default(self): """Test that SEARCH_CACHE_TTL defaults to 300.0 (5 minutes).""" assert SEARCH_CACHE_TTL == 300.0 + def test_get_env_int_valid(self): + """Test _get_env_int parses a valid integer.""" + import os + + original = os.environ.get("TEST_ENV_INT") + try: + os.environ["TEST_ENV_INT"] = "7" + assert _get_env_int("TEST_ENV_INT", 3) == 7 + finally: + if original is not None: + os.environ["TEST_ENV_INT"] = original + else: + os.environ.pop("TEST_ENV_INT", None) + + def test_get_env_int_invalid_returns_default(self): + """Test _get_env_int returns default on a non-integer value.""" + import os + + original = os.environ.get("TEST_ENV_INT_BAD") + try: + os.environ["TEST_ENV_INT_BAD"] = "not_an_int" + assert _get_env_int("TEST_ENV_INT_BAD", 3) == 3 + finally: + if original is not None: + os.environ["TEST_ENV_INT_BAD"] = original + else: + os.environ.pop("TEST_ENV_INT_BAD", None) + + def test_get_env_int_missing_returns_default(self): + """Test _get_env_int returns default when the var is unset.""" + assert _get_env_int("NONEXISTENT_INT_VAR_98765", 42) == 42 + + def test_search_budget_constants_sane(self): + """Test that the search-budget constants have sensible defaults.""" + assert DEFAULT_SEARCH_TOP_K >= 1 + assert MAX_SEARCH_TOP_K >= DEFAULT_SEARCH_TOP_K + assert SNIPPET_MAX_CHARS >= 100 + @patch("qiskit_docs_mcp_server.http.httpx.AsyncClient") def test_fetch_text_uses_http_timeout(self, mock_client_class): """Test that _get_http_client creates client with HTTP_TIMEOUT."""