feat(web_search): add Exa as a web search provider alongside Perplexity

This commit is contained in:
Ahmed Allam 2026-09-04 11:52:15 +00:00
parent f6d9790ecb
commit 3822cbb160
8 changed files with 276 additions and 27 deletions

View file

@ -273,7 +273,8 @@ export LLM_API_KEY="your-api-key"
# Optional
export LLM_API_BASE="your-api-base-url" # if using a local model, e.g. Ollama, LMStudio
export PERPLEXITY_API_KEY="your-api-key" # for search capabilities
export EXA_API_KEY="your-api-key" # for search capabilities (preferred)
export PERPLEXITY_API_KEY="your-api-key" # alternative search provider
```
> [!NOTE]

View file

@ -80,6 +80,14 @@ affecting the agents that do the actual testing.
API key for Perplexity AI. Enables real-time web search during scans for OSINT and vulnerability research.
</ParamField>
<ParamField path="EXA_API_KEY" type="string">
API key for Exa. Enables real-time web search through the Exa `/answer` endpoint. This is the preferred web search provider.
</ParamField>
<ParamField path="STRIX_WEB_SEARCH_PROVIDER" default="auto" type="string">
Web search provider: `auto`, `perplexity`, or `exa`. With `auto`, Strix uses Exa when `EXA_API_KEY` is set, and Perplexity otherwise. Set an explicit provider to pin one when you configure both keys.
</ParamField>
<ParamField path="POSTMAN_API_KEY" type="string">
Postman API key (`PMAK-…`). Enables fetching Postman collections by id as a target (`postman://<collection-uid>`), and Postman environments (`postman://<collection-uid>?env=<environment-uid>`) to resolve collection variables. Not needed when passing a local collection export file.
</ParamField>
@ -159,7 +167,8 @@ strix --target ./app --config /path/to/config.json
export STRIX_LLM="openrouter/z-ai/glm-5.3"
export LLM_API_KEY="sk-..."
# Optional: Enable web search
# Optional: Enable web search (Exa preferred, Perplexity supported)
export EXA_API_KEY="..."
export PERPLEXITY_API_KEY="pplx-..."
# Optional: Custom timeouts

View file

@ -28,6 +28,6 @@ Strix agents use specialized tools to test your applications like a real penetra
| -------------- | ---------------------------------------- |
| Python Runtime | Write and execute custom exploit scripts |
| File Editor | Read and modify source code |
| Web Search | Real-time OSINT via Perplexity |
| Web Search | Real-time OSINT with Exa or Perplexity |
| Notes | Document findings during the scan |
| Reporting | Generate vulnerability reports with PoCs |

View file

@ -120,6 +120,9 @@ class TelemetrySettings(BaseSettings):
enabled: bool = Field(default=True, alias="STRIX_TELEMETRY")
WebSearchProvider = Literal["auto", "perplexity", "exa"]
class IntegrationSettings(BaseSettings):
model_config = _BASE_CONFIG
@ -128,6 +131,15 @@ class IntegrationSettings(BaseSettings):
alias="PERPLEXITY_API_KEY",
repr=False,
)
exa_api_key: str | None = Field(
default=None,
alias="EXA_API_KEY",
repr=False,
)
web_search_provider: WebSearchProvider = Field(
default="auto",
alias="STRIX_WEB_SEARCH_PROVIDER",
)
postman_api_key: str | None = Field(
default=None,
alias="POSTMAN_API_KEY",

View file

@ -46,7 +46,8 @@ def validate_environment() -> None:
if not settings.llm.api_base:
missing_optional_vars.append("LLM_API_BASE")
if not settings.integrations.perplexity_api_key:
if not settings.integrations.perplexity_api_key and not settings.integrations.exa_api_key:
missing_optional_vars.append("EXA_API_KEY")
missing_optional_vars.append("PERPLEXITY_API_KEY")
if missing_required_vars:
@ -89,7 +90,14 @@ def validate_environment() -> None:
error_text.append("• ", style="white")
error_text.append("PERPLEXITY_API_KEY", style="bold cyan")
error_text.append(
" - API key for Perplexity AI web search (enables real-time research)\n",
" - API key for Perplexity AI web search (alternative to Exa)\n",
style="white",
)
elif var == "EXA_API_KEY":
error_text.append("• ", style="white")
error_text.append("EXA_API_KEY", style="bold cyan")
error_text.append(
" - API key for Exa web search (enables real-time research)\n",
style="white",
)
elif var == "STRIX_REASONING_EFFORT":
@ -116,6 +124,8 @@ def validate_environment() -> None:
error_text.append(
"export PERPLEXITY_API_KEY='your-perplexity-key-here'\n", style="dim white"
)
elif var == "EXA_API_KEY":
error_text.append("export EXA_API_KEY='your-exa-key-here'\n", style="dim white")
elif var == "STRIX_REASONING_EFFORT":
error_text.append(
"export STRIX_REASONING_EFFORT='high'\n",

View file

@ -1,11 +1,11 @@
"""``web_search`` — Perplexity-backed security-focused web search."""
"""``web_search`` — security-focused web search (Exa or Perplexity)."""
from __future__ import annotations
import asyncio
import json
import logging
from typing import Any
from typing import Any, cast
import requests
from agents import RunContextWrapper, function_tool
@ -41,22 +41,7 @@ Structure your response to be comprehensive yet concise, emphasizing the most cr
security implications and details."""
def _do_search(query: str) -> dict[str, Any]: # noqa: PLR0911 - each error class needs its own sanitized return
if not query or not query.strip():
return {"success": False, "error": "Query cannot be empty"}
api_key = load_settings().integrations.perplexity_api_key
if not api_key:
logger.warning("web_search invoked without PERPLEXITY_API_KEY configured")
return {
"success": False,
"error": (
"Web search is not configured for this scan "
"(operator needs to set PERPLEXITY_API_KEY). Proceed without it"
),
}
logger.info("web_search query (len=%d): %s", len(query), query[:120])
def _perplexity_content(api_key: str, query: str) -> str:
url = "https://api.perplexity.ai/chat/completions"
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
payload = {
@ -66,11 +51,88 @@ def _do_search(query: str) -> dict[str, Any]: # noqa: PLR0911 - each error clas
{"role": "user", "content": query},
],
}
with requests.post(url, headers=headers, json=payload, timeout=300) as response:
response.raise_for_status()
return str(response.json()["choices"][0]["message"]["content"])
def _exa_content(api_key: str, query: str) -> str:
url = "https://api.exa.ai/answer"
headers = {"x-api-key": api_key, "Content-Type": "application/json"}
payload = {
"query": f"{_SYSTEM_PROMPT}\n\n{query}",
"text": True,
}
with requests.post(url, headers=headers, json=payload, timeout=300) as response:
response.raise_for_status()
body: dict[str, Any] = response.json()
answer = str(body.get("answer") or "").strip()
citations: list[Any] = body.get("citations") or []
lines: list[str] = []
for citation in citations:
if not isinstance(citation, dict):
continue
entry = cast("dict[str, Any]", citation)
citation_url = str(entry.get("url") or entry.get("id") or "")
if not citation_url:
continue
title = str(entry.get("title") or citation_url)
lines.append(f"- {title}: {citation_url}")
if lines:
return answer + "\n\nSources:\n" + "\n".join(lines)
return answer
def _resolve_provider( # noqa: PLR0911 - each provider/missing-key case needs its own return
integrations: Any,
) -> tuple[str, str] | dict[str, Any]:
"""Pick the search provider and its key, or return a sanitized error dict."""
provider = integrations.web_search_provider
perplexity_key = integrations.perplexity_api_key
exa_key = integrations.exa_api_key
if provider == "perplexity":
if not perplexity_key:
return _not_configured_error("PERPLEXITY_API_KEY")
return ("perplexity", perplexity_key)
if provider == "exa":
if not exa_key:
return _not_configured_error("EXA_API_KEY")
return ("exa", exa_key)
if exa_key:
return ("exa", exa_key)
if perplexity_key:
return ("perplexity", perplexity_key)
return _not_configured_error("EXA_API_KEY or PERPLEXITY_API_KEY")
def _not_configured_error(missing: str) -> dict[str, Any]:
logger.warning("web_search invoked without %s configured", missing)
return {
"success": False,
"error": (
"Web search is not configured for this scan "
f"(operator needs to set {missing}). Proceed without it"
),
}
def _do_search(query: str) -> dict[str, Any]: # noqa: PLR0911 - each error class needs its own sanitized return
if not query or not query.strip():
return {"success": False, "error": "Query cannot be empty"}
resolved = _resolve_provider(load_settings().integrations)
if isinstance(resolved, dict):
return resolved
provider, api_key = resolved
logger.info("web_search provider=%s query (len=%d): %s", provider, len(query), query[:120])
try:
with requests.post(url, headers=headers, json=payload, timeout=300) as response:
response.raise_for_status()
content = response.json()["choices"][0]["message"]["content"]
if provider == "exa":
content = _exa_content(api_key, query)
else:
content = _perplexity_content(api_key, query)
except requests.exceptions.Timeout:
logger.warning("web_search timed out")
return {
@ -114,13 +176,14 @@ def _do_search(query: str) -> dict[str, Any]: # noqa: PLR0911 - each error clas
return {
"success": True,
"query": query,
"provider": provider,
"content": content,
}
@function_tool(timeout=330)
async def web_search(ctx: RunContextWrapper, query: str) -> str:
"""Real-time web search via Perplexity — your primary research tool.
"""Real-time web search (Exa or Perplexity) — your primary research tool.
Use it liberally for anything that's not in your training data:

View file

@ -30,6 +30,8 @@ _LLM_ENV_KEYS = [
"STRIX_FORCE_REQUIRED_TOOL_CHOICE",
"LLM_TIMEOUT",
"PERPLEXITY_API_KEY",
"EXA_API_KEY",
"STRIX_WEB_SEARCH_PROVIDER",
# RuntimeSettings
"STRIX_IMAGE",
"STRIX_RUNTIME_BACKEND",
@ -80,6 +82,17 @@ def test_read_json_overrides_maps_to_nested_settings(tmp_path: Path) -> None:
}
def test_read_json_overrides_maps_exa_and_provider(tmp_path: Path) -> None:
path = tmp_path / "cli-config.json"
path.write_text(
json.dumps({"env": {"EXA_API_KEY": "exa-key", "STRIX_WEB_SEARCH_PROVIDER": "exa"}}),
encoding="utf-8",
)
assert loader._read_json_overrides(path) == {
"integrations": {"exa_api_key": "exa-key", "web_search_provider": "exa"},
}
def test_read_json_overrides_skips_keys_already_in_environ(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:

141
tests/test_web_search.py Normal file
View file

@ -0,0 +1,141 @@
"""Tests for web_search provider selection and the Exa backend."""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
import requests
from strix.config.settings import IntegrationSettings
from strix.tools.web_search import tool
if TYPE_CHECKING:
from typing import Self
import pytest
class _FakeResponse:
def __init__(self, body: dict[str, Any]) -> None:
self._body = body
self.headers: dict[str, str] = {}
def __enter__(self) -> Self:
return self
def __exit__(self, *_exc: object) -> None:
return None
def raise_for_status(self) -> None:
return None
def json(self) -> dict[str, Any]:
return self._body
def test_auto_prefers_exa_when_both_keys_set() -> None:
integrations = IntegrationSettings(PERPLEXITY_API_KEY="pk", EXA_API_KEY="ek")
assert tool._resolve_provider(integrations) == ("exa", "ek")
def test_auto_falls_back_to_perplexity_when_only_perplexity_is_set() -> None:
integrations = IntegrationSettings(PERPLEXITY_API_KEY="pk")
assert tool._resolve_provider(integrations) == ("perplexity", "pk")
def test_explicit_exa_ignores_a_configured_perplexity_key() -> None:
integrations = IntegrationSettings(
PERPLEXITY_API_KEY="pk",
EXA_API_KEY="ek",
STRIX_WEB_SEARCH_PROVIDER="exa",
)
assert tool._resolve_provider(integrations) == ("exa", "ek")
def test_explicit_perplexity_ignores_a_configured_exa_key() -> None:
integrations = IntegrationSettings(
PERPLEXITY_API_KEY="pk",
EXA_API_KEY="ek",
STRIX_WEB_SEARCH_PROVIDER="perplexity",
)
assert tool._resolve_provider(integrations) == ("perplexity", "pk")
def test_explicit_exa_without_a_key_names_only_exa() -> None:
integrations = IntegrationSettings(
PERPLEXITY_API_KEY="pk",
STRIX_WEB_SEARCH_PROVIDER="exa",
)
resolved = tool._resolve_provider(integrations)
assert isinstance(resolved, dict)
assert resolved["success"] is False
assert "EXA_API_KEY" in resolved["error"]
assert "PERPLEXITY_API_KEY" not in resolved["error"]
def test_no_keys_names_both_providers() -> None:
resolved = tool._resolve_provider(IntegrationSettings())
assert isinstance(resolved, dict)
assert "EXA_API_KEY or PERPLEXITY_API_KEY" in resolved["error"]
def test_exa_content_appends_citation_sources(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(
{
"answer": "CVE-2024-0001 affects it.",
"citations": [
{"url": "https://nvd.example/cve", "title": "NVD entry"},
{"id": "https://blog.example/post"},
"not-a-dict",
{"title": "no url"},
],
}
)
monkeypatch.setattr(requests, "post", fake_post)
content = tool._exa_content("ek", "OpenSSH 7.4 RCE?")
assert captured["url"] == "https://api.exa.ai/answer"
assert captured["headers"]["x-api-key"] == "ek"
assert "OpenSSH 7.4 RCE?" in captured["json"]["query"]
assert content == (
"CVE-2024-0001 affects it.\n\nSources:\n"
"- NVD entry: https://nvd.example/cve\n"
"- https://blog.example/post: https://blog.example/post"
)
def test_exa_content_without_citations_returns_the_answer(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(
requests,
"post",
lambda *_a, **_kw: _FakeResponse({"answer": " Nothing found. "}),
)
assert tool._exa_content("ek", "q") == "Nothing found."
def test_do_search_reports_the_provider_it_used(monkeypatch: pytest.MonkeyPatch) -> None:
class _Settings:
integrations = IntegrationSettings(EXA_API_KEY="ek")
monkeypatch.setattr(tool, "load_settings", _Settings)
monkeypatch.setattr(tool, "_exa_content", lambda *_a: "answer")
result = tool._do_search("OpenSSH 7.4 RCE?")
assert result == {
"success": True,
"query": "OpenSSH 7.4 RCE?",
"provider": "exa",
"content": "answer",
}