diff --git a/backend/packages/harness/deerflow/community/browserless/tools.py b/backend/packages/harness/deerflow/community/browserless/tools.py
index aff213df0..2e2774073 100644
--- a/backend/packages/harness/deerflow/community/browserless/tools.py
+++ b/backend/packages/harness/deerflow/community/browserless/tools.py
@@ -1,9 +1,7 @@
 import asyncio
-import ipaddress
 import logging
 import os
 import re
-import socket
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import Annotated
@@ -13,6 +11,8 @@ from langchain.tools import InjectedToolCallId, tool
 from langchain_core.messages import ToolMessage
 from langgraph.types import Command
 
+from deerflow.community.url_safety import resolve_host_addresses as _resolve_host_addresses
+from deerflow.community.url_safety import validate_public_http_url
 from deerflow.config import get_app_config
 from deerflow.config.paths import VIRTUAL_PATH_PREFIX
 from deerflow.tools.types import Runtime
@@ -31,8 +31,6 @@ _OUTPUT_FORMAT_TO_EXTENSION = {
     "webp": "webp",
 }
 _SAFE_FILENAME_RE = re.compile(r"[^A-Za-z0-9._-]+")
-# Hostnames that always resolve to a loopback/link-local/cloud-metadata target.
-_BLOCKED_HOSTNAMES = {"localhost", "metadata.google.internal"}
 # Cap collision-suffix probing so a saturated outputs directory cannot spin forever.
 _MAX_FILENAME_COLLISION_PROBES = 1000
 
@@ -94,31 +92,6 @@ def _normalize_output_format(value: object) -> str:
     return output_format if output_format in _OUTPUT_FORMAT_TO_EXTENSION else "png"
 
 
-def _resolve_host_addresses(hostname: str) -> list[ipaddress._BaseAddress]:
-    """Resolve a hostname to all of its IP addresses for SSRF screening.
-
-    Returns an empty list when resolution fails so callers can decide how to
-    treat an unresolvable host.
-    """
-    addresses: list[ipaddress._BaseAddress] = []
-    try:
-        infos = socket.getaddrinfo(hostname, None)
-    except (socket.gaierror, UnicodeError):
-        return addresses
-    for info in infos:
-        sockaddr = info[4]
-        try:
-            addresses.append(ipaddress.ip_address(sockaddr[0]))
-        except ValueError:
-            continue
-    return addresses
-
-
-def _is_blocked_address(address: ipaddress._BaseAddress) -> bool:
-    """Return True for addresses that must never be reachable via this tool."""
-    return address.is_private or address.is_loopback or address.is_link_local or address.is_reserved or address.is_multicast or address.is_unspecified
-
-
 def _validate_capture_url(url: str, allow_private_addresses: bool = False) -> str | None:
     """Validate a capture URL for scheme and (unless opted out) SSRF safety.
 
@@ -127,39 +100,12 @@ def _validate_capture_url(url: str, allow_private_addresses: bool = False) -> st
     unspecified addresses. Operators who intentionally point the tool at an
     internal Browserless target can opt out via ``allow_private_addresses``.
     """
-    parsed = urlparse(url)
-    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
-        return "Error: Only http:// and https:// URLs are supported"
-
-    if allow_private_addresses:
-        return None
-
-    hostname = parsed.hostname
-    if not hostname:
-        return "Error: URL host could not be parsed"
-
-    normalized_host = hostname.strip().rstrip(".").lower()
-    if normalized_host in _BLOCKED_HOSTNAMES:
-        return "Error: Refusing to capture a private or loopback address"
-
-    # A literal IP host is screened directly; a name is screened across every
-    # address it resolves to, so a DNS record pointing at an internal IP is
-    # rejected rather than blindly fetched.
-    try:
-        literal_ip = ipaddress.ip_address(normalized_host)
-    except ValueError:
-        literal_ip = None
-
-    if literal_ip is not None:
-        candidates = [literal_ip]
-    else:
-        candidates = _resolve_host_addresses(hostname)
-        if not candidates:
-            return "Error: URL host could not be resolved"
-
-    if any(_is_blocked_address(addr) for addr in candidates):
-        return "Error: Refusing to capture a private, loopback, or metadata address"
-    return None
+    return validate_public_http_url(
+        url,
+        allow_private_addresses=allow_private_addresses,
+        action="capture",
+        resolver=_resolve_host_addresses,
+    )
 
 
 def _default_capture_stem(url: str) -> str:
@@ -255,7 +201,15 @@ async def web_fetch_tool(url: str) -> str:
         url: The URL to fetch the contents of.
     """
     try:
-        cfg = _get_tool_config("web_fetch")
+        cfg = _get_tool_config("web_fetch") or {}
+        allow_private_addresses = _as_bool(cfg.get("allow_private_addresses"), False)
+        url_error = validate_public_http_url(
+            url,
+            allow_private_addresses=allow_private_addresses,
+            resolver=_resolve_host_addresses,
+        )
+        if url_error:
+            return url_error
 
         wait_for_event = ""
         wait_for_timeout_ms = 0
@@ -264,11 +218,10 @@ async def web_fetch_tool(url: str) -> str:
         reject_resource_types: list[str] | None = None
         reject_request_pattern: list[str] | None = None
 
-        if cfg is not None:
-            wait_for_event = cfg.get("wait_for_event", wait_for_event)
-            raw_wait = cfg.get("wait_for_timeout_ms", wait_for_timeout_ms)
-            wait_for_timeout_ms = int(raw_wait) if not isinstance(raw_wait, int) else raw_wait
-            wait_for_selector = cfg.get("wait_for_selector", wait_for_selector)
+        wait_for_event = cfg.get("wait_for_event", wait_for_event)
+        raw_wait = cfg.get("wait_for_timeout_ms", wait_for_timeout_ms)
+        wait_for_timeout_ms = int(raw_wait) if not isinstance(raw_wait, int) else raw_wait
+        wait_for_selector = cfg.get("wait_for_selector", wait_for_selector)
 
         client = _get_browserless_client("web_fetch")
         html = await client.fetch_html(
diff --git a/backend/packages/harness/deerflow/community/crawl4ai/tools.py b/backend/packages/harness/deerflow/community/crawl4ai/tools.py
index 5971c892e..fbea3dc09 100644
--- a/backend/packages/harness/deerflow/community/crawl4ai/tools.py
+++ b/backend/packages/harness/deerflow/community/crawl4ai/tools.py
@@ -2,6 +2,7 @@ import logging
 
 from langchain.tools import tool
 
+from deerflow.community.url_safety import validate_public_http_url
 from deerflow.config import get_app_config
 
 from .crawl4ai_client import Crawl4AiClient
@@ -42,6 +43,18 @@ def _coerce_timeout(value: object, default: int) -> float:
     return float(default)
 
 
+def _coerce_bool(value: object, default: bool) -> bool:
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, str):
+        normalized = value.strip().lower()
+        if normalized in {"1", "true", "yes", "on"}:
+            return True
+        if normalized in {"0", "false", "no", "off"}:
+            return False
+    return default
+
+
 def _coerce_filter(value: object) -> str:
     """Normalize and validate the markdown filter, falling back to the default.
 
@@ -86,6 +99,10 @@ async def web_fetch_tool(url: str) -> str:
     """
     try:
         cfg = _get_tool_config("web_fetch")  # read config once; pass the values down
+        allow_private_addresses = _coerce_bool(cfg.get("allow_private_addresses") if cfg is not None else None, False)
+        url_error = validate_public_http_url(url, allow_private_addresses=allow_private_addresses)
+        if url_error:
+            return url_error
         filter_mode = _coerce_filter(cfg.get("filter") if cfg is not None else None)
         client = _build_client(cfg)
         markdown = await client.fetch_markdown(url, filter_mode=filter_mode)
diff --git a/backend/packages/harness/deerflow/community/fastcrw/tools.py b/backend/packages/harness/deerflow/community/fastcrw/tools.py
index f38fc6319..5bcdc18af 100644
--- a/backend/packages/harness/deerflow/community/fastcrw/tools.py
+++ b/backend/packages/harness/deerflow/community/fastcrw/tools.py
@@ -4,6 +4,7 @@ import os
 from firecrawl import FirecrawlApp
 from langchain.tools import tool
 
+from deerflow.community.url_safety import validate_public_http_url
 from deerflow.config import get_app_config
 
 # fastCRW is a Firecrawl-compatible web data engine (single Rust binary; self-host
@@ -29,6 +30,23 @@ def _get_fastcrw_client(tool_name: str = "web_search") -> FirecrawlApp:
     return FirecrawlApp(api_key=api_key, api_url=base_url)  # type: ignore[arg-type]
 
 
+def _get_tool_config_extra(tool_name: str) -> dict:
+    config = get_app_config().get_tool_config(tool_name)
+    return dict(config.model_extra or {}) if config is not None else {}
+
+
+def _coerce_bool(value: object, default: bool) -> bool:
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, str):
+        normalized = value.strip().lower()
+        if normalized in {"1", "true", "yes", "on"}:
+            return True
+        if normalized in {"0", "false", "no", "off"}:
+            return False
+    return default
+
+
 @tool("web_search", parse_docstring=True)
 def web_search_tool(query: str) -> str:
     """Search the web.
@@ -73,6 +91,11 @@ def web_fetch_tool(url: str) -> str:
         url: The URL to fetch the contents of.
     """
     try:
+        cfg = _get_tool_config_extra("web_fetch")
+        allow_private_addresses = _coerce_bool(cfg.get("allow_private_addresses"), False)
+        url_error = validate_public_http_url(url, allow_private_addresses=allow_private_addresses)
+        if url_error:
+            return url_error
         client = _get_fastcrw_client("web_fetch")
         result = client.scrape(url, formats=["markdown"])
 
diff --git a/backend/packages/harness/deerflow/community/url_safety.py b/backend/packages/harness/deerflow/community/url_safety.py
new file mode 100644
index 000000000..0151a467f
--- /dev/null
+++ b/backend/packages/harness/deerflow/community/url_safety.py
@@ -0,0 +1,78 @@
+"""Shared URL safety checks for server-side web tools."""
+
+from __future__ import annotations
+
+import ipaddress
+import socket
+from collections.abc import Callable
+from urllib.parse import urlparse
+
+_BLOCKED_HOSTNAMES = {"localhost", "metadata.google.internal"}
+
+
+def resolve_host_addresses(hostname: str) -> list[ipaddress._BaseAddress]:
+    """Resolve a hostname to all IP addresses for SSRF screening."""
+    addresses: list[ipaddress._BaseAddress] = []
+    try:
+        infos = socket.getaddrinfo(hostname, None)
+    except (socket.gaierror, UnicodeError):
+        return addresses
+    for info in infos:
+        sockaddr = info[4]
+        try:
+            addresses.append(ipaddress.ip_address(sockaddr[0]))
+        except ValueError:
+            continue
+    return addresses
+
+
+def is_blocked_address(address: ipaddress._BaseAddress) -> bool:
+    """Return True for addresses web tools should not reach by default."""
+    return address.is_private or address.is_loopback or address.is_link_local or address.is_reserved or address.is_multicast or address.is_unspecified
+
+
+def validate_public_http_url(
+    url: str,
+    *,
+    allow_private_addresses: bool = False,
+    action: str = "fetch",
+    resolver: Callable[[str], list[ipaddress._BaseAddress]] | None = None,
+) -> str | None:
+    """Validate an http(s) URL before a server-side web tool fetches it.
+
+    Returns an ``"Error: ..."`` string when the URL should be rejected, or
+    ``None`` when the caller may proceed.  The check is intentionally conservative
+    for self-hosted fetch/render services because those services run inside the
+    deployment network and can otherwise reach cloud metadata or private hosts.
+    """
+    parsed = urlparse(url)
+    if parsed.scheme not in {"http", "https"} or not parsed.netloc:
+        return "Error: Only http:// and https:// URLs are supported"
+
+    if allow_private_addresses:
+        return None
+
+    hostname = parsed.hostname
+    if not hostname:
+        return "Error: URL host could not be parsed"
+
+    normalized_host = hostname.strip().rstrip(".").lower()
+    if normalized_host in _BLOCKED_HOSTNAMES:
+        return f"Error: Refusing to {action} a private or loopback address"
+
+    try:
+        literal_ip = ipaddress.ip_address(normalized_host)
+    except ValueError:
+        literal_ip = None
+
+    if literal_ip is not None:
+        candidates = [literal_ip]
+    else:
+        resolve = resolver or resolve_host_addresses
+        candidates = resolve(hostname)
+        if not candidates:
+            return "Error: URL host could not be resolved"
+
+    if any(is_blocked_address(addr) for addr in candidates):
+        return f"Error: Refusing to {action} a private, loopback, or metadata address"
+    return None
diff --git a/backend/tests/test_browserless_client.py b/backend/tests/test_browserless_client.py
index e1591b3f4..7c6bc9119 100644
--- a/backend/tests/test_browserless_client.py
+++ b/backend/tests/test_browserless_client.py
@@ -285,6 +285,41 @@ class TestBrowserlessTools:
 
         assert result.startswith("Error:")
 
+    @patch("deerflow.community.browserless.tools._get_browserless_client")
+    async def test_web_fetch_tool_rejects_metadata_ip(self, mock_get_client):
+        """web_fetch_tool blocks the cloud-metadata link-local endpoint."""
+        with patch("deerflow.community.browserless.tools._get_tool_config", return_value=None):
+            result = await tools.web_fetch_tool.ainvoke("http://169.254.169.254/latest/meta-data/")
+
+        assert "private, loopback, or metadata" in result
+        mock_get_client.assert_not_called()
+
+    @patch("deerflow.community.browserless.tools._get_browserless_client")
+    async def test_web_fetch_tool_rejects_dns_resolving_to_private(self, mock_get_client):
+        """web_fetch_tool blocks hostnames that resolve to internal IPs."""
+        with patch("deerflow.community.browserless.tools._get_tool_config", return_value=None):
+            with patch(
+                "deerflow.community.browserless.tools._resolve_host_addresses",
+                return_value=[ipaddress.ip_address("10.0.0.5")],
+            ):
+                result = await tools.web_fetch_tool.ainvoke("https://internal.example.com/")
+
+        assert "private, loopback, or metadata" in result
+        mock_get_client.assert_not_called()
+
+    @patch("deerflow.community.browserless.tools._get_browserless_client")
+    async def test_web_fetch_tool_allows_private_when_opted_in(self, mock_get_client):
+        """web_fetch_tool allows internal targets only when explicitly configured."""
+        mock_client = MagicMock()
+        mock_client.fetch_html = AsyncMock(return_value="<html><body><article><p>internal</p></article></body></html>")
+        mock_get_client.return_value = mock_client
+
+        with patch("deerflow.community.browserless.tools._get_tool_config", return_value={"allow_private_addresses": True}):
+            result = await tools.web_fetch_tool.ainvoke("http://10.0.0.5/dashboard")
+
+        assert "Error:" not in result
+        mock_client.fetch_html.assert_called_once()
+
     @patch("deerflow.community.browserless.tools._get_browserless_client")
     async def test_web_capture_tool_writes_artifact(self, mock_get_client, tmp_path):
         """web_capture_tool writes screenshots into thread outputs and presents the artifact."""
diff --git a/backend/tests/test_crawl4ai_tools.py b/backend/tests/test_crawl4ai_tools.py
index b1a4a57bc..ff6388aca 100644
--- a/backend/tests/test_crawl4ai_tools.py
+++ b/backend/tests/test_crawl4ai_tools.py
@@ -1,5 +1,6 @@
 """Tests for Crawl4AI community tools."""
 
+import ipaddress
 import json
 from unittest.mock import MagicMock, patch
 
@@ -215,6 +216,44 @@ class TestCrawl4AiTools:
 
         assert result.startswith("Error:")
 
+    @patch("deerflow.community.crawl4ai.tools._build_client")
+    async def test_web_fetch_tool_rejects_metadata_ip(self, mock_build):
+        from deerflow.community.crawl4ai import tools
+
+        with patch("deerflow.community.crawl4ai.tools._get_tool_config", return_value=None):
+            result = await tools.web_fetch_tool.ainvoke("http://169.254.169.254/latest/meta-data/")
+
+        assert "private, loopback, or metadata" in result
+        mock_build.assert_not_called()
+
+    @patch("deerflow.community.crawl4ai.tools._build_client")
+    async def test_web_fetch_tool_rejects_dns_resolving_to_private(self, mock_build):
+        from deerflow.community.crawl4ai import tools
+
+        with patch("deerflow.community.crawl4ai.tools._get_tool_config", return_value=None):
+            with patch(
+                "deerflow.community.url_safety.resolve_host_addresses",
+                return_value=[ipaddress.ip_address("10.0.0.5")],
+            ):
+                result = await tools.web_fetch_tool.ainvoke("https://internal.example.com/")
+
+        assert "private, loopback, or metadata" in result
+        mock_build.assert_not_called()
+
+    @patch("deerflow.community.crawl4ai.tools._build_client")
+    async def test_web_fetch_tool_allows_private_when_opted_in(self, mock_build):
+        from deerflow.community.crawl4ai import tools
+
+        mock_client = MagicMock()
+        mock_client.fetch_markdown = AsyncMock(return_value="# internal")
+        mock_build.return_value = mock_client
+
+        with patch("deerflow.community.crawl4ai.tools._get_tool_config", return_value={"allow_private_addresses": True}):
+            result = await tools.web_fetch_tool.ainvoke("http://10.0.0.5/dashboard")
+
+        assert result == "# internal"
+        mock_client.fetch_markdown.assert_called_once()
+
     @patch("deerflow.community.crawl4ai.tools._build_client")
     async def test_web_fetch_tool_reads_config_once(self, mock_build):
         """Config is read exactly once per invocation (no split read on hot-reload)."""
diff --git a/backend/tests/test_fastcrw_tools.py b/backend/tests/test_fastcrw_tools.py
index 9d63585e1..1248d71f0 100644
--- a/backend/tests/test_fastcrw_tools.py
+++ b/backend/tests/test_fastcrw_tools.py
@@ -1,5 +1,6 @@
 """Unit tests for the fastCRW community tools."""
 
+import ipaddress
 import json
 from unittest.mock import MagicMock, patch
 
@@ -121,3 +122,56 @@ class TestWebFetchTool:
         from deerflow.community.fastcrw.tools import web_fetch_tool
 
         assert web_fetch_tool.invoke({"url": "https://example.com"}) == "Error: scrape failed"
+
+    @patch.dict("os.environ", {}, clear=True)
+    @patch("deerflow.community.fastcrw.tools.FirecrawlApp")
+    @patch("deerflow.community.fastcrw.tools.get_app_config")
+    def test_fetch_rejects_metadata_ip(self, mock_get_app_config, mock_fastcrw_cls):
+        mock_get_app_config.return_value.get_tool_config.return_value = None
+
+        from deerflow.community.fastcrw.tools import web_fetch_tool
+
+        result = web_fetch_tool.invoke({"url": "http://169.254.169.254/latest/meta-data/"})
+
+        assert "private, loopback, or metadata" in result
+        mock_fastcrw_cls.assert_not_called()
+
+    @patch.dict("os.environ", {}, clear=True)
+    @patch("deerflow.community.fastcrw.tools.FirecrawlApp")
+    @patch("deerflow.community.fastcrw.tools.get_app_config")
+    def test_fetch_rejects_dns_resolving_to_private(self, mock_get_app_config, mock_fastcrw_cls):
+        mock_get_app_config.return_value.get_tool_config.return_value = None
+
+        from deerflow.community.fastcrw.tools import web_fetch_tool
+
+        with patch(
+            "deerflow.community.url_safety.resolve_host_addresses",
+            return_value=[ipaddress.ip_address("10.0.0.5")],
+        ):
+            result = web_fetch_tool.invoke({"url": "https://internal.example.com/"})
+
+        assert "private, loopback, or metadata" in result
+        mock_fastcrw_cls.assert_not_called()
+
+    @patch.dict("os.environ", {}, clear=True)
+    @patch("deerflow.community.fastcrw.tools.FirecrawlApp")
+    @patch("deerflow.community.fastcrw.tools.get_app_config")
+    def test_fetch_allows_private_when_opted_in(self, mock_get_app_config, mock_fastcrw_cls):
+        fetch_config = MagicMock()
+        fetch_config.model_extra = {"allow_private_addresses": True, "base_url": "http://localhost:3000"}
+        mock_get_app_config.return_value.get_tool_config.return_value = fetch_config
+
+        mock_scrape_result = MagicMock()
+        mock_scrape_result.markdown = "Private markdown"
+        mock_scrape_result.metadata = MagicMock(title="Private Page")
+        mock_fastcrw_cls.return_value.scrape.return_value = mock_scrape_result
+
+        from deerflow.community.fastcrw.tools import web_fetch_tool
+
+        result = web_fetch_tool.invoke({"url": "http://10.0.0.5/dashboard"})
+
+        assert result == "# Private Page\n\nPrivate markdown"
+        mock_fastcrw_cls.return_value.scrape.assert_called_once_with(
+            "http://10.0.0.5/dashboard",
+            formats=["markdown"],
+        )
diff --git a/config.example.yaml b/config.example.yaml
index 288c1f10c..733359be5 100644
--- a/config.example.yaml
+++ b/config.example.yaml
@@ -688,6 +688,7 @@ tools:
   #   base_url: http://localhost:3032    # Browserless instance URL (default: :3032; Docker: http://browserless:3000)
   #   # token: $BROWSERLESS_TOKEN        # API token (required for Browserless Cloud; optional for self-hosted)
   #   timeout_s: 30                       # Request timeout in seconds
+  #   # allow_private_addresses: false    # SSRF guard: keep false in production. Set true ONLY for intentional internal targets.
   #   # wait_for_event: "networkidle"    # Wait for a page event before returning (e.g. "load", "networkidle")
   #   # wait_for_timeout_ms: 2000        # Extra wait after page load in milliseconds
   #   # wait_for_selector: "article"     # CSS selector to wait for before returning
@@ -704,6 +705,7 @@ tools:
   #   use: deerflow.community.crawl4ai.tools:web_fetch_tool
   #   base_url: http://localhost:11235   # Crawl4AI server URL (Docker: http://crawl4ai:11235)
   #   timeout: 30                         # Request timeout in seconds
+  #   # allow_private_addresses: false     # SSRF guard: keep false in production. Set true ONLY for intentional internal targets.
   #   # filter: fit                       # Markdown filter: fit (default) | raw | bm25 | llm
   #   # token: $CRAWL4AI_TOKEN            # Bearer token (only if the server has JWT auth enabled)
 
@@ -783,6 +785,7 @@ tools:
   #   use: deerflow.community.fastcrw.tools:web_fetch_tool
   #   # api_key: $CRW_API_KEY
   #   # base_url: https://fastcrw.com/api   # default cloud; set to e.g. http://localhost:3000 for self-host
+  #   # allow_private_addresses: false      # SSRF guard: keep false in production. Set true ONLY for intentional internal targets.
 
   # Image search tool (uses DuckDuckGo)
   # Use this to find reference images before image generation