diff --git a/vulnerabilities/fetcher.py b/vulnerabilities/fetcher.py
new file mode 100644
index 000000000..c30360d67
--- /dev/null
+++ b/vulnerabilities/fetcher.py
@@ -0,0 +1,496 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+import csv
+import logging
+import threading
+import time
+from http import HTTPStatus
+from io import StringIO
+from typing import Any
+from typing import Callable
+from typing import Dict
+from typing import Iterator
+from typing import Optional
+
+import requests
+import saneyaml
+import urllib3
+from urllib3.util.retry import Retry
+
+module_logger = logging.getLogger(__name__)
+
+
+class Fetcher:
+    """
+    Centralized HTTP client with logging, retries, rate limiting, and proxy support.
+
+    This class provides a unified interface for all network operations in VulnerableCode,
+    enabling consistent logging, error handling, and configuration across all importers.
+
+    Usage:
+        # Simple request
+        fetcher = Fetcher()
+        response = fetcher.get("https://api.example.com/data")
+
+        # With pipeline logging
+        fetcher = Fetcher(logger=self.log, timeout=60, rate_limit=10)
+        data = fetcher.fetch_json("https://api.example.com/endpoint")
+
+        # Streaming large files
+        for chunk in fetcher.stream(url, chunk_size=8192):
+            f.write(chunk)
+
+        # As context manager
+        with Fetcher(logger=self.log) as f:
+            data = f.fetch_json("https://api.example.com/endpoint")
+    """
+
+    def __init__(
+        self,
+        logger: Optional[Callable] = None,
+        user_agent: Optional[str] = None,
+        proxy: Optional[Dict[str, str]] = None,
+        timeout: int = 30,
+        retry_count: int = 3,
+        retry_statuses: tuple = (500, 502, 503, 504),
+        backoff_factor: float = 0.5,
+        rate_limit: Optional[float] = None,
+    ):
+        """
+        Initialize the Fetcher with configuration.
+
+        Args:
+            logger: Callable for logging (e.g., pipeline's self.log or logging.info).
+                   If None, uses module logger.
+            user_agent: Custom User-Agent string. If None, uses default.
+            proxy: Proxy configuration dict: {"http": "...", "https": "..."}.
+            timeout: Request timeout in seconds. Default: 30.
+            retry_count: Maximum number of retries for failed requests. Default: 3.
+            retry_statuses: HTTP status codes to retry on. Default: (500, 502, 503, 504).
+            backoff_factor: Exponential backoff multiplier for retries. Default: 0.5.
+            rate_limit: Maximum requests per second. None = unlimited. Default: None.
+        """
+        self.logger = logger if logger else module_logger.info
+        self.user_agent = (
+            user_agent
+            if user_agent
+            else "aboutcode/vulnerablecode (+https://github.com/aboutcode-org/vulnerablecode)"
+        )
+        self.proxy = proxy
+        self.timeout = timeout
+        # Treat 0 or None as unlimited rate
+        self.rate_limit = rate_limit if rate_limit and rate_limit > 0 else None
+
+        # Initialize rate limiting (token bucket algorithm)
+        self._rate_limit_lock = threading.Lock()
+        self._last_request_time = 0.0
+
+        # Create requests session with retry adapter
+        self._session = requests.Session()
+
+        # Configure retry logic
+        if retry_count > 0:
+            retry_strategy = Retry(
+                total=retry_count,
+                backoff_factor=backoff_factor,
+                status_forcelist=retry_statuses,
+                allowed_methods=["HEAD", "GET", "POST", "PUT", "DELETE", "OPTIONS", "TRACE"],
+            )
+            adapter = requests.adapters.HTTPAdapter(max_retries=retry_strategy)
+            self._session.mount("http://", adapter)
+            self._session.mount("https://", adapter)
+
+        # Configure proxy if provided
+        if self.proxy:
+            self._session.proxies.update(self.proxy)
+
+        # Set default headers
+        self._session.headers.update({"User-Agent": self.user_agent})
+
+    def _apply_rate_limit(self):
+        """
+        Apply rate limiting if configured.
+
+        Uses a simple token bucket algorithm to limit requests per second.
+        Thread-safe implementation using a lock.
+        """
+        if self.rate_limit is None:
+            return
+
+        with self._rate_limit_lock:
+            current_time = time.time()
+            time_since_last_request = current_time - self._last_request_time
+
+            # Calculate minimum time between requests
+            min_interval = 1.0 / self.rate_limit
+
+            # If not enough time has passed, sleep
+            if time_since_last_request < min_interval:
+                sleep_time = min_interval - time_since_last_request
+                time.sleep(sleep_time)
+
+            self._last_request_time = time.time()
+
+    def _log(self, message: str, level: str = "INFO"):
+        """
+        Log a message using the configured logger.
+
+        Args:
+            message: Message to log.
+            level: Log level (INFO, WARNING, ERROR). Included in message for parsing.
+        """
+        if self.logger:
+            # Simply call the logger with the message
+            # The level information is already in the message format (e.g., "FAILED")
+            # Pipeline loggers can parse the message if they need level information
+            self.logger(message)
+
+    def _make_request(self, method: str, url: str, **kwargs) -> requests.Response:
+        """
+        Make an HTTP request with logging and rate limiting.
+
+        Args:
+            method: HTTP method (GET, POST, HEAD, etc.).
+            url: URL to request.
+            **kwargs: Additional arguments passed to requests.Session.request().
+
+        Returns:
+            requests.Response: The HTTP response.
+
+        Raises:
+            requests.exceptions.HTTPError: For HTTP errors.
+            requests.exceptions.RequestException: For other request errors.
+        """
+        # Apply rate limiting
+        self._apply_rate_limit()
+
+        # Set default timeout if not provided
+        if "timeout" not in kwargs:
+            kwargs["timeout"] = self.timeout
+
+        # Make the request and measure time
+        start_time = time.time()
+
+        try:
+            response = self._session.request(method, url, **kwargs)
+            duration = time.time() - start_time
+
+            # Log successful request
+            self._log(
+                f"[Fetcher] {method.upper()} {url} ({response.status_code} {response.reason}, {duration:.2f}s)"
+            )
+
+            # Raise HTTPError for bad status codes
+            response.raise_for_status()
+
+            return response
+
+        except requests.exceptions.HTTPError as e:
+            duration = time.time() - start_time
+            self._log(
+                f"[Fetcher] {method.upper()} {url} FAILED ({e.response.status_code}, {duration:.2f}s)",
+                level="ERROR",
+            )
+            raise
+
+        except requests.exceptions.RequestException as e:
+            duration = time.time() - start_time
+            self._log(
+                f"[Fetcher] {method.upper()} {url} FAILED ({str(e)}, {duration:.2f}s)",
+                level="ERROR",
+            )
+            raise
+
+    # Core HTTP methods
+    def get(self, url: str, **kwargs) -> requests.Response:
+        """
+        Perform a GET request.
+
+        Args:
+            url: URL to request.
+            **kwargs: Additional arguments passed to requests.
+
+        Returns:
+            requests.Response: The HTTP response.
+        """
+        return self._make_request("GET", url, **kwargs)
+
+    def post(self, url: str, **kwargs) -> requests.Response:
+        """
+        Perform a POST request.
+
+        Args:
+            url: URL to request.
+            **kwargs: Additional arguments passed to requests.
+
+        Returns:
+            requests.Response: The HTTP response.
+        """
+        return self._make_request("POST", url, **kwargs)
+
+    def head(self, url: str, **kwargs) -> requests.Response:
+        """
+        Perform a HEAD request.
+
+        Args:
+            url: URL to request.
+            **kwargs: Additional arguments passed to requests.
+
+        Returns:
+            requests.Response: The HTTP response.
+        """
+        return self._make_request("HEAD", url, **kwargs)
+
+    # Convenience methods for common response types
+    def fetch_json(self, url: str, **kwargs) -> dict:
+        """
+        Fetch and parse JSON from URL.
+
+        Args:
+            url: URL to fetch.
+            **kwargs: Additional arguments passed to requests.
+
+        Returns:
+            dict: Parsed JSON data.
+
+        Raises:
+            ValueError: If response is not valid JSON.
+        """
+        response = self.get(url, **kwargs)
+        return response.json()
+
+    def fetch_yaml(self, url: str, **kwargs) -> Any:
+        """
+        Fetch and parse YAML from URL.
+
+        Args:
+            url: URL to fetch.
+            **kwargs: Additional arguments passed to requests.
+
+        Returns:
+            Any: Parsed YAML data.
+        """
+        response = self.get(url, **kwargs)
+        return saneyaml.load(response.content)
+
+    def fetch_text(self, url: str, encoding: str = "utf-8", **kwargs) -> str:
+        """
+        Fetch text content from URL.
+
+        Args:
+            url: URL to fetch.
+            encoding: Text encoding. Default: utf-8.
+            **kwargs: Additional arguments passed to requests.
+
+        Returns:
+            str: Text content.
+        """
+        response = self.get(url, **kwargs)
+        return response.content.decode(encoding)
+
+    def fetch_csv(self, url: str, **kwargs) -> csv.reader:
+        """
+        Fetch and parse CSV from URL.
+
+        Args:
+            url: URL to fetch.
+            **kwargs: Additional arguments passed to requests.
+
+        Returns:
+            csv.reader: CSV reader object.
+        """
+        response = self.get(url, **kwargs)
+        content = response.content.decode("utf-8")
+        return csv.reader(StringIO(content))
+
+    def stream(self, url: str, chunk_size: int = 8192, **kwargs) -> Iterator[bytes]:
+        """
+        Stream large files chunk by chunk.
+
+        Args:
+            url: URL to fetch.
+            chunk_size: Size of each chunk in bytes. Default: 8192.
+            **kwargs: Additional arguments passed to requests.
+
+        Yields:
+            bytes: Chunks of file content.
+        """
+        # Force streaming mode
+        kwargs["stream"] = True
+
+        response = self._make_request("GET", url, **kwargs)
+
+        for chunk in response.iter_content(chunk_size=chunk_size):
+            if chunk:  # Filter out keep-alive chunks
+                yield chunk
+
+    def fetch_graphql(
+        self,
+        url: str,
+        query: str,
+        variables: Optional[Dict] = None,
+        token: Optional[str] = None,
+        **kwargs,
+    ) -> dict:
+        """
+        Execute a GraphQL query.
+
+        Args:
+            url: GraphQL endpoint URL.
+            query: GraphQL query string.
+            variables: Optional query variables.
+            token: Optional bearer token for authentication.
+            **kwargs: Additional arguments passed to requests.
+
+        Returns:
+            dict: GraphQL response data.
+        """
+        headers = kwargs.pop("headers", {})
+
+        # Add authentication if token provided
+        if token:
+            headers["Authorization"] = f"bearer {token}"
+
+        # Prepare GraphQL request
+        json_data = {"query": query}
+        if variables:
+            json_data["variables"] = variables
+
+        response = self.post(url, json=json_data, headers=headers, **kwargs)
+        return response.json()
+
+    # Context manager support
+    def __enter__(self):
+        """Enter context manager."""
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        """Exit context manager and cleanup session."""
+        if self._session:
+            self._session.close()
+        return False
+
+
+def get_fetcher_from_settings(**overrides):
+    """
+    Create a Fetcher instance using Django settings.
+
+    This helper reads FETCHER_* configuration from Django settings and creates
+    a properly configured Fetcher instance. Use this in importers to get a
+    centrally-configured HTTP client.
+
+    Args:
+        **overrides: Override any default settings (e.g., logger=self.log, timeout=60)
+
+    Returns:
+        Fetcher: Configured Fetcher instance.
+
+    Example:
+        # In an importer pipeline:
+        fetcher = get_fetcher_from_settings(logger=self.log)
+        data = fetcher.fetch_json("https://api.example.com/data")
+
+        # With custom rate limiting:
+        fetcher = get_fetcher_from_settings(logger=self.log, rate_limit=5.0)
+        data = fetcher.fetch_yaml("https://example.com/advisories.yaml")
+    """
+    from django.conf import settings
+
+    # Build proxy dict if configured
+    proxy = {}
+    if settings.FETCHER_PROXY_HTTP:
+        proxy["http"] = settings.FETCHER_PROXY_HTTP
+    if settings.FETCHER_PROXY_HTTPS:
+        proxy["https"] = settings.FETCHER_PROXY_HTTPS
+
+    config = {
+        "user_agent": settings.FETCHER_USER_AGENT,
+        "timeout": settings.FETCHER_TIMEOUT,
+        "retry_count": settings.FETCHER_RETRY_COUNT,
+        "backoff_factor": settings.FETCHER_RETRY_BACKOFF,
+        "rate_limit": settings.FETCHER_RATE_LIMIT,
+        "proxy": proxy if proxy else None,
+    }
+
+    # Apply overrides
+    config.update(overrides)
+
+    return Fetcher(**config)
+
+
+# Migration Guide for Importers
+# ==============================
+#
+# This guide shows how to migrate legacy importers to use the centralized Fetcher.
+#
+# Option 1: Use get_fetcher_from_settings() (RECOMMENDED)
+# --------------------------------------------------------
+# This automatically uses all FETCHER_* settings from Django configuration.
+#
+# Before (legacy):
+#     response = requests.get(url)
+#     data = response.json()
+#
+# After (with Fetcher):
+#     from vulnerabilities.fetcher import get_fetcher_from_settings
+#
+#     fetcher = get_fetcher_from_settings(logger=self.log)
+#     data = fetcher.fetch_json(url)
+#
+# Option 2: Use backward-compatible utils (NO CHANGES NEEDED)
+# -----------------------------------------------------------
+# Existing code using utils.fetch_yaml() and utils.fetch_response() will
+# automatically use Fetcher internally with fallback to legacy behavior.
+#
+# Before:
+#     from vulnerabilities.utils import fetch_yaml
+#     data = fetch_yaml(url)
+#
+# After:
+#     from vulnerabilities.utils import fetch_yaml
+#     data = fetch_yaml(url)  # Now uses Fetcher internally!
+#
+# Option 3: Direct Fetcher instantiation (for advanced use cases)
+# ---------------------------------------------------------------
+# Use this when you need custom configuration not covered by settings.
+#
+# Before:
+#     session = requests.Session()
+#     session.headers.update({"Authorization": f"token {token}"})
+#     response = session.get(url)
+#
+# After:
+#     from vulnerabilities.fetcher import Fetcher
+#
+#     fetcher = Fetcher(
+#         logger=self.log,
+#         rate_limit=10.0,  # 10 requests/second
+#         timeout=60,
+#     )
+#     response = fetcher.get(url, headers={"Authorization": f"token {token}"})
+#
+# Available Convenience Methods
+# -----------------------------
+# - fetch_json(url, **kwargs) -> dict
+# - fetch_yaml(url, **kwargs) -> Any
+# - fetch_text(url, encoding="utf-8", **kwargs) -> str
+# - fetch_csv(url, **kwargs) -> csv.reader
+# - stream(url, chunk_size=8192, **kwargs) -> Iterator[bytes]
+# - fetch_graphql(url, query, variables=None, token=None, **kwargs) -> dict
+#
+# Benefits of Using Fetcher
+# -------------------------
+# - Automatic retries on 5xx errors with exponential backoff
+# - Rate limiting to avoid overwhelming APIs
+# - Centralized logging with [Fetcher] prefix
+# - Session reuse for better performance (connection pooling)
+# - Proxy support from environment variables
+# - Consistent User-Agent across all importers
+# - Context manager support for proper cleanup
diff --git a/vulnerabilities/tests/test_fetcher.py b/vulnerabilities/tests/test_fetcher.py
new file mode 100644
index 000000000..49465154f
--- /dev/null
+++ b/vulnerabilities/tests/test_fetcher.py
@@ -0,0 +1,478 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+import logging
+import time
+
+import pytest
+import responses
+
+from vulnerabilities.fetcher import Fetcher
+
+
+class TestFetcherHTTPMethods:
+    """Test basic HTTP methods (GET, POST, HEAD)."""
+
+    @responses.activate
+    def test_get_success(self):
+        """Test successful GET request."""
+        responses.add(
+            responses.GET,
+            "https://example.com/api",
+            json={"status": "ok"},
+            status=200,
+        )
+
+        fetcher = Fetcher()
+        response = fetcher.get("https://example.com/api")
+
+        assert response.status_code == 200
+        assert response.json() == {"status": "ok"}
+
+    @responses.activate
+    def test_post_success(self):
+        """Test successful POST request."""
+        responses.add(
+            responses.POST,
+            "https://example.com/api",
+            json={"created": True},
+            status=201,
+        )
+
+        fetcher = Fetcher()
+        response = fetcher.post("https://example.com/api", json={"data": "test"})
+
+        assert response.status_code == 201
+        assert response.json() == {"created": True}
+
+    @responses.activate
+    def test_head_success(self):
+        """Test successful HEAD request."""
+        responses.add(
+            responses.HEAD,
+            "https://example.com/api",
+            status=200,
+        )
+
+        fetcher = Fetcher()
+        response = fetcher.head("https://example.com/api")
+
+        assert response.status_code == 200
+
+    @responses.activate
+    def test_get_with_custom_timeout(self):
+        """Test GET request with custom timeout."""
+        responses.add(
+            responses.GET,
+            "https://example.com/api",
+            json={"status": "ok"},
+            status=200,
+        )
+
+        fetcher = Fetcher(timeout=60)
+        response = fetcher.get("https://example.com/api")
+
+        assert response.status_code == 200
+
+
+class TestFetcherRetryLogic:
+    """Test retry logic for failed requests."""
+
+    @responses.activate
+    def test_retry_on_500(self):
+        """Test retry on 500 errors."""
+        # First two requests fail with 500, third succeeds
+        responses.add(responses.GET, "https://example.com/api", status=500)
+        responses.add(responses.GET, "https://example.com/api", status=500)
+        responses.add(responses.GET, "https://example.com/api", json={"ok": True}, status=200)
+
+        fetcher = Fetcher(retry_count=3)
+        response = fetcher.get("https://example.com/api")
+
+        assert response.status_code == 200
+        assert len(responses.calls) == 3
+
+    @responses.activate
+    def test_retry_exhausted(self):
+        """Test that exception is raised after max retries."""
+        # All requests fail with 500
+        for _ in range(5):
+            responses.add(responses.GET, "https://example.com/api", status=500)
+
+        fetcher = Fetcher(retry_count=2)
+
+        with pytest.raises(Exception):
+            fetcher.get("https://example.com/api")
+
+    @responses.activate
+    def test_no_retry_on_404(self):
+        """Test that 4xx errors are not retried."""
+        responses.add(responses.GET, "https://example.com/api", status=404)
+
+        fetcher = Fetcher(retry_count=3)
+
+        with pytest.raises(Exception):
+            fetcher.get("https://example.com/api")
+
+        # Should only be called once (no retries for 404)
+        assert len(responses.calls) == 1
+
+
+class TestFetcherRateLimiting:
+    """Test rate limiting functionality."""
+
+    @responses.activate
+    def test_rate_limiting(self):
+        """Test that rate limiting delays requests."""
+        # Add 3 responses
+        for _ in range(3):
+            responses.add(responses.GET, "https://example.com/api", status=200)
+
+        fetcher = Fetcher(rate_limit=2.0)  # 2 requests per second
+
+        start = time.time()
+        for _ in range(3):
+            fetcher.get("https://example.com/api")
+        duration = time.time() - start
+
+        # 3 requests at 2/sec should take at least 1 second
+        assert duration >= 1.0
+
+    @responses.activate
+    def test_no_rate_limit(self):
+        """Test that requests are not delayed when rate_limit=None."""
+        # Add 3 responses
+        for _ in range(3):
+            responses.add(responses.GET, "https://example.com/api", status=200)
+
+        fetcher = Fetcher(rate_limit=None)
+
+        start = time.time()
+        for _ in range(3):
+            fetcher.get("https://example.com/api")
+        duration = time.time() - start
+
+        # Without rate limiting, should complete quickly (< 0.5 seconds)
+        assert duration < 0.5
+
+
+class TestFetcherLogging:
+    """Test logging functionality."""
+
+    @responses.activate
+    def test_logging_with_standard_logger(self, caplog):
+        """Test that requests are logged with standard logger."""
+        responses.add(responses.GET, "https://example.com/api", status=200)
+
+        logger = logging.getLogger(__name__)
+        with caplog.at_level(logging.INFO):
+            fetcher = Fetcher(logger=logger.info)
+            fetcher.get("https://example.com/api")
+
+        # Check that URL and status code are in logs
+        assert "https://example.com/api" in caplog.text
+        assert "200" in caplog.text
+
+    @responses.activate
+    def test_logging_with_custom_logger(self):
+        """Test that requests are logged with custom logger."""
+        responses.add(responses.GET, "https://example.com/api", status=200)
+
+        log_messages = []
+
+        def custom_logger(message, level=logging.INFO):
+            log_messages.append(message)
+
+        fetcher = Fetcher(logger=custom_logger)
+        fetcher.get("https://example.com/api")
+
+        assert len(log_messages) > 0
+        assert "https://example.com/api" in log_messages[0]
+        assert "200" in log_messages[0]
+
+
+class TestFetcherConvenienceMethods:
+    """Test convenience methods for common response types."""
+
+    @responses.activate
+    def test_fetch_json(self):
+        """Test JSON parsing convenience method."""
+        responses.add(
+            responses.GET,
+            "https://example.com/api.json",
+            json={"key": "value", "nested": {"data": 123}},
+        )
+
+        fetcher = Fetcher()
+        data = fetcher.fetch_json("https://example.com/api.json")
+
+        assert data == {"key": "value", "nested": {"data": 123}}
+
+    @responses.activate
+    def test_fetch_yaml(self):
+        """Test YAML parsing convenience method."""
+        yaml_content = """
+key: value
+list:
+  - item1
+  - item2
+nested:
+  data: 123
+"""
+        responses.add(
+            responses.GET,
+            "https://example.com/data.yaml",
+            body=yaml_content,
+        )
+
+        fetcher = Fetcher()
+        data = fetcher.fetch_yaml("https://example.com/data.yaml")
+
+        assert data["key"] == "value"
+        assert data["list"] == ["item1", "item2"]
+        assert data["nested"]["data"] == "123"  # saneyaml preserves as string
+
+    @responses.activate
+    def test_fetch_text(self):
+        """Test text fetching convenience method."""
+        text_content = "Hello, World!\nThis is a test."
+        responses.add(
+            responses.GET,
+            "https://example.com/file.txt",
+            body=text_content,
+        )
+
+        fetcher = Fetcher()
+        text = fetcher.fetch_text("https://example.com/file.txt")
+
+        assert text == text_content
+
+    @responses.activate
+    def test_fetch_csv(self):
+        """Test CSV parsing convenience method."""
+        csv_content = """name,age,city
+Alice,30,NYC
+Bob,25,LA
+Charlie,35,SF"""
+        responses.add(
+            responses.GET,
+            "https://example.com/data.csv",
+            body=csv_content,
+        )
+
+        fetcher = Fetcher()
+        csv_reader = fetcher.fetch_csv("https://example.com/data.csv")
+
+        rows = list(csv_reader)
+        assert len(rows) == 4
+        assert rows[0] == ["name", "age", "city"]
+        assert rows[1] == ["Alice", "30", "NYC"]
+        assert rows[2] == ["Bob", "25", "LA"]
+
+
+class TestFetcherStreaming:
+    """Test streaming functionality for large files."""
+
+    @responses.activate
+    def test_streaming(self):
+        """Test streaming for large files."""
+        large_content = b"x" * 10000
+        responses.add(
+            responses.GET,
+            "https://example.com/large.bin",
+            body=large_content,
+            stream=True,
+        )
+
+        fetcher = Fetcher()
+        chunks = list(fetcher.stream("https://example.com/large.bin", chunk_size=1000))
+
+        # Verify we got chunks
+        assert len(chunks) > 0
+
+        # Verify content is correct when reassembled
+        reassembled = b"".join(chunks)
+        assert reassembled == large_content
+
+    @responses.activate
+    def test_streaming_custom_chunk_size(self):
+        """Test streaming with custom chunk size."""
+        content = b"a" * 5000
+        responses.add(
+            responses.GET,
+            "https://example.com/file.bin",
+            body=content,
+            stream=True,
+        )
+
+        fetcher = Fetcher()
+        chunks = list(fetcher.stream("https://example.com/file.bin", chunk_size=500))
+
+        # Verify content is correct
+        assert b"".join(chunks) == content
+
+
+class TestFetcherGraphQL:
+    """Test GraphQL support."""
+
+    @responses.activate
+    def test_fetch_graphql_simple(self):
+        """Test simple GraphQL query."""
+        responses.add(
+            responses.POST,
+            "https://api.example.com/graphql",
+            json={"data": {"user": {"name": "Alice", "id": 123}}},
+            status=200,
+        )
+
+        fetcher = Fetcher()
+        query = "{ user(id: 123) { name id } }"
+        result = fetcher.fetch_graphql("https://api.example.com/graphql", query)
+
+        assert result["data"]["user"]["name"] == "Alice"
+
+    @responses.activate
+    def test_fetch_graphql_with_variables(self):
+        """Test GraphQL query with variables."""
+        responses.add(
+            responses.POST,
+            "https://api.example.com/graphql",
+            json={"data": {"user": {"name": "Bob", "id": 456}}},
+            status=200,
+        )
+
+        fetcher = Fetcher()
+        query = "query GetUser($id: Int!) { user(id: $id) { name id } }"
+        variables = {"id": 456}
+        result = fetcher.fetch_graphql(
+            "https://api.example.com/graphql", query, variables=variables
+        )
+
+        assert result["data"]["user"]["id"] == 456
+
+    @responses.activate
+    def test_fetch_graphql_with_auth(self):
+        """Test GraphQL query with authentication."""
+        responses.add(
+            responses.POST,
+            "https://api.example.com/graphql",
+            json={"data": {"viewer": {"login": "testuser"}}},
+            status=200,
+        )
+
+        fetcher = Fetcher()
+        query = "{ viewer { login } }"
+        result = fetcher.fetch_graphql(
+            "https://api.example.com/graphql", query, token="test-token-123"
+        )
+
+        assert result["data"]["viewer"]["login"] == "testuser"
+
+        # Verify Authorization header was sent
+        assert responses.calls[0].request.headers["Authorization"] == "bearer test-token-123"
+
+
+class TestFetcherContextManager:
+    """Test context manager protocol."""
+
+    @responses.activate
+    def test_context_manager(self):
+        """Test context manager cleanup."""
+        responses.add(
+            responses.GET,
+            "https://example.com/api",
+            json={"status": "ok"},
+            status=200,
+        )
+
+        with Fetcher() as fetcher:
+            response = fetcher.get("https://example.com/api")
+            assert response.status_code == 200
+
+        # Session should be closed after exiting context
+
+    @responses.activate
+    def test_context_manager_with_exception(self):
+        """Test context manager cleanup even when exception occurs."""
+        responses.add(
+            responses.GET,
+            "https://example.com/api",
+            status=500,
+        )
+
+        try:
+            with Fetcher(retry_count=0) as fetcher:
+                fetcher.get("https://example.com/api")
+        except Exception:
+            pass  # Expected exception
+
+        # Session should still be closed
+
+
+class TestFetcherConfiguration:
+    """Test configuration options."""
+
+    @responses.activate
+    def test_custom_user_agent(self):
+        """Test custom user agent configuration."""
+        responses.add(
+            responses.GET,
+            "https://example.com/api",
+            status=200,
+        )
+
+        fetcher = Fetcher(user_agent="custom-agent/1.0")
+        fetcher.get("https://example.com/api")
+
+        # Verify custom user agent was sent
+        assert responses.calls[0].request.headers["User-Agent"] == "custom-agent/1.0"
+
+    @responses.activate
+    def test_default_user_agent(self):
+        """Test default user agent."""
+        responses.add(
+            responses.GET,
+            "https://example.com/api",
+            status=200,
+        )
+
+        fetcher = Fetcher()
+        fetcher.get("https://example.com/api")
+
+        # Verify default user agent contains "vulnerablecode"
+        user_agent = responses.calls[0].request.headers["User-Agent"]
+        assert "vulnerablecode" in user_agent
+
+
+class TestFetcherErrorHandling:
+    """Test error handling."""
+
+    @responses.activate
+    def test_http_error_raises_exception(self):
+        """Test that HTTP errors raise exceptions."""
+        responses.add(
+            responses.GET,
+            "https://example.com/api",
+            status=404,
+        )
+
+        fetcher = Fetcher(retry_count=0)
+
+        with pytest.raises(Exception):
+            fetcher.get("https://example.com/api")
+
+    @responses.activate
+    def test_network_error_raises_exception(self):
+        """Test that network errors raise exceptions."""
+        # responses library will raise ConnectionError for unknown URLs
+        fetcher = Fetcher(retry_count=0)
+
+        with pytest.raises(Exception):
+            # Using a URL that won't be mocked
+            fetcher.get("https://nonexistent-domain-12345.com")
diff --git a/vulnerabilities/utils.py b/vulnerabilities/utils.py
index 999244498..29018a80b 100644
--- a/vulnerabilities/utils.py
+++ b/vulnerabilities/utils.py
@@ -75,8 +75,23 @@ def load_toml(path):
 
 
 def fetch_yaml(url):
-    response = requests.get(url)
-    return saneyaml.load(response.content)
+    """
+    Fetch and parse YAML from URL.
+
+    This is a backward-compatible wrapper around Fetcher.fetch_yaml().
+    """
+    try:
+        from vulnerabilities.fetcher import Fetcher
+
+        fetcher = Fetcher()
+        return fetcher.fetch_yaml(url)
+    except Exception as e:
+        # Fallback to old behavior if Fetcher fails
+        import logging
+
+        logging.warning(f"Fetcher failed for {url}: {e}, using fallback")
+        response = requests.get(url)
+        return saneyaml.load(response.content)
 
 
 # FIXME: Remove this entirely after complete importer-improver migration
@@ -377,12 +392,25 @@ def resolve_version_range(
 
 def fetch_response(url):
     """
-    Fetch and return `response` from the `url`
+    Fetch and return `response` from the `url`.
+
+    This is a backward-compatible wrapper around Fetcher.get().
     """
-    response = requests.get(url)
-    if response.status_code == HTTPStatus.OK:
+    try:
+        from vulnerabilities.fetcher import Fetcher
+
+        fetcher = Fetcher()
+        return fetcher.get(url)
+    except Exception as e:
+        # Fallback to old behavior if Fetcher fails
+        import logging
+
+        logging.warning(f"Fetcher failed for {url}: {e}, using fallback")
+        response = requests.get(url)
+        # Use raise_for_status() to be consistent with Fetcher behavior
+        # This accepts all 2xx status codes, not just 200
+        response.raise_for_status()
         return response
-    raise Exception(f"Failed to fetch data from {url!r} with status code: {response.status_code!r}")
 
 
 # This should be a method on PackageURL
diff --git a/vulnerablecode/settings.py b/vulnerablecode/settings.py
index 05a3d0fa8..47bd9e9bd 100644
--- a/vulnerablecode/settings.py
+++ b/vulnerablecode/settings.py
@@ -376,6 +376,34 @@
     }
 
 
+# Network Fetcher Configuration
+# Configure centralized HTTP client for all network operations in importers.
+#
+# Environment variables:
+#   FETCHER_USER_AGENT: Custom User-Agent string for HTTP requests
+#   FETCHER_TIMEOUT: Request timeout in seconds (default: 30)
+#   FETCHER_RETRY_COUNT: Number of retries for failed requests (default: 3)
+#   FETCHER_RETRY_BACKOFF: Exponential backoff factor for retries (default: 0.5)
+#   FETCHER_RATE_LIMIT: Max requests per second (0 = unlimited, default: 0)
+#   FETCHER_PROXY_HTTP: HTTP proxy URL (e.g., http://proxy.example.com:8080)
+#   FETCHER_PROXY_HTTPS: HTTPS proxy URL (e.g., https://proxy.example.com:8443)
+#
+# Example usage:
+#   export FETCHER_TIMEOUT=60
+#   export FETCHER_RETRY_COUNT=5
+#   export FETCHER_RATE_LIMIT=10.0  # 10 requests/second
+#
+FETCHER_USER_AGENT = env.str(
+    "FETCHER_USER_AGENT",
+    default="aboutcode/vulnerablecode (+https://github.com/aboutcode-org/vulnerablecode)",
+)
+FETCHER_TIMEOUT = env.int("FETCHER_TIMEOUT", default=30)
+FETCHER_RETRY_COUNT = env.int("FETCHER_RETRY_COUNT", default=3)
+FETCHER_RETRY_BACKOFF = env.float("FETCHER_RETRY_BACKOFF", default=0.5)
+FETCHER_RATE_LIMIT = env.float("FETCHER_RATE_LIMIT", default=0.0)
+FETCHER_PROXY_HTTP = env.str("FETCHER_PROXY_HTTP", default="")
+FETCHER_PROXY_HTTPS = env.str("FETCHER_PROXY_HTTPS", default="")
+
 VULNERABLECODE_PIPELINE_TIMEOUT = env.int("VULNERABLECODE_PIPELINE_TIMEOUT", default=24)
 RQ_QUEUES = {
     "default": {