diff --git a/agent/tools/web.py b/agent/tools/web.py
index d54c6c3..133da1c 100644
--- a/agent/tools/web.py
+++ b/agent/tools/web.py
@@ -2,6 +2,7 @@
import json
import socket
+import time
import urllib.request
import urllib.parse
import urllib.error
@@ -12,6 +13,29 @@
from agent.tools.base import BaseTool, ToolResult
+def _fetch_with_retry(
+ req: urllib.request.Request, timeout: float, max_retries: int
+) -> tuple[str, str]:
+ """Fetch a URL with retry on transient network failures.
+
+ Returns (body, content_type). HTTP error responses (4xx/5xx) are not
+ retried - they're permanent failures, so retrying just wastes time.
+ """
+ for attempt in range(max_retries + 1):
+ try:
+ resp = urllib.request.urlopen(req, timeout=timeout)
+ body = resp.read().decode("utf-8", errors="replace")
+ return body, resp.headers.get("Content-Type", "")
+ except urllib.error.HTTPError:
+ raise
+ except (socket.timeout, urllib.error.URLError):
+ if attempt < max_retries:
+ time.sleep(1 + attempt)
+ continue
+ raise
+ raise AssertionError("unreachable")
+
+
class WebSearchTool(BaseTool):
"""Search the web using DuckDuckGo."""
@@ -39,6 +63,7 @@ class WebSearchTool(BaseTool):
def __init__(self, config=None):
self.timeout = config.request_timeout if config else 15
self.user_agent = config.user_agent if config else "CodeAgent/1.0"
+ self.max_retries = 2
def execute(self, query: str, max_results: int = 5, **kw) -> ToolResult:
try:
@@ -54,8 +79,7 @@ def execute(self, query: str, max_results: int = 5, **kw) -> ToolResult:
},
)
- resp = urllib.request.urlopen(req, timeout=self.timeout)
- body = resp.read().decode("utf-8", errors="replace")
+ body, _ = _fetch_with_retry(req, self.timeout, self.max_retries)
# Parse results from HTML
results = self._parse_ddg_html(body, max_results)
@@ -142,6 +166,7 @@ class WebFetchTool(BaseTool):
def __init__(self, config=None):
self.timeout = config.request_timeout if config else 15
self.user_agent = config.user_agent if config else "CodeAgent/1.0"
+ self.max_retries = 2
def execute(self, url: str, max_length: int = 10000, **kw) -> ToolResult:
try:
@@ -156,10 +181,7 @@ def execute(self, url: str, max_length: int = 10000, **kw) -> ToolResult:
},
)
- resp = urllib.request.urlopen(req, timeout=self.timeout)
- content_type = resp.headers.get("Content-Type", "")
-
- body = resp.read().decode("utf-8", errors="replace")
+ body, content_type = _fetch_with_retry(req, self.timeout, self.max_retries)
if "text/html" in content_type:
text = self._html_to_text(body)
diff --git a/tests/test_web.py b/tests/test_web.py
new file mode 100644
index 0000000..686e201
--- /dev/null
+++ b/tests/test_web.py
@@ -0,0 +1,107 @@
+"""Tests for the web search/fetch tools."""
+
+import io
+import sys
+import unittest
+import urllib.error
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from agent.tools.web import WebFetchTool, WebSearchTool, _fetch_with_retry
+
+
+def _fake_response(body: bytes, content_type: str = "text/html"):
+ resp = io.BytesIO(body)
+ resp.headers = type("Headers", (), {"get": lambda self, k, d=None: content_type})()
+ return resp
+
+
+class TestHtmlToText(unittest.TestCase):
+ def setUp(self):
+ self.tool = WebFetchTool()
+
+ def test_br_tag_becomes_newline(self):
+ self.assertEqual(self.tool._html_to_text("a
b"), "a\nb")
+
+ def test_self_closing_br_tag_becomes_newline(self):
+ self.assertEqual(self.tool._html_to_text("a
b"), "a\nb")
+
+ def test_br_with_space_becomes_newline(self):
+ self.assertEqual(self.tool._html_to_text("a
b"), "a\nb")
+
+
+class TestFetchWithRetry(unittest.TestCase):
+ def test_returns_body_and_content_type_on_success(self):
+ with patch("urllib.request.urlopen", return_value=_fake_response(b"hello", "text/plain")):
+ body, content_type = _fetch_with_retry(req=None, timeout=1, max_retries=2)
+ self.assertEqual(body, "hello")
+ self.assertEqual(content_type, "text/plain")
+
+ def test_retries_on_transient_url_error_then_succeeds(self):
+ calls = {"n": 0}
+
+ def fake_urlopen(req, timeout=None):
+ calls["n"] += 1
+ if calls["n"] < 2:
+ raise urllib.error.URLError("connection refused")
+ return _fake_response(b"ok")
+
+ with patch("urllib.request.urlopen", side_effect=fake_urlopen), \
+ patch("time.sleep"):
+ body, _ = _fetch_with_retry(req=None, timeout=1, max_retries=2)
+
+ self.assertEqual(body, "ok")
+ self.assertEqual(calls["n"], 2)
+
+ def test_gives_up_after_max_retries(self):
+ with patch("urllib.request.urlopen", side_effect=urllib.error.URLError("down")), \
+ patch("time.sleep"):
+ with self.assertRaises(urllib.error.URLError):
+ _fetch_with_retry(req=None, timeout=1, max_retries=2)
+
+ def test_http_error_is_not_retried(self):
+ calls = {"n": 0}
+
+ def fake_urlopen(req, timeout=None):
+ calls["n"] += 1
+ raise urllib.error.HTTPError("http://x", 404, "Not Found", {}, None)
+
+ with patch("urllib.request.urlopen", side_effect=fake_urlopen), \
+ patch("time.sleep") as mock_sleep:
+ with self.assertRaises(urllib.error.HTTPError):
+ _fetch_with_retry(req=None, timeout=1, max_retries=2)
+
+ self.assertEqual(calls["n"], 1)
+ mock_sleep.assert_not_called()
+
+
+class TestWebFetchTool(unittest.TestCase):
+ def test_execute_issues_a_single_request(self):
+ tool = WebFetchTool()
+ with patch(
+ "urllib.request.urlopen",
+ return_value=_fake_response(b"hi
there
", "text/html"),
+ ) as mock_urlopen:
+ result = tool.execute(url="https://example.com")
+
+ self.assertEqual(mock_urlopen.call_count, 1)
+ self.assertTrue(result.success)
+ self.assertIn("hi\nthere", result.output)
+
+
+class TestWebSearchTool(unittest.TestCase):
+ def test_execute_returns_search_failed_on_persistent_error(self):
+ tool = WebSearchTool()
+ with patch(
+ "urllib.request.urlopen", side_effect=urllib.error.URLError("down")
+ ), patch("time.sleep"):
+ result = tool.execute(query="test")
+
+ self.assertFalse(result.success)
+ self.assertIn("Search failed", result.error)
+
+
+if __name__ == "__main__":
+ unittest.main()