diff --git a/docs/advanced/configuration.mdx b/docs/advanced/configuration.mdx index ab067bf4..105fb727 100644 --- a/docs/advanced/configuration.mdx +++ b/docs/advanced/configuration.mdx @@ -81,7 +81,7 @@ affecting the agents that do the actual testing. - API key for Exa. Enables real-time web search through the Exa `/search` endpoint. This is the preferred web search provider. + API key for Exa. Enables real-time web search through the Exa `/search` endpoint. Exa also powers the `web_get_contents` tool, which fetches the full text of a page through the Exa `/contents` endpoint. This is the preferred web search provider. @@ -93,7 +93,11 @@ affecting the agents that do the actual testing. - Number of Exa results to return, from `1` to `100`. Each result includes short query-relevant highlights. This setting applies only to the Exa provider. + Number of Exa results to return, from `1` to `100`. This setting applies only to the Exa provider. + + + + Content that each Exa result includes: `summary` or `highlights`. With `summary`, each result includes a short security-focused summary of the page. With `highlights`, each result includes verbatim query-relevant excerpts. To read a full page, the agent calls `web_get_contents` with the result URL. This setting applies only to the Exa provider. diff --git a/strix/agents/factory.py b/strix/agents/factory.py index c55320f1..b2fcbf08 100644 --- a/strix/agents/factory.py +++ b/strix/agents/factory.py @@ -69,7 +69,7 @@ from strix.tools.todo.tools import ( mark_todo_pending, update_todo, ) -from strix.tools.web_search.tool import web_search +from strix.tools.web_search.tool import web_get_contents, web_search if TYPE_CHECKING: @@ -579,6 +579,7 @@ _BASE_TOOLS: tuple[Tool, ...] = ( save_threat_model, amend_threat_model, web_search, + web_get_contents, create_vulnerability_report, create_dependency_report, update_vulnerability_report, diff --git a/strix/config/settings.py b/strix/config/settings.py index 9309ac39..cc6847bc 100644 --- a/strix/config/settings.py +++ b/strix/config/settings.py @@ -122,6 +122,7 @@ class TelemetrySettings(BaseSettings): WebSearchProvider = Literal["auto", "perplexity", "exa"] ExaSearchType = Literal["auto", "fast", "instant", "deep-lite", "deep", "deep-reasoning"] +ExaContentMode = Literal["summary", "highlights"] class IntegrationSettings(BaseSettings): @@ -151,6 +152,10 @@ class IntegrationSettings(BaseSettings): le=100, alias="STRIX_EXA_NUM_RESULTS", ) + exa_content_mode: ExaContentMode = Field( + default="summary", + alias="STRIX_EXA_CONTENT_MODE", + ) postman_api_key: str | None = Field( default=None, alias="POSTMAN_API_KEY", diff --git a/strix/tools/web_search/tool.py b/strix/tools/web_search/tool.py index 86878c78..7dc33e15 100644 --- a/strix/tools/web_search/tool.py +++ b/strix/tools/web_search/tool.py @@ -1,11 +1,11 @@ -"""``web_search`` — security-focused web search (Exa or Perplexity).""" +"""Security-focused web research tools (Exa or Perplexity).""" from __future__ import annotations import asyncio import json import logging -from typing import Any, cast +from typing import TYPE_CHECKING, Any, cast import requests from agents import RunContextWrapper, function_tool @@ -13,6 +13,10 @@ from agents import RunContextWrapper, function_tool from strix.config import load_settings +if TYPE_CHECKING: + from collections.abc import Callable + + logger = logging.getLogger(__name__) @@ -57,6 +61,20 @@ def _perplexity_content(api_key: str, query: str) -> str: _EXA_HIGHLIGHT_MAX_CHARS = 750 +_EXA_PAGE_MAX_CHARS = 20000 +_EXA_MAX_CONTENT_URLS = 10 +_EXA_SUMMARY_PROMPT = ( + "Summarize this page for a penetration tester. Keep concrete technical detail: " + "affected products and exact versions, CVE and CWE identifiers, CVSS scores, " + "exploitation preconditions, payloads or commands, and mitigations. " + "Leave out marketing copy and navigation text." +) + + +def _exa_search_contents(content_mode: str) -> dict[str, Any]: + if content_mode == "highlights": + return {"highlights": {"maxCharacters": _EXA_HIGHLIGHT_MAX_CHARS}} + return {"summary": {"query": _EXA_SUMMARY_PROMPT}} def _exa_result_block(result: dict[str, Any]) -> str | None: @@ -65,6 +83,9 @@ def _exa_result_block(result: dict[str, Any]) -> str | None: return None title = str(result.get("title") or result_url) parts = [f"### {title}\n{result_url}"] + summary = str(result.get("summary") or "").strip() + if summary: + parts.append(summary) highlights = result.get("highlights") or [] if isinstance(highlights, list): excerpts = [str(item).strip() for item in highlights if str(item).strip()] @@ -73,31 +94,70 @@ def _exa_result_block(result: dict[str, Any]) -> str | None: return "\n".join(parts) -def _exa_content(api_key: str, query: str, search_type: str, num_results: int) -> str: - url = "https://api.exa.ai/search" - headers = {"x-api-key": api_key, "Content-Type": "application/json"} - payload = { - "query": f"{_SYSTEM_PROMPT}\n\n{query}", - "type": search_type, - "numResults": num_results, - "contents": {"highlights": {"maxCharacters": _EXA_HIGHLIGHT_MAX_CHARS}}, - } - with requests.post(url, headers=headers, json=payload, timeout=300) as response: - response.raise_for_status() - body: dict[str, Any] = response.json() - results: list[Any] = body.get("results") or [] +def _exa_page_block(result: dict[str, Any]) -> str | None: + result_url = str(result.get("url") or result.get("id") or "") + text = str(result.get("text") or "").strip() + if not result_url or not text: + return None + if len(text) > _EXA_PAGE_MAX_CHARS: + text = f"{text[:_EXA_PAGE_MAX_CHARS]}\n[truncated at {_EXA_PAGE_MAX_CHARS} characters]" + title = str(result.get("title") or result_url) + return f"### {title}\n{result_url}\n\n{text}" + + +def _exa_blocks( + results: list[Any], + render: Callable[[dict[str, Any]], str | None], +) -> list[str]: blocks: list[str] = [] for result in results: if not isinstance(result, dict): continue - block = _exa_result_block(cast("dict[str, Any]", result)) + block = render(cast("dict[str, Any]", result)) if block: blocks.append(block) + return blocks + + +def _exa_post(api_key: str, endpoint: str, payload: dict[str, Any]) -> dict[str, Any]: + headers = {"x-api-key": api_key, "Content-Type": "application/json"} + with requests.post(endpoint, headers=headers, json=payload, timeout=300) as response: + response.raise_for_status() + body: dict[str, Any] = response.json() + return body + + +def _exa_content( + api_key: str, + query: str, + search_type: str, + num_results: int, + content_mode: str, +) -> str: + body = _exa_post( + api_key, + "https://api.exa.ai/search", + { + "query": f"{_SYSTEM_PROMPT}\n\n{query}", + "type": search_type, + "numResults": num_results, + "contents": _exa_search_contents(content_mode), + }, + ) + blocks = _exa_blocks(body.get("results") or [], _exa_result_block) if not blocks: raise ValueError("Exa response has no results") return "\n\n".join(blocks) +def _exa_page_text(api_key: str, urls: list[str]) -> str: + body = _exa_post(api_key, "https://api.exa.ai/contents", {"urls": urls, "text": True}) + blocks = _exa_blocks(body.get("results") or [], _exa_page_block) + if not blocks: + raise ValueError("Exa returned no page contents") + return "\n\n".join(blocks) + + def _resolve_provider( # noqa: PLR0911 - each provider/missing-key case needs its own return integrations: Any, ) -> tuple[str, str] | dict[str, Any]: @@ -133,7 +193,35 @@ def _not_configured_error(missing: str) -> dict[str, Any]: } -def _do_search(query: str) -> dict[str, Any]: # noqa: PLR0911 - each error class needs its own sanitized return +def _guarded_call( # noqa: PLR0911 - each error class needs its own sanitized return + tool: str, + rejected_hint: str, + fetch: Callable[[], str], +) -> str | dict[str, Any]: + """Run a provider call and translate any failure into a sanitized error dict.""" + try: + return fetch() + except requests.exceptions.Timeout: + logger.warning("%s timed out", tool) + return {"success": False, "error": f"{tool} timed out. Try again or narrow the request"} + except requests.exceptions.HTTPError as exc: + status = exc.response.status_code if exc.response is not None else None + logger.exception("%s HTTP error status=%s", tool, status) + if status is not None and 400 <= status < 500: + return {"success": False, "error": rejected_hint} + return {"success": False, "error": f"{tool} service is unavailable. Try again later"} + except requests.exceptions.RequestException: + logger.exception("%s network error", tool) + return {"success": False, "error": f"{tool} network error. Try again later"} + except (KeyError, IndexError, ValueError): + logger.exception("%s response shape unexpected", tool) + return {"success": False, "error": f"{tool} returned an unexpected response. Try again"} + except Exception: + logger.exception("%s failed", tool) + return {"success": False, "error": f"{tool} failed unexpectedly"} + + +def _do_search(query: str) -> dict[str, Any]: if not query or not query.strip(): return {"success": False, "error": "Query cannot be empty"} @@ -144,63 +232,75 @@ def _do_search(query: str) -> dict[str, Any]: # noqa: PLR0911 - each error clas provider, api_key = resolved logger.info("web_search provider=%s query (len=%d): %s", provider, len(query), query[:120]) - try: + def fetch() -> str: if provider == "exa": - content = _exa_content( + return _exa_content( api_key, query, integrations.exa_search_type, integrations.exa_num_results, + integrations.exa_content_mode, ) - else: - content = _perplexity_content(api_key, query) - except requests.exceptions.Timeout: - logger.warning("web_search timed out") + return _perplexity_content(api_key, query) + + outcome = _guarded_call( + "Web search", + ( + "Web search rejected the query. Refine it " + "(more specific, shorter, no unusual characters) and retry" + ), + fetch, + ) + if isinstance(outcome, dict): + return outcome + return { + "success": True, + "query": query, + "provider": provider, + "content": outcome, + } + + +def _do_get_contents(urls: list[str]) -> dict[str, Any]: + cleaned = [url.strip() for url in urls if url and url.strip()] + if not cleaned: + return {"success": False, "error": "Provide at least one URL"} + if len(cleaned) > _EXA_MAX_CONTENT_URLS: return { "success": False, - "error": "Web search timed out. Try again or shorten the query", + "error": f"Too many URLs. Pass at most {_EXA_MAX_CONTENT_URLS} per call", } - except requests.exceptions.HTTPError as exc: - status = exc.response.status_code if exc.response is not None else None - logger.exception("web_search HTTP error status=%s", status) - if status is not None and 400 <= status < 500: - return { - "success": False, - "error": ( - "Web search rejected the query. Refine it " - "(more specific, shorter, no unusual characters) and retry" - ), - } + + integrations = load_settings().integrations + api_key = integrations.exa_api_key + if not api_key: + return _not_configured_error("EXA_API_KEY") + if integrations.web_search_provider == "perplexity": + logger.warning("web_get_contents invoked while the provider is pinned to Perplexity") return { "success": False, - "error": "Web search service is unavailable. Try again later", - } - except requests.exceptions.RequestException: - logger.exception("web_search network error") - return { - "success": False, - "error": "Web search network error. Try again later", - } - except (KeyError, IndexError, ValueError): - logger.exception("web_search response shape unexpected") - return { - "success": False, - "error": "Web search returned an unexpected response. Try again", - } - except Exception: - logger.exception("web_search failed") - return { - "success": False, - "error": "Web search failed unexpectedly", - } - else: - return { - "success": True, - "query": query, - "provider": provider, - "content": content, + "error": ( + "Page fetching needs the Exa provider " + "(operator pinned STRIX_WEB_SEARCH_PROVIDER to perplexity). " + "Use web_search instead" + ), } + logger.info("web_get_contents urls=%d", len(cleaned)) + outcome = _guarded_call( + "Page fetch", + "Page fetch was rejected. Check the URLs are complete, public, and correctly formed", + lambda: _exa_page_text(api_key, cleaned), + ) + if isinstance(outcome, dict): + return outcome + return { + "success": True, + "urls": cleaned, + "provider": "exa", + "content": outcome, + } + @function_tool(timeout=330) async def web_search(ctx: RunContextWrapper, query: str) -> str: @@ -234,6 +334,13 @@ async def web_search(ctx: RunContextWrapper, query: str) -> str: exploits, Kali-compatible tooling, and concrete code/command examples. + With the Exa provider you get a ranked list of results, each with a + title, URL, and a short security-focused summary (or verbatim + highlights when ``STRIX_EXA_CONTENT_MODE=highlights``). Read the + result you need, then call ``web_get_contents`` with its URL to pull + the full page text when a summary is not enough. With Perplexity you + get a single synthesized cited answer. + **Good example queries** (each is a full sentence, names a version/product, and asks one concrete thing): @@ -261,3 +368,28 @@ async def web_search(ctx: RunContextWrapper, query: str) -> str: """ result = await asyncio.to_thread(_do_search, query) return json.dumps(result, ensure_ascii=False, default=str) + + +@function_tool(timeout=330) +async def web_get_contents(ctx: RunContextWrapper, urls: list[str]) -> str: + """Fetch the full, cleaned text of specific web pages (Exa only). + + Use this as the drill-down step after ``web_search``: when a result's + summary or highlights are not enough, pass that result's URL here to + read the whole page. Good for reading a full advisory, a CVE writeup, + an exploit proof-of-concept, or vendor documentation end to end. + + Prefer ``web_search`` first to find the right pages, then fetch only + the few URLs worth reading in full — each page can be large, so avoid + fetching many pages you do not need. + + This tool needs the Exa provider (``EXA_API_KEY``). When the operator + pins the provider to Perplexity, it returns an error and you should + use ``web_search`` instead. + + Args: + urls: The page URLs to fetch, at most 10 per call. Use complete, + public URLs (for example the ones returned by ``web_search``). + """ + result = await asyncio.to_thread(_do_get_contents, urls) + return json.dumps(result, ensure_ascii=False, default=str) diff --git a/tests/test_web_search.py b/tests/test_web_search.py index ffaae6b5..f2e821d1 100644 --- a/tests/test_web_search.py +++ b/tests/test_web_search.py @@ -1,4 +1,4 @@ -"""Tests for web_search provider selection and the Exa backend.""" +"""Tests for web_search/web_get_contents provider selection and the Exa backend.""" from __future__ import annotations @@ -104,7 +104,7 @@ def test_exa_content_renders_results_with_highlights(monkeypatch: pytest.MonkeyP monkeypatch.setattr(requests, "post", fake_post) - content = tool._exa_content("ek", "OpenSSH 7.4 RCE?", "auto", 5) + content = tool._exa_content("ek", "OpenSSH 7.4 RCE?", "auto", 5, "highlights") assert captured["url"] == "https://api.exa.ai/search" assert captured["headers"]["x-api-key"] == "ek" @@ -112,13 +112,42 @@ def test_exa_content_renders_results_with_highlights(monkeypatch: pytest.MonkeyP assert captured["json"]["type"] == "auto" assert captured["json"]["numResults"] == 5 assert captured["json"]["contents"]["highlights"]["maxCharacters"] > 0 + assert "summary" not in captured["json"]["contents"] assert content == ( "### NVD entry\nhttps://nvd.example/cve\n> CVE-2024-0001 affects it.\n\n" "### https://blog.example/post\nhttps://blog.example/post" ) -def test_exa_content_renders_a_result_without_highlights( +def test_exa_content_defaults_to_summary_and_renders_it( + monkeypatch: pytest.MonkeyPatch, +) -> None: + captured: dict[str, Any] = {} + + def fake_post(_url: str, **kwargs: Any) -> _FakeResponse: + captured["json"] = kwargs["json"] + return _FakeResponse( + { + "results": [ + { + "url": "https://nvd.example/cve", + "title": "NVD entry", + "summary": " CVE-2024-0001 is a heap overflow. ", + }, + ], + } + ) + + monkeypatch.setattr(requests, "post", fake_post) + + content = tool._exa_content("ek", "q", "auto", 5, "summary") + + assert captured["json"]["contents"]["summary"]["query"] + assert "highlights" not in captured["json"]["contents"] + assert content == "### NVD entry\nhttps://nvd.example/cve\nCVE-2024-0001 is a heap overflow." + + +def test_exa_content_renders_a_result_without_contents( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr( @@ -128,7 +157,7 @@ def test_exa_content_renders_a_result_without_highlights( {"results": [{"url": "https://ex.example", "title": "Ex"}]} ), ) - assert tool._exa_content("ek", "q", "auto", 5) == "### Ex\nhttps://ex.example" + assert tool._exa_content("ek", "q", "auto", 5, "summary") == "### Ex\nhttps://ex.example" @pytest.mark.parametrize("body", [{}, {"results": None}, {"results": []}, {"results": ["x"]}]) @@ -137,7 +166,28 @@ def test_exa_content_rejects_empty_results( ) -> None: monkeypatch.setattr(requests, "post", lambda *_a, **_kw: _FakeResponse(body)) with pytest.raises(ValueError, match="no results"): - tool._exa_content("ek", "q", "auto", 5) + tool._exa_content("ek", "q", "auto", 5, "summary") + + +def test_exa_content_mode_is_configurable(monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict[str, Any] = {} + + class _Settings: + integrations = IntegrationSettings( + EXA_API_KEY="ek", + STRIX_EXA_CONTENT_MODE="highlights", + ) + + def fake_post(_url: str, **kwargs: Any) -> _FakeResponse: + captured["json"] = kwargs["json"] + return _FakeResponse({"results": [{"url": "https://ex.example", "title": "Ex"}]}) + + monkeypatch.setattr(tool, "load_settings", _Settings) + monkeypatch.setattr(requests, "post", fake_post) + + assert tool._do_search("q")["success"] is True + assert "highlights" in captured["json"]["contents"] + assert "summary" not in captured["json"]["contents"] def test_do_search_reports_empty_exa_results_as_unexpected( @@ -198,6 +248,141 @@ def test_exa_search_type_and_num_results_are_configurable( assert captured["json"]["numResults"] == 3 +def test_exa_page_text_requests_full_text_and_renders_pages( + monkeypatch: pytest.MonkeyPatch, +) -> None: + captured: dict[str, Any] = {} + + def fake_post(url: str, **kwargs: Any) -> _FakeResponse: + captured["url"] = url + captured["headers"] = kwargs["headers"] + captured["json"] = kwargs["json"] + return _FakeResponse( + { + "results": [ + { + "url": "https://nvd.example/cve", + "title": "NVD entry", + "text": " Full advisory body. ", + }, + {"url": "https://empty.example", "text": " "}, + "not-a-dict", + {"text": "no url"}, + ], + } + ) + + monkeypatch.setattr(requests, "post", fake_post) + + content = tool._exa_page_text("ek", ["https://nvd.example/cve"]) + + assert captured["url"] == "https://api.exa.ai/contents" + assert captured["headers"]["x-api-key"] == "ek" + assert captured["json"] == {"urls": ["https://nvd.example/cve"], "text": True} + assert content == "### NVD entry\nhttps://nvd.example/cve\n\nFull advisory body." + + +def test_exa_page_text_truncates_a_long_page(monkeypatch: pytest.MonkeyPatch) -> None: + body = "A" * (tool._EXA_PAGE_MAX_CHARS + 500) + monkeypatch.setattr( + requests, + "post", + lambda *_a, **_kw: _FakeResponse( + {"results": [{"url": "https://ex.example", "text": body}]} + ), + ) + content = tool._exa_page_text("ek", ["https://ex.example"]) + assert "truncated at" in content + assert content.count("A") == tool._EXA_PAGE_MAX_CHARS + + +@pytest.mark.parametrize("body", [{}, {"results": []}, {"results": [{"url": "u"}]}]) +def test_exa_page_text_rejects_pages_without_text( + monkeypatch: pytest.MonkeyPatch, body: dict[str, Any] +) -> None: + monkeypatch.setattr(requests, "post", lambda *_a, **_kw: _FakeResponse(body)) + with pytest.raises(ValueError, match="no page contents"): + tool._exa_page_text("ek", ["https://ex.example"]) + + +@pytest.mark.parametrize("urls", [[], ["", " "]]) +def test_do_get_contents_requires_a_url(urls: list[str]) -> None: + result = tool._do_get_contents(urls) + assert result["success"] is False + assert "at least one URL" in result["error"] + + +def test_do_get_contents_caps_the_url_count() -> None: + urls = [f"https://ex{index}.example" for index in range(tool._EXA_MAX_CONTENT_URLS + 1)] + result = tool._do_get_contents(urls) + assert result["success"] is False + assert "Too many URLs" in result["error"] + + +def test_do_get_contents_needs_an_exa_key(monkeypatch: pytest.MonkeyPatch) -> None: + class _Settings: + integrations = IntegrationSettings(PERPLEXITY_API_KEY="pk") + + monkeypatch.setattr(tool, "load_settings", _Settings) + + result = tool._do_get_contents(["https://ex.example"]) + + assert result["success"] is False + assert "EXA_API_KEY" in result["error"] + + +def test_do_get_contents_refuses_a_perplexity_pinned_provider( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class _Settings: + integrations = IntegrationSettings( + EXA_API_KEY="ek", + PERPLEXITY_API_KEY="pk", + STRIX_WEB_SEARCH_PROVIDER="perplexity", + ) + + monkeypatch.setattr(tool, "load_settings", _Settings) + + result = tool._do_get_contents(["https://ex.example"]) + + assert result["success"] is False + assert "web_search" in result["error"] + + +def test_do_get_contents_returns_page_text(monkeypatch: pytest.MonkeyPatch) -> None: + class _Settings: + integrations = IntegrationSettings(EXA_API_KEY="ek") + + monkeypatch.setattr(tool, "load_settings", _Settings) + monkeypatch.setattr(tool, "_exa_page_text", lambda *_a: "page") + + result = tool._do_get_contents([" https://ex.example "]) + + assert result == { + "success": True, + "urls": ["https://ex.example"], + "provider": "exa", + "content": "page", + } + + +def test_do_get_contents_sanitizes_a_network_error(monkeypatch: pytest.MonkeyPatch) -> None: + class _Settings: + integrations = IntegrationSettings(EXA_API_KEY="ek") + + def boom(*_args: Any, **_kwargs: Any) -> None: + raise requests.exceptions.ConnectionError + + monkeypatch.setattr(tool, "load_settings", _Settings) + monkeypatch.setattr(requests, "post", boom) + + result = tool._do_get_contents(["https://ex.example"]) + + assert result["success"] is False + assert "network error" in result["error"] + assert "ek" not in result["error"] + + def test_do_search_reports_the_provider_it_used(monkeypatch: pytest.MonkeyPatch) -> None: class _Settings: integrations = IntegrationSettings(EXA_API_KEY="ek")