feat(community): add SpecStream web search and fetch provider (#7968)

* feat(community): add SpecStream web search and fetch provider

* fix(specstream): keep a direct answer when sources are empty, add changelog entry

Signed-off-by: Nataliia <nati@macbook-pro-nataliia.home>

---------

Signed-off-by: Nataliia <nati@macbook-pro-nataliia.home>
Co-authored-by: Specstreamai <333137486+Specstreamai@users.noreply.github.com>
This commit is contained in:
SpecStream 2026-11-09 13:32:22 +01:00 committed by GitHub
parent 7adb2062eb
commit fa8e015023
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
15 changed files with 657 additions and 7 deletions

View File

@ -30,6 +30,9 @@ INFOQUEST_API_KEY=your-infoquest-api-key
# Sofya API Key (web search and fetch) - https://sofya.co
SOFYA_API_KEY=your-sofya-api-key
# SpecStream API Key (web search and fetch) - https://specstream.ai
SPECSTREAM_API_KEY=your-specstream-api-key
# Unbrowse API Key (web fetch) - https://unbrowse.ai/app
UNBROWSE_API_KEY=your-unbrowse-api-key

View File

@ -27,6 +27,10 @@ This section accumulates work toward the **2.3.0** milestone
Exact-user allowlists remain the default; shared access requires matching
host-verified identity and grants no management permissions. ([#7944](https://github.com/bytedance/deer-flow/pull/7944))
- **community:** New SpecStream `web_search` and `web_fetch` provider. Search returns
ranked results with a direct answer when the sources agree on one, and fetch
returns a page or document as markdown. ([#7968](https://github.com/bytedance/deer-flow/pull/7968))
- **subagents:** Durable batch items can wait for explicitly named prerequisites.
An optional, once-only lead-Agent follow-up completes a previously requested
summary or delivery after the batch ends, with persisted admission, cancellation

View File

@ -1304,8 +1304,8 @@ offline mocked tests verify request composition and local filtering, not provide
retrieval behavior. No paid API calls are needed for the regression suite.
**Built-in Tools**:
- `web_search` - Search the web (DuckDuckGo, Tavily, Brave, Serper, Serply, Exa, InfoQuest, Tencent Cloud WSA, Firecrawl, fastCRW, GroundRoute, Sofya, You.com)
- `web_fetch` - Fetch web pages (Jina AI, Crawl4AI, Exa, InfoQuest, Firecrawl, fastCRW, GroundRoute, Browserless, Sofya, Unbrowse)
- `web_search` - Search the web (DuckDuckGo, Tavily, Brave, Serper, Serply, Exa, InfoQuest, Tencent Cloud WSA, Firecrawl, fastCRW, GroundRoute, Sofya, SpecStream, You.com)
- `web_fetch` - Fetch web pages (Jina AI, Crawl4AI, Exa, InfoQuest, Firecrawl, fastCRW, GroundRoute, Browserless, Sofya, SpecStream, Unbrowse)
- `web_capture` - Capture rendered webpage screenshots as artifacts (Browserless)
- `image_search` - Search for reference images (DuckDuckGo, InfoQuest, Serper, Brave)
- `ls` - List directory contents
@ -2077,6 +2077,7 @@ models:
- `SERPLY_API_KEY` - [Serply](https://serply.io) key for `web_search` (Google Search, plus Google News and Google Scholar via `vertical`)
- `GROUNDROUTE_API_KEY` - GroundRoute meta-search API key for `web_search` and `web_fetch` (routes across Serper, Brave, Exa, Tavily, Firecrawl, Perplexity with gain-share pricing)
- `SOFYA_API_KEY` - [Sofya](https://sofya.co) key for `web_search` and `web_fetch`
- `SPECSTREAM_API_KEY` - [SpecStream](https://specstream.ai) key for `web_search` and `web_fetch`
- `UNBROWSE_API_KEY` - [Unbrowse](https://unbrowse.ai) key for `web_fetch`
- `YDC_API_KEY` - You.com Search API key for `web_search` (optional: without it the provider uses You.com's keyless free-tier endpoint)
- `BROWSERLESS_TOKEN` - Browserless token for `web_fetch` (Browserless provider) and `web_capture`, sent as the `token` query parameter (required by Browserless Cloud and by a self-hosted instance started with `TOKEN`)

View File

@ -0,0 +1,3 @@
from .tools import web_fetch_tool, web_search_tool
__all__ = ["web_fetch_tool", "web_search_tool"]

View File

@ -0,0 +1,196 @@
"""
Web search and fetch tools powered by SpecStream.
SpecStream is a live web data engine with a Search API and an Extract API.
Search answers a question from the live web and returns the search results it
read, ranked, with a direct answer when the sources agree on one. Fetch returns
the main text of a page, PDF, Word or Excel file as markdown. An API key is
required. Sign up at https://specstream.ai to get one.
"""
import json
import logging
import os
import httpx
from langchain.tools import tool
from deerflow.community.search_max_results import coerce_max_results
from deerflow.community.url_safety import safe_public_result_url
from deerflow.config import get_app_config
logger = logging.getLogger(__name__)
_SPECSTREAM_BASE_URL = "https://api.specstream.ai"
_SPECSTREAM_SEARCH_PATH = "/v1/web-search"
_SPECSTREAM_EXTRACT_PATH = "/v1/extract"
# The search API takes a question of 1 to 2000 characters.
_SPECSTREAM_MAX_QUERY_CHARS = 2000
# Each fetched URL gets up to 30 seconds on the SpecStream side.
_SPECSTREAM_TIMEOUT = 60
_SPECSTREAM_FETCH_MAX_CHARS = 4096
_api_key_warned: set[str] = set()
def _get_api_key(tool_name: str) -> str | None:
config = get_app_config().get_tool_config(tool_name)
if config is not None:
api_key = config.model_extra.get("api_key")
if isinstance(api_key, str) and api_key.strip():
return api_key.strip()
env_key = os.getenv("SPECSTREAM_API_KEY")
if isinstance(env_key, str) and env_key.strip():
return env_key.strip()
return None
def _missing_key_message(tool_name: str) -> str:
if tool_name not in _api_key_warned:
_api_key_warned.add(tool_name)
logger.warning("SpecStream API key is not set for '%s'. Set SPECSTREAM_API_KEY in your environment or provide api_key in config.yaml. Sign up at https://specstream.ai", tool_name)
return "SPECSTREAM_API_KEY is not configured"
def _specstream_request(method: str, path: str, api_key: str, *, params: dict | None = None, payload: dict | None = None) -> tuple[dict | None, str | None]:
"""Send a request to a SpecStream endpoint.
Returns a ``(data, error)`` tuple: on success ``data`` is the object inside
the response's ``data`` envelope and ``error`` is ``None``; on failure
``data`` is ``None`` and ``error`` is a message the caller can hand back to
the model.
"""
headers = {"Authorization": f"Bearer {api_key}"}
try:
with httpx.Client(timeout=_SPECSTREAM_TIMEOUT) as client:
response = client.request(method, f"{_SPECSTREAM_BASE_URL}{path}", headers=headers, params=params, json=payload)
response.raise_for_status()
body = response.json()
except httpx.HTTPStatusError as e:
logger.error("SpecStream API returned HTTP %s: %s", e.response.status_code, (e.response.text or "")[:500])
return None, f"SpecStream API error: HTTP {e.response.status_code}"
except Exception as e:
logger.error("SpecStream request failed: %s: %s", type(e).__name__, str(e)[:500])
return None, str(e)[:500]
data = body.get("data") if isinstance(body, dict) else None
if not isinstance(data, dict):
logger.error("SpecStream returned an unexpected payload type: %s", type(body).__name__)
return None, "SpecStream returned an unexpected response format"
return data, None
def _response_objects(data: dict, container: str) -> list[dict] | None:
"""Return the object entries of a SpecStream response container, or None if malformed.
A missing or null container is empty. A non-list container, or a non-empty
list holding no objects, is malformed: the caller reports a format error
rather than "No results found", which would tell the agent the query
legitimately matched nothing.
"""
value = data.get(container)
if value is None:
return []
if not isinstance(value, list):
logger.error("SpecStream returned an unexpected '%s' payload type: %s", container, type(value).__name__)
return None
usable = [item for item in value if isinstance(item, dict)]
if value and not usable:
logger.error("SpecStream returned '%s' with no usable result objects", container)
return None
return usable
def _normalize_answer(answer: object) -> dict | None:
"""Keep the direct answer with the page it was read from, or None when there is none."""
if not isinstance(answer, dict) or not answer.get("value"):
return None
return {
"value": answer.get("value"),
"url": safe_public_result_url(answer.get("page")),
}
@tool("web_search", parse_docstring=True)
def web_search_tool(query: str, max_results: int | None = None) -> str:
"""Search the web for information. Use this tool to find current information, news, articles, and facts from the internet.
Args:
query: Search keywords or a question describing what you want to find. Be specific for better results.
max_results: Maximum number of results to return. If omitted, uses the configured value (default 5).
"""
config = get_app_config().get_tool_config("web_search")
config_extra = (config.model_extra or {}) if config is not None else {}
# Honor the caller-supplied max_results; fall back to config only when omitted.
if max_results is None:
max_results = config_extra.get("max_results")
max_results = coerce_max_results(max_results, provider="SpecStream", logger=logger)
api_key = _get_api_key("web_search")
if not api_key:
return json.dumps({"error": _missing_key_message("web_search"), "query": query}, ensure_ascii=False)
data, error = _specstream_request("GET", _SPECSTREAM_SEARCH_PATH, api_key, params={"q": query.strip()[:_SPECSTREAM_MAX_QUERY_CHARS]})
if error is not None:
return json.dumps({"error": error, "query": query}, ensure_ascii=False)
sources = _response_objects(data, "sources")
if sources is None:
return json.dumps({"error": "SpecStream returned an unexpected response format", "query": query}, ensure_ascii=False)
answer = _normalize_answer(data.get("answer"))
if not sources and answer is None:
return json.dumps({"error": "No results found", "query": query}, ensure_ascii=False)
output: dict[str, object] = {"query": query}
if answer is not None:
output["answer"] = answer
results = [
{
"title": source.get("title") or "",
"url": safe_public_result_url(source.get("url")),
"snippet": source.get("snippet") or "",
}
for source in sources[:max_results]
]
output["total_results"] = len(results)
output["results"] = results
return json.dumps(output, indent=2, ensure_ascii=False)
@tool("web_fetch", parse_docstring=True)
def web_fetch_tool(url: str) -> str:
"""Fetch the contents of a web page at a given URL.
Only fetch EXACT URLs that have been provided directly by the user or have been returned in results from the web_search and web_fetch tools.
This tool can NOT access content that requires authentication, such as private Google Docs or pages behind login walls.
Do NOT add www. to URLs that do NOT have them.
URLs must include the schema: https://example.com is a valid URL while example.com is an invalid URL.
Args:
url: The URL to fetch the contents of.
"""
api_key = _get_api_key("web_fetch")
if not api_key:
return f"Error: {_missing_key_message('web_fetch')}"
data, error = _specstream_request("POST", _SPECSTREAM_EXTRACT_PATH, api_key, payload={"urls": [url], "format": "markdown"})
if error is not None:
return f"Error: {error}"
results = _response_objects(data, "results")
if results is None:
return "Error: SpecStream returned an unexpected response format"
if not results:
return "Error: No results found"
result = results[0]
status = result.get("status")
if status != "ok":
return f"Error: SpecStream could not read the page ({status or 'unknown status'})"
content = result.get("text")
content = content[:_SPECSTREAM_FETCH_MAX_CHARS] if isinstance(content, str) else ""
if not content:
return "Error: No content found"
title = result.get("title") or "Untitled"
return f"# {title}\n\n{content}"

View File

@ -104,6 +104,7 @@ Durable MCP task-management tools are added only while the process-local task su
**Community tools** (`packages/harness/deerflow/community/`): optional integrations, each in its own subpackage and wired through `config.yaml`. Documented examples:
- `tavily/` - Web search (5 results default) and web fetch (4KB limit)
- `sofya/` - Web search (5 results default, per-result content capped at 2000 chars) and web fetch (4KB limit)
- `specstream/` - Web search (5 results default, with a direct answer when the sources agree) and web fetch (4KB limit)
- `jina_ai/` - Web fetch via Jina reader API with readability extraction
- `unbrowse/` - Web fetch (4KB limit) via one JSON-RPC `tools/call` to Unbrowse's hosted MCP endpoint; `render: auto|never|always` picks plain HTTP or its cloud browser
- `firecrawl/` - Web scraping via Firecrawl API
@ -114,7 +115,7 @@ Durable MCP task-management tools are added only while the process-local task su
The shared SSRF guard (`community/url_safety.py`) resolves hostnames with blocking `socket.getaddrinfo`, so async tools, Gateway routes, and the Playwright request guard (`BrowserSession._install_request_guard`, which runs on the shared browser loop) call `validate_public_http_url` (or a wrapper such as `validate_browser_url`) through `asyncio.to_thread`. The guard checks at validation time only, so a launched browser sends every TCP connection (HTTP, HTTPS, WebSocket) through its per-session loopback SOCKS5 proxy (`browser_automation/egress.py`, `bypass: <-loopback>`): Chromium hands it every hostname, and it resolves once via `resolve_browser_egress` → `url_safety.resolve_public_addresses` and connects to exactly those addresses, closing the DNS-rebinding window (`mcp/personal_network.py` pins the same way). Every `acquire_session`/`get_session` caller must pass `egress_resolver`, because a thread's first caller fixes its launch options. WebRTC UDP does not traverse a SOCKS proxy and is not pinned (out of scope, so this is not a complete egress guarantee). Delegated fetch services (Browserless, crawl4ai, fastcrw, firecrawl, and Exa with a configured `base_url`) resolve on their own side and cannot be pinned, so each `web_fetch`/`web_capture` entry point instead screens its own backend base URL with `validate_delegated_backend_url` and fails closed on a self-hosted backend unless the operator confirms the backend's egress is isolated via `network_isolation_confirmed`; CDP-attached Chrome has neither guard. The strict blocking-IO gate carries a `socket.getaddrinfo` rule for this; `tests/blocking_io/test_web_tool_url_validation.py` drives each web tool entry point, the request guard, and the egress proxy; `tests/test_browser_egress.py` pins the proxy protocol, the session/manager/router wiring, and (with Playwright installed) a real-Chromium rebinding case.
Additional providers also live here (`boxlite`, `brave`, `browserless`, `crawl4ai`, `ddg_search`, `e2b_sandbox`, `exa`, `fastcrw`, `groundroute`, `infoquest`, `searxng`, `serper`, `serply`, `sofya`, `tencent_wsa`, `tenki`, `unbrowse`, `youcom`); see each subpackage for specifics. Browserless authenticates only from the `token` query parameter or `Authorization` header and rejects unknown body keys, so `BrowserlessClient._auth_params()` carries the token for every endpoint; never add it to the JSON body. `tencent_wsa` uses Tencent Cloud Web Search's service API key endpoint (`TENCENTCLOUD_WSA_APIKEY`), not Tencent Cloud SecretId/SecretKey signing. Its `max_results` is capped at 50; requests above 10 use Tencent's optional `Cnt` parameter, which needs a Tencent Cloud plan that supports it. E2B bootstrap is required. If it fails, the provider kills and closes the unusable remote sandbox. New sandbox creation raises an error. Warm-pool reclaim and remote discovery discard the sandbox and continue acquisition. E2B mounts remain optional.
Additional providers also live here (`boxlite`, `brave`, `browserless`, `crawl4ai`, `ddg_search`, `e2b_sandbox`, `exa`, `fastcrw`, `groundroute`, `infoquest`, `searxng`, `serper`, `serply`, `sofya`, `specstream`, `tencent_wsa`, `tenki`, `unbrowse`, `youcom`); see each subpackage for specifics. Browserless authenticates only from the `token` query parameter or `Authorization` header and rejects unknown body keys, so `BrowserlessClient._auth_params()` carries the token for every endpoint; never add it to the JSON body. `tencent_wsa` uses Tencent Cloud Web Search's service API key endpoint (`TENCENTCLOUD_WSA_APIKEY`), not Tencent Cloud SecretId/SecretKey signing. Its `max_results` is capped at 50; requests above 10 use Tencent's optional `Cnt` parameter, which needs a Tencent Cloud plan that supports it. E2B bootstrap is required. If it fails, the provider kills and closes the unusable remote sandbox. New sandbox creation raises an error. Warm-pool reclaim and remote discovery discard the sandbox and continue acquisition. E2B mounts remain optional.
E2B output sync records remote file versions and actual host file metadata in a thread-local manifest. The manifest binds to the remote sandbox ID. A complete output listing removes entries for deleted files. This avoids repeat downloads when the host filesystem rounds modification times. A single release-time sync pass is bounded by aggregate ceilings (`_MAX_SYNC_TOTAL_BYTES`, `_MAX_SYNC_FILES`, `_SYNC_DEADLINE_SECONDS`) on top of the per-file `_MAX_DOWNLOAD_SIZE` cap, so a pathological outputs tree cannot make release download unboundedly; a truncated pass logs what it dropped and leaves the manifest un-pruned (only entries observed in that pass are reconciled), so files it never reached are retried on the next release rather than being forgotten.

View File

@ -640,6 +640,22 @@ class TestCheckWebSearch:
assert result.status == "warn"
assert "SOFYA_API_KEY" in (result.fix or "")
def test_specstream_with_key_ok(self, tmp_path, monkeypatch):
monkeypatch.setenv("SPECSTREAM_API_KEY", "test-key")
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\ntools:\n - name: web_search\n use: deerflow.community.specstream.tools:web_search_tool\n")
result = doctor.check_web_search(cfg)
assert result.status == "ok"
assert "specstream" in result.detail
def test_specstream_without_key_warns(self, tmp_path, monkeypatch):
monkeypatch.delenv("SPECSTREAM_API_KEY", raising=False)
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\ntools:\n - name: web_search\n use: deerflow.community.specstream.tools:web_search_tool\n")
result = doctor.check_web_search(cfg)
assert result.status == "warn"
assert "SPECSTREAM_API_KEY" in (result.fix or "")
def test_serper_with_key_ok(self, tmp_path, monkeypatch):
monkeypatch.setenv("SERPER_API_KEY", "test-key")
cfg = tmp_path / "config.yaml"
@ -769,6 +785,14 @@ class TestCheckWebFetch:
assert result.status == "warn"
assert "SOFYA_API_KEY" in (result.fix or "")
def test_specstream_without_key_warns(self, tmp_path, monkeypatch):
monkeypatch.delenv("SPECSTREAM_API_KEY", raising=False)
cfg = tmp_path / "config.yaml"
cfg.write_text("config_version: 5\ntools:\n - name: web_fetch\n use: deerflow.community.specstream.tools:web_fetch_tool\n")
result = doctor.check_web_fetch(cfg)
assert result.status == "warn"
assert "SPECSTREAM_API_KEY" in (result.fix or "")
def test_unbrowse_with_key_ok(self, tmp_path, monkeypatch):
monkeypatch.setenv("UNBROWSE_API_KEY", "test-key")
cfg = tmp_path / "config.yaml"

View File

@ -88,6 +88,11 @@ def _stub_tencent_wsa(module, url):
return [patch.object(module, "_get_api_key", return_value="k"), patch.object(module, "_search", return_value=(data, None))], _sync(module.web_search_tool)
def _stub_specstream(module, url):
data = {"sources": [{"title": "t", "url": url, "snippet": "c"}]}
return [patch.object(module, "_get_api_key", return_value="k"), patch.object(module, "_specstream_request", return_value=(data, None))], _sync(module.web_search_tool)
def _stub_serply(module, url):
data = {"results": [{"title": "t", "link": url, "description": "c"}]}
return [patch.object(module, "_get_api_key", return_value="k"), patch.object(module, "_serply_get", return_value=(data, None))], _sync(module.web_search_tool)
@ -169,6 +174,7 @@ _TEXT_SEARCH_STUBS = {
"serply": _stub_serply,
"groundroute": _stub_groundroute,
"sofya": _stub_sofya,
"specstream": _stub_specstream,
"ddg_search": _stub_ddg_search,
"fastcrw": _stub_fastcrw,
"firecrawl": _stub_firecrawl,

View File

@ -32,6 +32,7 @@ from deerflow.community.firecrawl import tools as firecrawl_tools
from deerflow.community.image_search import tools as image_search_tools
from deerflow.community.search_max_results import DEFAULT_MAX_RESULTS, coerce_max_results
from deerflow.community.serper import tools as serper_tools
from deerflow.community.specstream import tools as specstream_tools
from deerflow.community.youcom import tools as youcom_tools
COMMUNITY_ROOT = Path(ddg_search_tools.__file__).resolve().parent.parent
@ -47,6 +48,7 @@ SHARED_OWNER_PROVIDERS = {
"fastcrw": ("fastCRW", fastcrw_tools),
"firecrawl": ("Firecrawl", firecrawl_tools),
"youcom": ("You.com", youcom_tools),
"specstream": ("SpecStream", specstream_tools),
}
# Providers that still normalize locally, including Exa's generic helper and

View File

@ -0,0 +1,332 @@
"""Unit tests for the SpecStream community web search and fetch tools."""
import json
import logging
from unittest.mock import MagicMock, patch
import httpx
import pytest
_MODULE = "deerflow.community.specstream.tools"
@pytest.fixture(autouse=True)
def reset_api_key_warned():
"""Reset the module-level warning flag before each test."""
import deerflow.community.specstream.tools as specstream_mod
specstream_mod._api_key_warned = set()
yield
specstream_mod._api_key_warned = set()
@pytest.fixture
def mock_config_with_key():
with patch(f"{_MODULE}.get_app_config") as mock:
tool_config = MagicMock()
tool_config.model_extra = {"api_key": "test-specstream-key", "max_results": 5}
mock.return_value.get_tool_config.return_value = tool_config
yield mock
@pytest.fixture
def mock_config_no_key():
with patch(f"{_MODULE}.get_app_config") as mock:
tool_config = MagicMock()
tool_config.model_extra = {}
mock.return_value.get_tool_config.return_value = tool_config
yield mock
def _make_response(data: object) -> MagicMock:
mock_resp = MagicMock()
mock_resp.json.return_value = {"data": data}
mock_resp.raise_for_status = MagicMock()
return mock_resp
def _source(n: int, **extra) -> dict:
return {"url": f"https://example.com/{n}", "title": f"Result {n}", "site": "example.com", "snippet": f"Snippet {n}", "cited": False, **extra}
class TestGetApiKey:
def test_returns_config_key_when_present(self):
with patch(f"{_MODULE}.get_app_config") as mock:
tool_config = MagicMock()
tool_config.model_extra = {"api_key": "from-config"}
mock.return_value.get_tool_config.return_value = tool_config
from deerflow.community.specstream.tools import _get_api_key
assert _get_api_key("web_search") == "from-config"
def test_falls_back_to_env_when_config_key_whitespace(self):
with patch(f"{_MODULE}.get_app_config") as mock:
tool_config = MagicMock()
tool_config.model_extra = {"api_key": " "}
mock.return_value.get_tool_config.return_value = tool_config
with patch.dict("os.environ", {"SPECSTREAM_API_KEY": "env-key"}):
from deerflow.community.specstream.tools import _get_api_key
assert _get_api_key("web_search") == "env-key"
def test_uses_env_when_tool_is_not_configured(self):
with patch(f"{_MODULE}.get_app_config") as mock:
mock.return_value.get_tool_config.return_value = None
with patch.dict("os.environ", {"SPECSTREAM_API_KEY": "env-only"}):
from deerflow.community.specstream.tools import _get_api_key
assert _get_api_key("web_fetch") == "env-only"
def test_returns_none_when_no_key_anywhere(self):
with patch(f"{_MODULE}.get_app_config") as mock:
mock.return_value.get_tool_config.return_value = None
with patch.dict("os.environ", {}, clear=True):
from deerflow.community.specstream.tools import _get_api_key
assert _get_api_key("web_search") is None
class TestMissingKeyMessage:
def test_warns_once_per_tool_name(self, caplog):
from deerflow.community.specstream.tools import _missing_key_message
with caplog.at_level(logging.WARNING, logger=_MODULE):
assert _missing_key_message("web_search") == "SPECSTREAM_API_KEY is not configured"
_missing_key_message("web_search")
assert len([r for r in caplog.records if "SPECSTREAM_API_KEY" in r.getMessage()]) == 1
class TestWebSearchTool:
def test_basic_search_returns_normalized_results(self, mock_config_with_key):
data = {"sources": [_source(1), _source(2)], "answer": None}
with patch(f"{_MODULE}.httpx.Client") as mock_client_cls:
mock_request = mock_client_cls.return_value.__enter__.return_value.request
mock_request.return_value = _make_response(data)
from deerflow.community.specstream.tools import web_search_tool
parsed = json.loads(web_search_tool.invoke({"query": "python tutorial"}))
assert parsed["query"] == "python tutorial"
assert parsed["total_results"] == 2
assert parsed["results"][0] == {"title": "Result 1", "url": "https://example.com/1", "snippet": "Snippet 1"}
assert "answer" not in parsed
assert mock_request.call_args.args == ("GET", "https://api.specstream.ai/v1/web-search")
assert mock_request.call_args.kwargs["headers"]["Authorization"] == "Bearer test-specstream-key"
assert mock_request.call_args.kwargs["params"] == {"q": "python tutorial"}
def test_direct_answer_is_returned_with_its_page(self, mock_config_with_key):
data = {"sources": [_source(1)], "answer": {"value": "1889", "page": "https://example.com/1", "seen_at": "2026-11-01T00:00:00Z"}}
with patch(f"{_MODULE}.httpx.Client") as mock_client_cls:
mock_client_cls.return_value.__enter__.return_value.request.return_value = _make_response(data)
from deerflow.community.specstream.tools import web_search_tool
parsed = json.loads(web_search_tool.invoke({"query": "when was it built"}))
assert parsed["answer"] == {"value": "1889", "url": "https://example.com/1"}
def test_answer_page_is_screened(self, mock_config_with_key):
data = {"sources": [_source(1)], "answer": {"value": "x", "page": "http://169.254.169.254/latest/meta-data/"}}
with patch(f"{_MODULE}.httpx.Client") as mock_client_cls:
mock_client_cls.return_value.__enter__.return_value.request.return_value = _make_response(data)
from deerflow.community.specstream.tools import web_search_tool
parsed = json.loads(web_search_tool.invoke({"query": "q"}))
assert parsed["answer"]["url"] == ""
def test_long_query_is_cut_to_the_api_limit(self, mock_config_with_key):
with patch(f"{_MODULE}.httpx.Client") as mock_client_cls:
mock_request = mock_client_cls.return_value.__enter__.return_value.request
mock_request.return_value = _make_response({"sources": [_source(1)]})
from deerflow.community.specstream.tools import web_search_tool
web_search_tool.invoke({"query": "a" * 2500})
assert len(mock_request.call_args.kwargs["params"]["q"]) == 2000
def test_respects_max_results_from_config(self, mock_config_with_key):
mock_config_with_key.return_value.get_tool_config.return_value.model_extra["max_results"] = 2
with patch(f"{_MODULE}.httpx.Client") as mock_client_cls:
mock_client_cls.return_value.__enter__.return_value.request.return_value = _make_response({"sources": [_source(n) for n in range(5)]})
from deerflow.community.specstream.tools import web_search_tool
parsed = json.loads(web_search_tool.invoke({"query": "test"}))
assert parsed["total_results"] == 2
def test_caller_max_results_wins_over_config(self, mock_config_with_key):
with patch(f"{_MODULE}.httpx.Client") as mock_client_cls:
mock_client_cls.return_value.__enter__.return_value.request.return_value = _make_response({"sources": [_source(n) for n in range(8)]})
from deerflow.community.specstream.tools import web_search_tool
parsed = json.loads(web_search_tool.invoke({"query": "test", "max_results": 3}))
assert parsed["total_results"] == 3
def test_empty_sources_return_error_json(self, mock_config_with_key):
with patch(f"{_MODULE}.httpx.Client") as mock_client_cls:
mock_client_cls.return_value.__enter__.return_value.request.return_value = _make_response({"sources": []})
from deerflow.community.specstream.tools import web_search_tool
parsed = json.loads(web_search_tool.invoke({"query": "test"}))
assert parsed["error"] == "No results found"
@pytest.mark.parametrize("sources", [[], None])
def test_answer_without_sources_is_still_returned(self, mock_config_with_key, sources):
data = {"sources": sources, "answer": {"value": "1889", "page": "https://example.com/1"}}
with patch(f"{_MODULE}.httpx.Client") as mock_client_cls:
mock_client_cls.return_value.__enter__.return_value.request.return_value = _make_response(data)
from deerflow.community.specstream.tools import web_search_tool
parsed = json.loads(web_search_tool.invoke({"query": "when was it built"}))
assert "error" not in parsed
assert parsed["answer"] == {"value": "1889", "url": "https://example.com/1"}
assert parsed["total_results"] == 0
assert parsed["results"] == []
@pytest.mark.parametrize("sources", ["oops", [1, "a"]])
def test_malformed_sources_return_format_error(self, mock_config_with_key, sources, caplog):
with patch(f"{_MODULE}.httpx.Client") as mock_client_cls:
mock_client_cls.return_value.__enter__.return_value.request.return_value = _make_response({"sources": sources})
from deerflow.community.specstream.tools import web_search_tool
with caplog.at_level(logging.ERROR, logger=_MODULE):
parsed = json.loads(web_search_tool.invoke({"query": "test"}))
assert parsed["error"] == "SpecStream returned an unexpected response format"
def test_reply_without_data_envelope_returns_format_error(self, mock_config_with_key):
mock_resp = MagicMock()
mock_resp.json.return_value = {"sources": []}
with patch(f"{_MODULE}.httpx.Client") as mock_client_cls:
mock_client_cls.return_value.__enter__.return_value.request.return_value = mock_resp
from deerflow.community.specstream.tools import web_search_tool
parsed = json.loads(web_search_tool.invoke({"query": "test"}))
assert parsed["error"] == "SpecStream returned an unexpected response format"
def test_http_error_returns_error_json(self, mock_config_with_key):
request = httpx.Request("GET", "https://api.specstream.ai/v1/web-search")
response = httpx.Response(402, text="balance is empty", request=request)
mock_resp = MagicMock()
mock_resp.raise_for_status.side_effect = httpx.HTTPStatusError("error", request=request, response=response)
with patch(f"{_MODULE}.httpx.Client") as mock_client_cls:
mock_client_cls.return_value.__enter__.return_value.request.return_value = mock_resp
from deerflow.community.specstream.tools import web_search_tool
parsed = json.loads(web_search_tool.invoke({"query": "test"}))
assert parsed["error"] == "SpecStream API error: HTTP 402"
def test_network_error_returns_error_json(self, mock_config_with_key):
with patch(f"{_MODULE}.httpx.Client") as mock_client_cls:
mock_client_cls.return_value.__enter__.return_value.request.side_effect = httpx.ConnectError("boom")
from deerflow.community.specstream.tools import web_search_tool
parsed = json.loads(web_search_tool.invoke({"query": "test"}))
assert parsed["error"] == "boom"
def test_missing_key_returns_error_json(self, mock_config_no_key):
with patch.dict("os.environ", {}, clear=True):
from deerflow.community.specstream.tools import web_search_tool
parsed = json.loads(web_search_tool.invoke({"query": "test"}))
assert parsed["error"] == "SPECSTREAM_API_KEY is not configured"
assert parsed["query"] == "test"
class TestWebFetchTool:
def _fetch(self, data):
with patch(f"{_MODULE}.httpx.Client") as mock_client_cls:
mock_request = mock_client_cls.return_value.__enter__.return_value.request
mock_request.return_value = _make_response(data)
from deerflow.community.specstream.tools import web_fetch_tool
return web_fetch_tool.invoke({"url": "https://example.com/post"}), mock_request
def test_returns_title_and_content(self, mock_config_with_key):
result, mock_request = self._fetch({"results": [{"url": "https://example.com/post", "status": "ok", "title": "Example", "text": "Hello world"}]})
assert result == "# Example\n\nHello world"
assert mock_request.call_args.args == ("POST", "https://api.specstream.ai/v1/extract")
assert mock_request.call_args.kwargs["headers"]["Authorization"] == "Bearer test-specstream-key"
assert mock_request.call_args.kwargs["json"] == {"urls": ["https://example.com/post"], "format": "markdown"}
def test_truncates_long_content(self, mock_config_with_key):
result, _ = self._fetch({"results": [{"status": "ok", "title": "T", "text": "x" * 10000}]})
assert len(result) == len("# T\n\n") + 4096
def test_falls_back_to_untitled(self, mock_config_with_key):
result, _ = self._fetch({"results": [{"status": "ok", "text": "body"}]})
assert result.startswith("# Untitled\n\n")
@pytest.mark.parametrize("status", ["blocked", "not_found", "empty", "unsupported", "invalid_url", "failed"])
def test_page_that_was_not_read_returns_its_status(self, mock_config_with_key, status):
result, _ = self._fetch({"results": [{"status": status, "text": ""}]})
assert result == f"Error: SpecStream could not read the page ({status})"
def test_ok_page_without_text_returns_error(self, mock_config_with_key):
result, _ = self._fetch({"results": [{"status": "ok", "text": None}]})
assert result == "Error: No content found"
def test_no_results_returns_error(self, mock_config_with_key):
result, _ = self._fetch({"results": []})
assert result == "Error: No results found"
def test_results_without_any_objects_returns_format_error(self, mock_config_with_key):
result, _ = self._fetch({"results": ["oops"]})
assert result == "Error: SpecStream returned an unexpected response format"
def test_http_error_returns_error_string(self, mock_config_with_key):
request = httpx.Request("POST", "https://api.specstream.ai/v1/extract")
response = httpx.Response(402, text="balance is empty", request=request)
mock_resp = MagicMock()
mock_resp.raise_for_status.side_effect = httpx.HTTPStatusError("error", request=request, response=response)
with patch(f"{_MODULE}.httpx.Client") as mock_client_cls:
mock_client_cls.return_value.__enter__.return_value.request.return_value = mock_resp
from deerflow.community.specstream.tools import web_fetch_tool
result = web_fetch_tool.invoke({"url": "https://example.com"})
assert result == "Error: SpecStream API error: HTTP 402"
def test_missing_key_returns_error_string(self, mock_config_no_key):
with patch.dict("os.environ", {}, clear=True):
from deerflow.community.specstream.tools import web_fetch_tool
result = web_fetch_tool.invoke({"url": "https://example.com"})
assert result == "Error: SPECSTREAM_API_KEY is not configured"

View File

@ -1062,6 +1062,17 @@ tools:
# contents_max_characters: 2000 # per-result content cap; 0 disables it
# # api_key: $SOFYA_API_KEY # Optional explicit env-var reference
# Web search tool (uses SpecStream, requires SPECSTREAM_API_KEY)
# SpecStream answers from the live web: ranked results with a snippet each, plus a
# direct answer when the sources agree on one. Sign up at https://specstream.ai to get a key.
# Note: set SPECSTREAM_API_KEY in your environment before starting the app.
# Avoid putting literal API keys in config.yaml; use the $VAR form instead.
# - name: web_search
# group: web
# use: deerflow.community.specstream.tools:web_search_tool
# max_results: 5
# # api_key: $SPECSTREAM_API_KEY # Optional explicit env-var reference
# Web search tool (uses You.com; YDC_API_KEY optional)
# You.com runs its own web index and returns extracted page snippets rather
# than a single meta description, so results are usable without a follow-up
@ -1295,6 +1306,15 @@ tools:
# use: deerflow.community.sofya.tools:web_fetch_tool
# # api_key: $SOFYA_API_KEY
# Web fetch tool (uses SpecStream, requires SPECSTREAM_API_KEY)
# Returns the page as markdown. Also reads PDF, Word and Excel files.
# NOTE: Only one web_fetch provider can be active at a time.
# Comment out the Jina AI web_fetch entry above before enabling this one.
# - name: web_fetch
# group: web
# use: deerflow.community.specstream.tools:web_fetch_tool
# # api_key: $SPECSTREAM_API_KEY
# Web fetch tool (uses Unbrowse, requires UNBROWSE_API_KEY)
# Returns the page as markdown. Fetches over plain HTTP when that is enough and
# falls back to a hosted cloud browser for pages that need JavaScript.

View File

@ -127,7 +127,7 @@ Community tools connect the agent to external services. Each `tools:` entry in `
### Web search
<Tabs items={["DuckDuckGo (default)", "Tavily", "Brave", "Serply", "Exa", "InfoQuest", "Firecrawl", "GroundRoute", "Sofya", "You.com"]}>
<Tabs items={["DuckDuckGo (default)", "Tavily", "Brave", "Serply", "Exa", "InfoQuest", "Firecrawl", "GroundRoute", "Sofya", "SpecStream", "You.com"]}>
<Tabs.Tab>
```yaml
tools:
@ -240,6 +240,16 @@ Results include the content of each page, not only a snippet. Requires a [Sofya]
</Tabs.Tab>
<Tabs.Tab>
```yaml
tools:
- name: web_search
group: web
use: deerflow.community.specstream.tools:web_search_tool
api_key: $SPECSTREAM_API_KEY
```
Results carry a snippet and a source URL, and a direct answer when the sources agree on one. Requires a [SpecStream](https://specstream.ai) API key. Also supports `web_fetch`.
</Tabs.Tab>
<Tabs.Tab>
```yaml
tools:
- name: web_search
group: web
@ -260,7 +270,7 @@ No extra dependency required.
### Web fetch (page content extraction)
<Tabs items={["Jina AI (default)", "Browserless", "Exa", "InfoQuest", "Firecrawl", "GroundRoute", "Sofya", "Unbrowse"]}>
<Tabs items={["Jina AI (default)", "Browserless", "Exa", "InfoQuest", "Firecrawl", "GroundRoute", "Sofya", "SpecStream", "Unbrowse"]}>
<Tabs.Tab>
```yaml
tools:
@ -331,6 +341,16 @@ Returns the page as markdown. Requires a [Sofya](https://sofya.co) API key.
</Tabs.Tab>
<Tabs.Tab>
```yaml
tools:
- name: web_fetch
group: web
use: deerflow.community.specstream.tools:web_fetch_tool
api_key: $SPECSTREAM_API_KEY
```
Returns the page as markdown, including PDF, Word and Excel files. Requires a [SpecStream](https://specstream.ai) API key.
</Tabs.Tab>
<Tabs.Tab>
```yaml
tools:
- name: web_fetch
group: web

View File

@ -124,7 +124,7 @@ tools:
### 网络搜索
<Tabs items={["DuckDuckGo(默认)", "Tavily", "Brave", "Serply", "Exa", "InfoQuest", "Firecrawl", "GroundRoute", "Sofya", "You.com"]}>
<Tabs items={["DuckDuckGo(默认)", "Tavily", "Brave", "Serply", "Exa", "InfoQuest", "Firecrawl", "GroundRoute", "Sofya", "SpecStream", "You.com"]}>
<Tabs.Tab>
```yaml
tools:
@ -234,6 +234,16 @@ tools:
</Tabs.Tab>
<Tabs.Tab>
```yaml
tools:
- name: web_search
group: web
use: deerflow.community.specstream.tools:web_search_tool
api_key: $SPECSTREAM_API_KEY
```
搜索结果包含摘要和来源 URL,来源一致时还会给出直接答案。需要 [SpecStream](https://specstream.ai) API Key。同时支持 `web_fetch`。
</Tabs.Tab>
<Tabs.Tab>
```yaml
tools:
- name: web_search
group: web
@ -251,7 +261,7 @@ API Key 可选,可在 [You.com](https://you.com/platform) 获取。
### 网页内容抓取
<Tabs items={["Jina AI(默认)", "Browserless", "Exa", "InfoQuest", "Firecrawl", "GroundRoute", "Sofya", "Unbrowse"]}>
<Tabs items={["Jina AI(默认)", "Browserless", "Exa", "InfoQuest", "Firecrawl", "GroundRoute", "Sofya", "SpecStream", "Unbrowse"]}>
<Tabs.Tab>
```yaml
tools:
@ -321,6 +331,16 @@ tools:
</Tabs.Tab>
<Tabs.Tab>
```yaml
tools:
- name: web_fetch
group: web
use: deerflow.community.specstream.tools:web_fetch_tool
api_key: $SPECSTREAM_API_KEY
```
以 Markdown 形式返回页面内容,也支持 PDF、Word 和 Excel 文件。需要 [SpecStream](https://specstream.ai) API Key。
</Tabs.Tab>
<Tabs.Tab>
```yaml
tools:
- name: web_fetch
group: web

View File

@ -691,6 +691,7 @@ def check_web_tool(config_path: Path, *, tool_name: str, label: str) -> CheckRes
"serper": "SERPER_API_KEY",
"serply": "SERPLY_API_KEY",
"sofya": "SOFYA_API_KEY",
"specstream": "SPECSTREAM_API_KEY",
"tencent_wsa": "TENCENTCLOUD_WSA_APIKEY",
"webz": "WEBZ_API_KEY",
},
@ -700,6 +701,7 @@ def check_web_tool(config_path: Path, *, tool_name: str, label: str) -> CheckRes
"firecrawl": "FIRECRAWL_API_KEY",
"fastcrw": "CRW_API_KEY",
"sofya": "SOFYA_API_KEY",
"specstream": "SPECSTREAM_API_KEY",
"unbrowse": "UNBROWSE_API_KEY",
},
"image_search": {

View File

@ -734,6 +734,14 @@ SEARCH_PROVIDERS: list[SearchProvider] = [
env_var=None,
extra_config={"max_results": 5},
),
SearchProvider(
name="specstream",
display_name="SpecStream",
description="Answers from the live web with a source for each result, API key required",
use="deerflow.community.specstream.tools:web_search_tool",
env_var="SPECSTREAM_API_KEY",
extra_config={"max_results": 5},
),
]
WEB_FETCH_PROVIDERS: list[WebProvider] = [
@ -812,4 +820,12 @@ WEB_FETCH_PROVIDERS: list[WebProvider] = [
tool_name="web_fetch",
extra_config={"base_url": "http://localhost:11235", "timeout": 30},
),
WebProvider(
name="specstream",
display_name="SpecStream",
description="Page text as markdown, also PDF, Word and Excel, API key required",
use="deerflow.community.specstream.tools:web_fetch_tool",
env_var="SPECSTREAM_API_KEY",
tool_name="web_fetch",
),
]