seb/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore(web): remove web_crawl tool + provider crawl plumbing (#33824)

Browse Source 
The web_crawl_tool() function was an orphan — no model schema registered
it, no skill or CLI command called it, and the agent had no way to invoke
it. PR #32608 proposed wiring it up as a model-callable tool; we've
decided not to expose crawl as a separate capability since web_search +
web_extract cover the use cases we want models to have.

Removed:
- tools/web_tools.py: web_crawl_tool() (~230 LOC)
- plugins/web/firecrawl/provider.py: supports_crawl() + crawl()
- plugins/web/tavily/provider.py: supports_crawl() + crawl()
- plugins/web/xai/provider.py: supports_crawl() override
- agent/web_search_provider.py: supports_crawl() + crawl() ABC methods
- agent/web_search_registry.py: get_active_crawl_provider() +
  the 'crawl' branch in _resolve()
- agent/display.py: web_crawl tool-progress rendering
- hermes_cli/config.py: 'web_crawl' from TAVILY_API_KEY.tools
- tools/website_policy.py: stale comment reference
- Tests: removed TestWebCrawlTavily class, the two website-policy
  web_crawl tests, the searxng/ddgs/brave-free crawl-error tests,
  the integration test_web_crawl method, and the
  test_unconfigured_crawl_emits_top_level_error test. Trimmed the
  capability-flag parametrize list and the WebSearchProvider ABC
  conformance tests.
- Docs: trimmed the Crawl column from capability tables in both EN
  and zh-Hans, updated the developer-guide ABC table.

Net: 25 files, +115/-1067.

Closes #33762 (the schema-text bug only existed if #32608 landed).
Supersedes #32608.
This commit is contained in:
Teknium
2026-05-28 04:52:42 -07:00
committed by GitHub
parent b243afb68b
commit 5e1f793430
25 changed files with 115 additions and 1067 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/integration/test_web_tools.py
-111
View File
@@ -30,7 +30,6 @@ from typing import List
from tools.web_tools import (
web_search_tool,
web_extract_tool,
web_crawl_tool,
check_firecrawl_api_key,
check_web_api_key,
check_auxiliary_model,
@@ -404,113 +403,6 @@ class WebToolsTester:
except Exception as e:
self.log_result("Extract (with LLM)", "failed", str(e))
async def test_web_crawl(self):
"""Test web crawling functionality"""
print_section("Test 4: Web Crawl")
test_sites = [
("https://docs.firecrawl.dev", None, 2), # Test docs site
("https://firecrawl.dev", None, 3), # Test main site
]
for url, instructions, expected_min_pages in test_sites:
try:
print(f"\n Testing crawl of: {url}")
if instructions:
print(f" Instructions: {instructions}")
else:
print(f" No instructions (general crawl)")
print(f" Expected minimum pages: {expected_min_pages}")
# Show what's being called
if self.verbose:
print(f" Calling web_crawl_tool(url='{url}', instructions={instructions}, use_llm_processing=False)")
result = await web_crawl_tool(
url,
instructions=instructions,
use_llm_processing=False # Disable LLM for faster testing
)
# Check if result is valid JSON
try:
data = json.loads(result)
except json.JSONDecodeError as e:
self.log_result(f"Crawl: {url}", "failed", f"Invalid JSON response: {e}")
if self.verbose:
print(f" Raw response (first 500 chars): {result[:500]}...")
continue
# Check for errors
if "error" in data:
self.log_result(f"Crawl: {url}", "failed", f"API error: {data['error']}")
continue
# Get results
results = data.get("results", [])
if not results:
self.log_result(f"Crawl: {url}", "failed", "No pages in results array")
if self.verbose:
print(f" Full response: {json.dumps(data, indent=2)[:1000]}...")
continue
# Analyze pages
valid_pages = 0
empty_pages = 0
total_content = 0
page_details = []
for i, page in enumerate(results):
content = page.get("content", "")
title = page.get("title", "Untitled")
error = page.get("error")
if error:
page_details.append(f"Page {i+1}: ERROR - {error}")
elif content:
valid_pages += 1
content_len = len(content)
total_content += content_len
page_details.append(f"Page {i+1}: {title[:40]}... ({content_len} chars)")
else:
empty_pages += 1
page_details.append(f"Page {i+1}: {title[:40]}... (EMPTY)")
# Show detailed results if verbose
if self.verbose:
print(f"\n Crawl Results:")
print(f" Total pages returned: {len(results)}")
print(f" Valid pages (with content): {valid_pages}")
print(f" Empty pages: {empty_pages}")
print(f" Total content size: {total_content} characters")
print(f"\n Page Details:")
for detail in page_details[:10]: # Show first 10 pages
print(f" - {detail}")
if len(page_details) > 10:
print(f" ... and {len(page_details) - 10} more pages")
# Determine pass/fail
if valid_pages >= expected_min_pages:
self.log_result(
f"Crawl: {url}",
"passed",
f"{valid_pages}/{len(results)} valid pages, {total_content} chars total"
)
else:
self.log_result(
f"Crawl: {url}",
"failed",
f"Only {valid_pages} valid pages (expected >= {expected_min_pages}), {empty_pages} empty, {len(results)} total"
)
except Exception as e:
self.log_result(f"Crawl: {url}", "failed", f"Exception: {type(e).__name__}: {str(e)}")
if self.verbose:
import traceback
print(f" Traceback:")
print(" " + "\n ".join(traceback.format_exc().split("\n")))
async def run_all_tests(self):
"""Run all tests"""
self.start_time = datetime.now()
@@ -533,9 +425,6 @@ class WebToolsTester:
if self.test_llm:
await self.test_web_extract_with_llm(urls if urls else None)
# Test crawling
await self.test_web_crawl()
# Print summary
self.end_time = datetime.now()
duration = (self.end_time - self.start_time).total_seconds()
tests/plugins/web/test_web_search_provider_plugins.py
+9 -46
View File
@@ -90,20 +90,17 @@ class TestBundledPluginsRegister:
]
@pytest.mark.parametrize(
"plugin_name,expected_search,expected_extract,expected_crawl",
"plugin_name,expected_search,expected_extract",
[
("brave-free", True, False, False),
("ddgs", True, False, False),
("searxng", True, False, False),
("exa", True, True, False),
("parallel", True, True, False),
("tavily", True, True, True),
# firecrawl: search + extract + crawl. Crawl was originally
# disabled in the migration (fell through to a legacy inline
# path); the follow-up commit enabled it natively.
("firecrawl", True, True, True),
("brave-free", True, False),
("ddgs", True, False),
("searxng", True, False),
("exa", True, True),
("parallel", True, True),
("tavily", True, True),
("firecrawl", True, True),
# xai: search-only via Grok's agentic web_search tool.
("xai", True, False, False),
("xai", True, False),
],
)
def test_capability_flags_match_spec(
@@ -111,7 +108,6 @@ class TestBundledPluginsRegister:
plugin_name: str,
expected_search: bool,
expected_extract: bool,
expected_crawl: bool,
) -> None:
_ensure_plugins_loaded()
from agent.web_search_registry import get_provider
@@ -120,7 +116,6 @@ class TestBundledPluginsRegister:
assert provider is not None, f"plugin {plugin_name!r} not registered"
assert provider.supports_search() is expected_search
assert provider.supports_extract() is expected_extract
assert provider.supports_crawl() is expected_crawl
@pytest.mark.parametrize(
"plugin_name",
@@ -457,38 +452,6 @@ class TestErrorResponseShapes:
if result: # if anything came back, it should be an error entry
assert "error" in result[0]
def test_tavily_crawl_returns_error_dict_when_unconfigured(self) -> None:
_ensure_plugins_loaded()
from agent.web_search_registry import get_provider
p = get_provider("tavily")
assert p is not None
result = p.crawl("https://example.com")
assert isinstance(result, dict)
assert "results" in result
assert isinstance(result["results"], list)
if result["results"]:
assert "error" in result["results"][0]
def test_firecrawl_crawl_returns_error_dict_when_unconfigured(self):
"""firecrawl crawl is async (wraps SDK in to_thread); error must be
surfaced via the per-page result shape, not raised."""
_ensure_plugins_loaded()
from agent.web_search_registry import get_provider
p = get_provider("firecrawl")
assert p is not None
assert inspect.iscoroutinefunction(p.crawl)
result = asyncio.run(p.crawl("https://example.com"))
assert isinstance(result, dict)
assert "results" in result
assert isinstance(result["results"], list)
# Without FIRECRAWL_API_KEY, the plugin's _get_firecrawl_client()
# raises ValueError which is caught and returned as a per-page error.
assert len(result["results"]) >= 1
assert "error" in result["results"][0]
assert result["results"][0]["url"] == "https://example.com"
def test_firecrawl_config_error_points_paid_users_to_nous_subscription(self, monkeypatch):
from plugins.web.firecrawl import provider as firecrawl_provider
tests/tools/test_web_providers.py
+6 -37
View File
@@ -29,7 +29,7 @@ class TestWebProviderABCs:
in-tree ABCs at ``tools.web_providers.base`` (separate
``WebSearchProvider`` + ``WebExtractProvider``) were deleted in the
same PR — providers now advertise capabilities via
``supports_search() / supports_extract() / supports_crawl()`` flags.
``supports_search() / supports_extract()`` flags.
"""
def test_cannot_instantiate_abc_directly(self):
@@ -65,7 +65,6 @@ class TestWebProviderABCs:
assert d.is_available() is True
assert d.supports_search() is True
assert d.supports_extract() is False # default
assert d.supports_crawl() is False # default
assert d.search("test")["success"] is True
def test_concrete_multi_capability_provider_works(self):
@@ -89,27 +88,19 @@ class TestWebProviderABCs:
def supports_extract(self) -> bool:
return True
def supports_crawl(self) -> bool:
return True
def search(self, query: str, limit: int = 5) -> Dict[str, Any]:
return {"success": True, "data": {"web": []}}
def extract(self, urls: List[str], **kwargs: Any) -> List[Dict[str, Any]]:
return [{"url": urls[0], "content": "x"}]
def crawl(self, url: str, **kwargs: Any) -> Dict[str, Any]:
return {"results": [{"url": url, "content": "x"}]}
d = Dummy()
assert d.supports_search() is True
assert d.supports_extract() is True
assert d.supports_crawl() is True
assert d.extract(["https://example.com"])[0]["url"] == "https://example.com"
assert d.crawl("https://example.com")["results"][0]["url"] == "https://example.com"
def test_search_only_provider_skips_extract_and_crawl(self):
"""Search-only providers don't have to implement extract() / crawl()."""
def test_search_only_provider_skips_extract(self):
"""Search-only providers don't have to implement extract()."""
from agent.web_search_provider import WebSearchProvider
class SearchOnly(WebSearchProvider):
@@ -130,13 +121,12 @@ class TestWebProviderABCs:
def search(self, query: str, limit: int = 5) -> Dict[str, Any]:
return {"success": True, "data": {"web": []}}
# Should instantiate fine — extract/crawl have default
# supports_*() returning False and aren't required to be
# overridden when not advertised.
# Should instantiate fine — extract has default supports_*()
# returning False and isn't required to be overridden when not
# advertised.
s = SearchOnly()
assert s.supports_search() is True
assert s.supports_extract() is False
assert s.supports_crawl() is False
# ---------------------------------------------------------------------------
@@ -322,24 +312,3 @@ class TestUnconfiguredErrorEnvelopeParity:
# No per-result burying
assert "results" not in result
def test_unconfigured_crawl_emits_top_level_error(self, monkeypatch):
"""``web_crawl_tool`` with no creds returns ``{"success": False, "error": "web_crawl requires Firecrawl..."}``
— the dispatcher gates on ``provider.is_available()`` BEFORE
delegating to the plugin so pre-config errors don't get wrapped
into ``results[]``.
"""
import asyncio
import json
from tools import web_tools
self._clear_web_creds(monkeypatch)
monkeypatch.setattr(web_tools, "_firecrawl_client", None, raising=False)
monkeypatch.setattr(web_tools, "_firecrawl_client_config", None, raising=False)
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {})
result = json.loads(asyncio.run(web_tools.web_crawl_tool("https://example.com", use_llm_processing=False)))
assert result.get("success") is False
assert "error" in result, f"expected top-level 'error' key, got {result}"
assert "web_crawl requires Firecrawl" in result["error"]
# Crucially: no per-page burying
assert "results" not in result
tests/tools/test_web_providers_brave_free.py
+2 -22
View File
@@ -8,7 +8,7 @@ Covers:
- _is_backend_available("brave-free") integration
- _get_backend() recognizes "brave-free" as a valid configured backend
- check_web_api_key() includes brave-free in availability check
- web_extract / web_crawl return search-only errors when brave-free is active
- web_extract returns a search-only error when brave-free is active
"""
from __future__ import annotations
@@ -238,7 +238,7 @@ class TestBraveFreeBackendWiring:
# ---------------------------------------------------------------------------
# brave-free is search-only: web_extract / web_crawl return clear errors
# brave-free is search-only: web_extract returns a clear error
# ---------------------------------------------------------------------------
@@ -269,23 +269,3 @@ class TestBraveFreeSearchOnlyErrors:
assert result["success"] is False
assert "search-only" in result["error"].lower()
assert "brave" in result["error"].lower()
def test_web_crawl_returns_search_only_error(self, monkeypatch):
import asyncio
from tools import web_tools
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {"backend": "brave-free"})
monkeypatch.setenv("BRAVE_SEARCH_API_KEY", "BSAkey123")
monkeypatch.setattr(web_tools, "_is_tool_gateway_ready", lambda: False)
monkeypatch.setattr(web_tools, "check_firecrawl_api_key", lambda: False)
monkeypatch.setattr(web_tools, "is_safe_url", lambda url: True)
monkeypatch.setattr(web_tools, "check_website_access", lambda url: None)
monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False, raising=False)
result_str = asyncio.get_event_loop().run_until_complete(
web_tools.web_crawl_tool("https://example.com")
)
result = json.loads(result_str)
assert result["success"] is False
assert "search-only" in result["error"].lower()
assert "brave" in result["error"].lower()
tests/tools/test_web_providers_ddgs.py
+2 -22
View File
@@ -5,7 +5,7 @@ Covers:
- DDGSWebSearchProvider.search() — happy path, missing package, runtime error
- Result normalization (title, url, description, position)
- _is_backend_available("ddgs") / _get_backend() integration
- web_extract / web_crawl return search-only errors when ddgs is active
- web_extract returns a search-only error when ddgs is active
"""
from __future__ import annotations
@@ -209,7 +209,7 @@ class TestDDGSBackendWiring:
# ---------------------------------------------------------------------------
# ddgs is search-only: web_extract / web_crawl return clear errors
# ddgs is search-only: web_extract returns a clear error
# ---------------------------------------------------------------------------
@@ -240,23 +240,3 @@ class TestDDGSSearchOnlyErrors:
assert result["success"] is False
assert "search-only" in result["error"].lower()
assert "duckduckgo" in result["error"].lower() or "ddgs" in result["error"].lower()
def test_web_crawl_returns_search_only_error(self, monkeypatch):
import asyncio
from tools import web_tools
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {"backend": "ddgs"})
monkeypatch.setattr(web_tools, "_ddgs_package_importable", lambda: True)
monkeypatch.setattr(web_tools, "_is_tool_gateway_ready", lambda: False)
monkeypatch.setattr(web_tools, "check_firecrawl_api_key", lambda: False)
monkeypatch.setattr(web_tools, "is_safe_url", lambda url: True)
monkeypatch.setattr(web_tools, "check_website_access", lambda url: None)
monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False, raising=False)
result_str = asyncio.get_event_loop().run_until_complete(
web_tools.web_crawl_tool("https://example.com")
)
result = json.loads(result_str)
assert result["success"] is False
assert "search-only" in result["error"].lower()
assert "duckduckgo" in result["error"].lower() or "ddgs" in result["error"].lower()
tests/tools/test_web_providers_searxng.py
+1 -21
View File
@@ -296,7 +296,7 @@ class TestCheckWebApiKey:
# ---------------------------------------------------------------------------
# searxng-only: web_extract and web_crawl return clear errors
# searxng-only: web_extract returns a clear error
# ---------------------------------------------------------------------------
@@ -312,26 +312,6 @@ class TestSearXNGOnlyExtractCrawlErrors:
from agent.web_search_registry import _reset_for_tests
_reset_for_tests()
def test_web_crawl_searxng_returns_clear_error(self, monkeypatch):
import asyncio
from tools import web_tools
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {"backend": "searxng"})
monkeypatch.setenv("SEARXNG_URL", "http://localhost:8080")
monkeypatch.setattr(web_tools, "_is_tool_gateway_ready", lambda: False)
monkeypatch.setattr(web_tools, "check_firecrawl_api_key", lambda: False)
monkeypatch.setattr(web_tools, "is_safe_url", lambda url: True)
monkeypatch.setattr(web_tools, "check_website_access", lambda url: None)
monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False, raising=False)
import json
result_str = asyncio.get_event_loop().run_until_complete(
web_tools.web_crawl_tool("https://example.com")
)
result = json.loads(result_str)
assert result["success"] is False
assert "search-only" in result["error"].lower() or "SearXNG" in result["error"]
def test_web_extract_searxng_returns_clear_error(self, monkeypatch):
import asyncio
from tools import web_tools
tests/tools/test_web_providers_xai.py
-1
View File
@@ -66,7 +66,6 @@ class TestXAIProviderIdentity:
p = XAIWebSearchProvider()
assert p.supports_search() is True
assert p.supports_extract() is False
assert p.supports_crawl() is False
def test_display_name(self):
from plugins.web.xai.provider import XAIWebSearchProvider
tests/tools/test_web_tools_tavily.py
+2 -61
View File
@@ -3,8 +3,8 @@
Coverage:
_tavily_request() — API key handling, endpoint construction, error propagation.
_normalize_tavily_search_results() — search response normalization.
_normalize_tavily_documents() — extract/crawl response normalization, failed_results.
web_search_tool / web_extract_tool / web_crawl_tool — Tavily dispatch paths.
_normalize_tavily_documents() — extract response normalization, failed_results.
web_search_tool / web_extract_tool — Tavily dispatch paths.
"""
import json
@@ -225,62 +225,3 @@ class TestWebExtractTavily:
assert len(result["results"]) == 1
assert result["results"][0]["url"] == "https://example.com"
# ─── web_crawl_tool (Tavily dispatch) ─────────────────────────────────────────
class TestWebCrawlTavily:
"""Test web_crawl_tool dispatch to Tavily."""
_register_providers = staticmethod(register_all_web_providers)
@pytest.fixture(autouse=True)
def _populate_web_registry(self):
self._register_providers()
yield
from agent.web_search_registry import _reset_for_tests
_reset_for_tests()
def test_crawl_dispatches_to_tavily(self):
mock_response = MagicMock()
mock_response.json.return_value = {
"results": [
{"url": "https://example.com/page1", "raw_content": "Page 1 content", "title": "Page 1"},
{"url": "https://example.com/page2", "raw_content": "Page 2 content", "title": "Page 2"},
]
}
mock_response.raise_for_status = MagicMock()
with patch("tools.web_tools._get_backend", return_value="tavily"), \
patch.dict(os.environ, {"TAVILY_API_KEY": "tvly-test"}), \
patch("tools.web_tools.httpx.post", return_value=mock_response), \
patch("tools.web_tools.check_website_access", return_value=None), \
patch("tools.web_tools.is_safe_url", return_value=True), \
patch("tools.interrupt.is_interrupted", return_value=False):
from tools.web_tools import web_crawl_tool
result = json.loads(asyncio.get_event_loop().run_until_complete(
web_crawl_tool("https://example.com", use_llm_processing=False)
))
assert "results" in result
assert len(result["results"]) == 2
assert result["results"][0]["title"] == "Page 1"
def test_crawl_sends_instructions(self):
"""Instructions are included in the Tavily crawl payload."""
mock_response = MagicMock()
mock_response.json.return_value = {"results": []}
mock_response.raise_for_status = MagicMock()
with patch("tools.web_tools._get_backend", return_value="tavily"), \
patch.dict(os.environ, {"TAVILY_API_KEY": "tvly-test"}), \
patch("tools.web_tools.httpx.post", return_value=mock_response) as mock_post, \
patch("tools.web_tools.check_website_access", return_value=None), \
patch("tools.web_tools.is_safe_url", return_value=True), \
patch("tools.interrupt.is_interrupted", return_value=False):
from tools.web_tools import web_crawl_tool
asyncio.get_event_loop().run_until_complete(
web_crawl_tool("https://example.com", instructions="Find docs", use_llm_processing=False)
)
call_kwargs = mock_post.call_args
payload = call_kwargs.kwargs.get("json") or call_kwargs[1].get("json")
assert payload["instructions"] == "Find docs"
assert payload["url"] == "https://example.com"
tests/tools/test_website_policy.py
+2 -93
View File
@@ -350,7 +350,7 @@ def test_browser_navigate_allows_when_shared_file_missing(monkeypatch, tmp_path)
class TestWebToolPolicy:
"""Tests that exercise web_extract_tool / web_crawl_tool with website-policy gates.
"""Tests that exercise web_extract_tool with website-policy gates.
These tests need the bundled web providers to be registered in the
agent.web_search_registry so the tool dispatchers can find an active
@@ -376,8 +376,7 @@ class TestWebToolPolicy:
monkeypatch.setattr(web_tools, "is_safe_url", lambda url: True)
# The per-URL website-policy gate moved into the firecrawl plugin's
# extract() during the web-provider migration. Patch it at the new
# location; the dispatcher-level gate (used by web_crawl_tool's
# pre-flight) still lives on tools.web_tools.
# location.
monkeypatch.setattr(
firecrawl_provider,
"check_website_access",
@@ -445,96 +444,6 @@ class TestWebToolPolicy:
assert result["results"][0]["content"] == ""
assert result["results"][0]["blocked_by_policy"]["rule"] == "blocked.test"
@pytest.mark.asyncio
async def test_web_crawl_short_circuits_blocked_url(self, monkeypatch):
from tools import web_tools
# web_crawl_tool checks for Firecrawl env before website policy
monkeypatch.setenv("FIRECRAWL_API_KEY", "fake-key")
# Allow test URLs past SSRF check so website policy is what gets tested
monkeypatch.setattr(web_tools, "is_safe_url", lambda url: True)
# The dispatcher-level (seed-URL) policy gate still lives on web_tools.
# No per-page gate runs in this test because the dispatcher returns
# immediately when the seed is blocked, before delegating to the plugin.
monkeypatch.setattr(
web_tools,
"check_website_access",
lambda url: {
"host": "blocked.test",
"rule": "blocked.test",
"source": "config",
"message": "Blocked by website policy",
},
)
# If the dispatcher ever reaches the firecrawl plugin's crawl(), the test
# fails — pin the plugin module's client lookup so we'd notice.
from plugins.web.firecrawl import provider as firecrawl_provider
monkeypatch.setattr(
firecrawl_provider,
"_get_firecrawl_client",
lambda: pytest.fail("firecrawl plugin should not run for blocked crawl URL"),
)
monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False)
result = json.loads(await web_tools.web_crawl_tool("https://blocked.test", use_llm_processing=False))
assert result["results"][0]["url"] == "https://blocked.test"
assert result["results"][0]["blocked_by_policy"]["rule"] == "blocked.test"
@pytest.mark.asyncio
async def test_web_crawl_blocks_redirected_final_url(self, monkeypatch):
from tools import web_tools
from plugins.web.firecrawl import provider as firecrawl_provider
# Force the firecrawl plugin to be the active crawl provider.
monkeypatch.setenv("FIRECRAWL_API_KEY", "fake-key")
# Allow test URLs past SSRF check so website policy is what gets tested
monkeypatch.setattr(web_tools, "is_safe_url", lambda url: True)
def fake_check(url):
# Dispatcher seed-URL gate (web_tools.check_website_access call)
# and plugin per-page gate (firecrawl_provider.check_website_access
# call) both flow through this single fake_check.
if url == "https://allowed.test":
return None
if url == "https://blocked.test/final":
return {
"host": "blocked.test",
"rule": "blocked.test",
"source": "config",
"message": "Blocked by website policy",
}
pytest.fail(f"unexpected URL checked: {url}")
class FakeCrawlClient:
def crawl(self, url, **kwargs):
return {
"data": [
{
"markdown": "secret crawl content",
"metadata": {
"title": "Redirected crawl page",
"sourceURL": "https://blocked.test/final",
},
}
]
}
# After PR #25182 follow-up: per-page policy gate lives in
# plugins.web.firecrawl.provider.crawl(). Patch the gate + client at
# the plugin location. The dispatcher-level (seed) gate also reads
# web_tools.check_website_access — patch both.
monkeypatch.setattr(web_tools, "check_website_access", fake_check)
monkeypatch.setattr(firecrawl_provider, "check_website_access", fake_check)
monkeypatch.setattr(firecrawl_provider, "_get_firecrawl_client", lambda: FakeCrawlClient())
monkeypatch.setattr("tools.interrupt.is_interrupted", lambda: False)
result = json.loads(await web_tools.web_crawl_tool("https://allowed.test", use_llm_processing=False))
assert result["results"][0]["content"] == ""
assert result["results"][0]["error"] == "Blocked by website policy"
assert result["results"][0]["blocked_by_policy"]["rule"] == "blocked.test"
def test_check_website_access_fails_open_on_malformed_config(tmp_path, monkeypatch):
"""Malformed config with default path should fail open (return None), not crash."""