diff --git a/src/fetch/README.md b/src/fetch/README.md
index ed6d2262f4..dcef9a8fc3 100644
--- a/src/fetch/README.md
+++ b/src/fetch/README.md
@@ -7,7 +7,7 @@ A Model Context Protocol server that provides web content fetching capabilities.
 Source: https://github.com/modelcontextprotocol/servers/tree/main/src/fetch
 
 > [!CAUTION]
-> This server can access local/internal IP addresses and may represent a security risk. Exercise caution when using this MCP server to ensure this does not expose any sensitive data.
+> By default this server refuses to fetch loopback, private (RFC1918), link-local, and cloud-metadata IP addresses to mitigate server-side request forgery (SSRF). If you deliberately need to fetch internal hosts, you can re-enable this with `--allow-internal-ips` (see [Customization - Internal IPs](#customization---internal-ips)) — doing so may expose internal services and cloud instance metadata to the model, so only use it in trusted environments. This protection does not apply when requests are routed through a proxy; see [Customization - Proxy](#customization---proxy).
 
 The fetch tool will truncate the response, but by using the `start_index` argument, you can specify where to start the content extraction. This lets models read a webpage in chunks, until they find the information they need.
 
@@ -172,6 +172,15 @@ This can be customized by adding the argument `--user-agent=YourUserAgent` to th
 
 The server can be configured to use a proxy by using the `--proxy-url` argument.
 
+> [!IMPORTANT]
+> The SSRF protection described below cannot be enforced when requests go through a proxy. The server resolves each destination itself, but the proxy resolves the hostname again for the forwarded request, and an attacker-controlled hostname can return a public address to the server and an internal one to the proxy. Note that this applies to `HTTP_PROXY`, `HTTPS_PROXY`, and `ALL_PROXY` in the environment as well as to `--proxy-url`, because the HTTP client honors those by default. The server warns on startup when it detects either. Restrict internal destinations at the proxy in that case.
+
+### Customization - Internal IPs
+
+By default the server blocks requests to loopback, private (RFC1918), link-local, cloud-metadata (169.254.169.254), and other non-public IP addresses to prevent SSRF, and it re-validates the destination on every redirect hop. If you need to fetch internal hosts (for example a service on `localhost`), add the argument `--allow-internal-ips` to the `args` list in the configuration. This disables the SSRF protection, so only enable it in trusted environments.
+
+Two limitations are worth knowing about. The destination is validated before the request and resolved again when the connection is made, so a hostname whose DNS answer changes between those two points can still be reached (DNS rebinding). And when a proxy is configured the guard cannot enforce the destination at all — see [Customization - Proxy](#customization---proxy).
+
 ## Windows Configuration
 
 If you're experiencing timeout issues on Windows, you may need to set the `PYTHONIOENCODING` environment variable to ensure proper character encoding:
diff --git a/src/fetch/src/mcp_server_fetch/__init__.py b/src/fetch/src/mcp_server_fetch/__init__.py
index 09744ce319..a2a83a3fbe 100644
--- a/src/fetch/src/mcp_server_fetch/__init__.py
+++ b/src/fetch/src/mcp_server_fetch/__init__.py
@@ -16,9 +16,15 @@ def main():
         help="Ignore robots.txt restrictions",
     )
     parser.add_argument("--proxy-url", type=str, help="Proxy URL to use for requests")
+    parser.add_argument(
+        "--allow-internal-ips",
+        action="store_true",
+        help="Allow fetching loopback, private, link-local, and cloud-metadata "
+        "addresses. Disables SSRF protection; only use in trusted environments.",
+    )
 
     args = parser.parse_args()
-    asyncio.run(serve(args.user_agent, args.ignore_robots_txt, args.proxy_url))
+    asyncio.run(serve(args.user_agent, args.ignore_robots_txt, args.proxy_url, args.allow_internal_ips))
 
 
 if __name__ == "__main__":
diff --git a/src/fetch/src/mcp_server_fetch/server.py b/src/fetch/src/mcp_server_fetch/server.py
index b42c7b1f6b..e5a0e2c4b7 100644
--- a/src/fetch/src/mcp_server_fetch/server.py
+++ b/src/fetch/src/mcp_server_fetch/server.py
@@ -1,3 +1,8 @@
+import asyncio
+import ipaddress
+import os
+import socket
+import sys
 from typing import Annotated, Tuple
 from urllib.parse import urlparse, urlunparse
 
@@ -63,7 +68,215 @@ def get_robots_txt_url(url: str) -> str:
     return robots_url
 
 
-async def check_may_autonomously_fetch_url(url: str, user_agent: str, proxy_url: str | None = None) -> None:
+# Cap on the number of redirect hops we will follow (and re-validate).
+MAX_REDIRECTS = 20
+
+# HTTP status codes that indicate a redirect with a Location header.
+REDIRECT_STATUS_CODES = (301, 302, 303, 307, 308)
+
+# Carrier-grade NAT range: not globally reachable, but not flagged by
+# is_private on Python < 3.13, so it is checked explicitly.
+_CGNAT_NETWORK = ipaddress.ip_network("100.64.0.0/10")
+
+# 6to4 (RFC 3056) carries an IPv4 address in bits 16-48, so 2002:a9fe:a9fe::
+# is a route to 169.254.169.254 wherever a 6to4 relay is reachable. Neither
+# is_reserved nor is_private covers it, so the prefix is unwrapped explicitly.
+_SIXTOFOUR_NETWORK = ipaddress.ip_network("2002::/16")
+
+# Deprecated IPv6 site-local prefix (RFC 3879): still routed internally by some
+# stacks and not flagged by is_private, so it is blocked explicitly.
+_SITE_LOCAL_NETWORK = ipaddress.ip_network("fec0::/10")
+
+# Proxy environment variables httpx honors: AsyncClient defaults to
+# trust_env=True, so these route requests through a proxy even when no
+# --proxy-url is passed.
+_PROXY_ENV_VARS = ("HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY")
+
+
+def _configured_proxy_sources(proxy_url: str | None) -> list[str]:
+    """Return the settings that will route requests through a proxy, if any."""
+    sources = ["--proxy-url"] if proxy_url else []
+    sources.extend(
+        name
+        for name in _PROXY_ENV_VARS
+        if os.environ.get(name) or os.environ.get(name.lower())
+    )
+    return sources
+
+
+def _is_blocked_ip(
+    ip: ipaddress.IPv4Address | ipaddress.IPv6Address,
+) -> bool:
+    """Return True if an IP address is not safe to fetch (SSRF target).
+
+    Blocks loopback, private (RFC1918), carrier-grade NAT, link-local
+    (including the cloud metadata address 169.254.169.254), unique-local,
+    site-local, multicast, reserved, and unspecified addresses. IPv6 forms
+    that embed an IPv4 address are unwrapped so an internal destination
+    cannot be reached by wrapping it in an address that looks routable.
+    """
+    # Unwrap IPv4-mapped IPv6 addresses (e.g. ::ffff:127.0.0.1) so the
+    # underlying IPv4 address is classified rather than the wrapper.
+    if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None:
+        ip = ip.ipv4_mapped
+    # Unwrap 6to4 (2002:a.b.c.d::/48), which embeds the IPv4 address in
+    # bits 16-48 rather than in the low 32 bits.
+    elif isinstance(ip, ipaddress.IPv6Address) and ip in _SIXTOFOUR_NETWORK:
+        ip = ipaddress.IPv4Address((int(ip) >> 80) & 0xFFFFFFFF)
+    # Unwrap deprecated IPv4-compatible IPv6 addresses (::a.b.c.d, ::/96),
+    # other than :: and ::1 which are classified as unspecified/loopback
+    # below, so the embedded IPv4 address (e.g. ::127.0.0.1) is checked.
+    elif (
+        isinstance(ip, ipaddress.IPv6Address)
+        and int(ip) >> 32 == 0
+        and int(ip) not in (0, 1)
+    ):
+        ip = ipaddress.IPv4Address(int(ip) & 0xFFFFFFFF)
+
+    if isinstance(ip, ipaddress.IPv4Address) and ip in _CGNAT_NETWORK:
+        return True
+
+    if isinstance(ip, ipaddress.IPv6Address) and ip in _SITE_LOCAL_NETWORK:
+        return True
+
+    return (
+        ip.is_loopback
+        or ip.is_private
+        or ip.is_link_local
+        or ip.is_multicast
+        or ip.is_reserved
+        or ip.is_unspecified
+    )
+
+
+async def _resolve_host_ips(
+    host: str, port: int | None, scheme: str
+) -> list[ipaddress.IPv4Address | ipaddress.IPv6Address]:
+    """Resolve a host to the IP addresses it points at.
+
+    If the host is already an IP literal it is returned directly; otherwise
+    DNS resolution is performed and every returned address is checked.
+    """
+    try:
+        return [ipaddress.ip_address(host)]
+    except ValueError:
+        pass
+
+    default_port = 443 if scheme == "https" else 80
+    try:
+        infos = await asyncio.get_running_loop().getaddrinfo(
+            host, port or default_port, type=socket.SOCK_STREAM
+        )
+    except socket.gaierror as e:
+        raise McpError(ErrorData(
+            code=INVALID_PARAMS,
+            message=f"Failed to resolve host {host}: {e}",
+        ))
+
+    ips: list[ipaddress.IPv4Address | ipaddress.IPv6Address] = []
+    for info in infos:
+        # Strip any IPv6 zone/scope id (e.g. "fe80::1%eth0") before parsing so
+        # scoped addresses are still classified rather than silently skipped.
+        addr = info[4][0].split("%", 1)[0]
+        try:
+            ips.append(ipaddress.ip_address(addr))
+        except ValueError:
+            continue
+
+    # Fail closed: if resolution produced no address we could parse and
+    # classify, refuse rather than fall through to an empty (allow-all) check.
+    if not ips:
+        raise McpError(ErrorData(
+            code=INVALID_PARAMS,
+            message=f"Failed to resolve host {host} to a usable IP address.",
+        ))
+    return ips
+
+
+async def _validate_url_is_safe(url: str, *, check_private_ips: bool = True) -> None:
+    """Guard a URL against SSRF before it is fetched.
+
+    Rejects non-http(s) schemes and any URL whose host resolves to a
+    non-public IP address. The scheme (and host-presence) check is always
+    enforced; --allow-internal-ips only relaxes the private-IP check
+    (check_private_ips=False), so it never unlocks non-http(s) schemes.
+    """
+    parsed = urlparse(url)
+    if parsed.scheme not in ("http", "https"):
+        raise McpError(ErrorData(
+            code=INVALID_PARAMS,
+            message=f"Cannot fetch {url}: only http and https URLs are supported.",
+        ))
+
+    host = parsed.hostname
+    if not host:
+        raise McpError(ErrorData(
+            code=INVALID_PARAMS,
+            message=f"Cannot fetch {url}: URL has no host.",
+        ))
+
+    if not check_private_ips:
+        return
+
+    for ip in await _resolve_host_ips(host, parsed.port, parsed.scheme):
+        if _is_blocked_ip(ip):
+            raise McpError(ErrorData(
+                code=INVALID_PARAMS,
+                message=(
+                    f"Cannot fetch {url}: host {host} resolves to non-public IP "
+                    f"address {ip}. Fetching loopback, private, link-local, and "
+                    f"cloud-metadata addresses is blocked to prevent SSRF. Start "
+                    f"the server with --allow-internal-ips to override this."
+                ),
+            ))
+
+
+async def _get_following_redirects(
+    client,
+    url: str,
+    *,
+    headers: dict,
+    timeout: float,
+    allow_internal_ips: bool,
+):
+    """GET a URL, following redirects manually so every hop is re-validated.
+
+    httpx's automatic redirect handling never re-checks the destination, so
+    a public URL that 302-redirects to an internal address would bypass the
+    guard. Following redirects manually lets us validate each hop.
+    """
+    from httpx import URL
+
+    current_url = url
+    redirects = 0
+    while True:
+        # The scheme lock is always enforced; --allow-internal-ips only relaxes
+        # the private-IP check, so it can never unlock file:// et al.
+        await _validate_url_is_safe(
+            current_url, check_private_ips=not allow_internal_ips
+        )
+        response = await client.get(
+            current_url,
+            follow_redirects=False,
+            headers=headers,
+            timeout=timeout,
+        )
+        if response.status_code not in REDIRECT_STATUS_CODES:
+            return response
+        location = response.headers.get("location")
+        if not location:
+            return response
+        # Enforce the cap before following (and re-validating) the next hop.
+        if redirects >= MAX_REDIRECTS:
+            raise McpError(ErrorData(
+                code=INTERNAL_ERROR,
+                message=f"Cannot fetch {url}: exceeded the maximum of {MAX_REDIRECTS} redirects.",
+            ))
+        redirects += 1
+        current_url = str(URL(current_url).join(location))
+
+
+async def check_may_autonomously_fetch_url(url: str, user_agent: str, proxy_url: str | None = None, allow_internal_ips: bool = False) -> None:
     """
     Check if the URL can be fetched by the user agent according to the robots.txt file.
     Raises a McpError if not.
@@ -74,10 +287,12 @@ async def check_may_autonomously_fetch_url(url: str, user_agent: str, proxy_url:
 
     async with AsyncClient(proxy=proxy_url) as client:
         try:
-            response = await client.get(
+            response = await _get_following_redirects(
+                client,
                 robot_txt_url,
-                follow_redirects=True,
                 headers={"User-Agent": user_agent},
+                timeout=30,
+                allow_internal_ips=allow_internal_ips,
             )
         except HTTPError:
             raise McpError(ErrorData(
@@ -109,7 +324,7 @@ async def check_may_autonomously_fetch_url(url: str, user_agent: str, proxy_url:
 
 
 async def fetch_url(
-    url: str, user_agent: str, force_raw: bool = False, proxy_url: str | None = None
+    url: str, user_agent: str, force_raw: bool = False, proxy_url: str | None = None, allow_internal_ips: bool = False
 ) -> Tuple[str, str]:
     """
     Fetch the URL and return the content in a form ready for the LLM, as well as a prefix string with status information.
@@ -118,11 +333,12 @@ async def fetch_url(
 
     async with AsyncClient(proxy=proxy_url) as client:
         try:
-            response = await client.get(
+            response = await _get_following_redirects(
+                client,
                 url,
-                follow_redirects=True,
                 headers={"User-Agent": user_agent},
                 timeout=30,
+                allow_internal_ips=allow_internal_ips,
             )
         except HTTPError as e:
             raise McpError(ErrorData(code=INTERNAL_ERROR, message=f"Failed to fetch {url}: {e!r}"))
@@ -182,6 +398,7 @@ async def serve(
     custom_user_agent: str | None = None,
     ignore_robots_txt: bool = False,
     proxy_url: str | None = None,
+    allow_internal_ips: bool = False,
 ) -> None:
     """Run the fetch MCP server.
 
@@ -189,7 +406,24 @@ async def serve(
         custom_user_agent: Optional custom User-Agent string to use for requests
         ignore_robots_txt: Whether to ignore robots.txt restrictions
         proxy_url: Optional proxy URL to use for requests
+        allow_internal_ips: Allow fetching loopback/private/link-local/metadata
+            addresses (disables SSRF protection). Off by default.
     """
+    # The SSRF guard resolves each destination in this process, but a proxy
+    # resolves the hostname again for the forwarded request or CONNECT. The two
+    # answers can differ, so the guard cannot enforce the destination a proxy
+    # actually reaches. Warn rather than refuse, so existing proxy deployments
+    # keep working, and point at where enforcement has to live instead.
+    proxy_sources = _configured_proxy_sources(proxy_url)
+    if proxy_sources and not allow_internal_ips:
+        print(
+            f"WARNING: requests are routed through a proxy ({', '.join(proxy_sources)}), "
+            "so the SSRF guard cannot enforce the destination: the proxy resolves "
+            "hostnames itself and may reach an internal address this process "
+            "classified as public. Restrict internal destinations at the proxy.",
+            file=sys.stderr,
+        )
+
     server = Server("mcp-fetch")
     user_agent_autonomous = custom_user_agent or DEFAULT_USER_AGENT_AUTONOMOUS
     user_agent_manual = custom_user_agent or DEFAULT_USER_AGENT_MANUAL
@@ -232,10 +466,10 @@ async def call_tool(name, arguments: dict) -> list[TextContent]:
             raise McpError(ErrorData(code=INVALID_PARAMS, message="URL is required"))
 
         if not ignore_robots_txt:
-            await check_may_autonomously_fetch_url(url, user_agent_autonomous, proxy_url)
+            await check_may_autonomously_fetch_url(url, user_agent_autonomous, proxy_url, allow_internal_ips)
 
         content, prefix = await fetch_url(
-            url, user_agent_autonomous, force_raw=args.raw, proxy_url=proxy_url
+            url, user_agent_autonomous, force_raw=args.raw, proxy_url=proxy_url, allow_internal_ips=allow_internal_ips
         )
         original_length = len(content)
         if args.start_index >= original_length:
@@ -262,7 +496,7 @@ async def get_prompt(name: str, arguments: dict | None) -> GetPromptResult:
         url = arguments["url"]
 
         try:
-            content, prefix = await fetch_url(url, user_agent_manual, proxy_url=proxy_url)
+            content, prefix = await fetch_url(url, user_agent_manual, proxy_url=proxy_url, allow_internal_ips=allow_internal_ips)
             # TODO: after SDK bug is addressed, don't catch the exception
         except McpError as e:
             return GetPromptResult(
diff --git a/src/fetch/tests/test_server.py b/src/fetch/tests/test_server.py
index 96c1cb38c7..5575a4d234 100644
--- a/src/fetch/tests/test_server.py
+++ b/src/fetch/tests/test_server.py
@@ -1,5 +1,7 @@
 """Tests for the fetch MCP server."""
 
+import asyncio
+
 import pytest
 from unittest.mock import AsyncMock, patch, MagicMock
 from mcp.shared.exceptions import McpError
@@ -9,6 +11,8 @@
     get_robots_txt_url,
     check_may_autonomously_fetch_url,
     fetch_url,
+    _validate_url_is_safe,
+    _configured_proxy_sources,
     DEFAULT_USER_AGENT_AUTONOMOUS,
 )
 
@@ -91,6 +95,13 @@ def test_empty_content_returns_error(self):
 class TestCheckMayAutonomouslyFetchUrl:
     """Tests for check_may_autonomously_fetch_url function."""
 
+    @pytest.fixture(autouse=True)
+    def _skip_ssrf_guard(self):
+        """These tests exercise robots.txt handling, not the SSRF guard, and
+        mock the HTTP client, so stub host validation to avoid real DNS."""
+        with patch("mcp_server_fetch.server._validate_url_is_safe", new=AsyncMock()):
+            yield
+
     @pytest.mark.asyncio
     async def test_allows_when_robots_txt_404(self):
         """Test that fetching is allowed when robots.txt returns 404."""
@@ -187,6 +198,13 @@ async def test_blocks_when_robots_txt_disallows_all(self):
 class TestFetchUrl:
     """Tests for fetch_url function."""
 
+    @pytest.fixture(autouse=True)
+    def _skip_ssrf_guard(self):
+        """These tests exercise fetch/content handling, not the SSRF guard, and
+        mock the HTTP client, so stub host validation to avoid real DNS."""
+        with patch("mcp_server_fetch.server._validate_url_is_safe", new=AsyncMock()):
+            yield
+
     @pytest.mark.asyncio
     async def test_fetch_html_page(self):
         """Test fetching an HTML page returns markdown content."""
@@ -324,3 +342,220 @@ async def test_fetch_with_proxy(self):
 
             # Verify AsyncClient was called with proxy
             mock_client_class.assert_called_once_with(proxy="http://proxy.example.com:8080")
+
+
+class TestValidateUrlIsSafe:
+    """Tests for the SSRF guard (_validate_url_is_safe).
+
+    These use IP literals so no DNS resolution (and no network) is required.
+    """
+
+    @pytest.mark.asyncio
+    async def test_blocks_loopback_ipv4(self):
+        with pytest.raises(McpError):
+            await _validate_url_is_safe("http://127.0.0.1/")
+
+    @pytest.mark.asyncio
+    async def test_blocks_cloud_metadata_address(self):
+        with pytest.raises(McpError):
+            await _validate_url_is_safe("http://169.254.169.254/latest/meta-data/")
+
+    @pytest.mark.asyncio
+    async def test_blocks_private_ranges(self):
+        for host in ("10.0.0.1", "192.168.1.1", "172.16.0.1"):
+            with pytest.raises(McpError):
+                await _validate_url_is_safe(f"http://{host}/")
+
+    @pytest.mark.asyncio
+    async def test_blocks_unspecified_address(self):
+        with pytest.raises(McpError):
+            await _validate_url_is_safe("http://0.0.0.0/")
+
+    @pytest.mark.asyncio
+    async def test_blocks_carrier_grade_nat(self):
+        with pytest.raises(McpError):
+            await _validate_url_is_safe("http://100.64.0.1/")
+
+    @pytest.mark.asyncio
+    async def test_blocks_ipv4_compatible_ipv6_loopback(self):
+        with pytest.raises(McpError):
+            await _validate_url_is_safe("http://[::127.0.0.1]/")
+
+    @pytest.mark.asyncio
+    async def test_blocks_ipv6_loopback(self):
+        with pytest.raises(McpError):
+            await _validate_url_is_safe("http://[::1]/")
+
+    @pytest.mark.asyncio
+    async def test_blocks_ipv4_mapped_ipv6_loopback(self):
+        with pytest.raises(McpError):
+            await _validate_url_is_safe("http://[::ffff:127.0.0.1]/")
+
+    @pytest.mark.asyncio
+    async def test_blocks_6to4_loopback(self):
+        # 2002::/16 embeds the IPv4 address in bits 16-48: 2002:7f00:1:: is
+        # a route to 127.0.0.1 wherever a 6to4 relay is reachable.
+        with pytest.raises(McpError):
+            await _validate_url_is_safe("http://[2002:7f00:1::]/")
+
+    @pytest.mark.asyncio
+    async def test_blocks_6to4_cloud_metadata(self):
+        with pytest.raises(McpError):
+            await _validate_url_is_safe("http://[2002:a9fe:a9fe::]/")
+
+    @pytest.mark.asyncio
+    async def test_blocks_ipv6_site_local(self):
+        with pytest.raises(McpError):
+            await _validate_url_is_safe("http://[fec0::1]/")
+
+    @pytest.mark.asyncio
+    async def test_blocks_nat64_cloud_metadata(self):
+        with pytest.raises(McpError):
+            await _validate_url_is_safe("http://[64:ff9b::a9fe:a9fe]/")
+
+    @pytest.mark.asyncio
+    async def test_blocks_non_http_scheme(self):
+        with pytest.raises(McpError):
+            await _validate_url_is_safe("file:///etc/passwd")
+
+    @pytest.mark.asyncio
+    async def test_allows_public_ip_literal(self):
+        # Public IP literal: should not raise (no DNS needed).
+        await _validate_url_is_safe("https://1.1.1.1/")
+
+    @pytest.mark.asyncio
+    async def test_allows_public_ipv6_literal(self):
+        # The unwrapping above must not over-block ordinary global IPv6.
+        await _validate_url_is_safe("https://[2606:4700:4700::1111]/")
+
+    @pytest.mark.asyncio
+    async def test_blocked_url_rejected_by_fetch_url(self):
+        """The guard is enforced end-to-end through fetch_url by default."""
+        with pytest.raises(McpError):
+            await fetch_url(
+                "http://169.254.169.254/latest/meta-data/",
+                DEFAULT_USER_AGENT_AUTONOMOUS,
+            )
+
+    @pytest.mark.asyncio
+    async def test_allow_internal_ips_bypasses_guard(self):
+        """With allow_internal_ips=True the guard is skipped (no McpError from
+        validation); the request proceeds to the mocked client."""
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.text = "internal"
+        mock_response.headers = {"content-type": "text/plain"}
+
+        with patch("httpx.AsyncClient") as mock_client_class:
+            mock_client = AsyncMock()
+            mock_client.get = AsyncMock(return_value=mock_response)
+            mock_client_class.return_value.__aenter__ = AsyncMock(return_value=mock_client)
+            mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None)
+
+            content, _ = await fetch_url(
+                "http://127.0.0.1/secret",
+                DEFAULT_USER_AGENT_AUTONOMOUS,
+                allow_internal_ips=True,
+            )
+            assert content == "internal"
+
+    @pytest.mark.asyncio
+    async def test_blocks_redirect_from_public_to_internal_ip(self):
+        """A public URL that redirects to an internal/metadata IP is rejected
+        at the redirect hop, before the internal host is ever fetched."""
+        redirect_response = MagicMock()
+        redirect_response.status_code = 302
+        redirect_response.headers = {
+            "location": "http://169.254.169.254/latest/meta-data/"
+        }
+
+        with patch("httpx.AsyncClient") as mock_client_class:
+            mock_client = AsyncMock()
+            mock_client.get = AsyncMock(return_value=redirect_response)
+            mock_client_class.return_value.__aenter__ = AsyncMock(return_value=mock_client)
+            mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None)
+
+            with pytest.raises(McpError):
+                await fetch_url(
+                    "http://1.1.1.1/redirect",
+                    DEFAULT_USER_AGENT_AUTONOMOUS,
+                )
+
+            # Only the initial public URL should have been requested; the
+            # internal redirect target must be blocked before any fetch.
+            assert mock_client.get.await_count == 1
+            assert mock_client.get.await_args_list[0].args[0] == "http://1.1.1.1/redirect"
+
+    @pytest.mark.asyncio
+    async def test_strips_ipv6_zone_id_and_blocks_link_local(self):
+        """A resolved IPv6 address carrying a zone id (fe80::1%eth0) must have
+        the zone stripped and still be classified as link-local (blocked),
+        not silently skipped."""
+        loop = asyncio.get_running_loop()
+        fake = [(0, 0, 0, "", ("fe80::1%eth0", 80, 0, 3))]
+        with patch.object(loop, "getaddrinfo", new=AsyncMock(return_value=fake)):
+            with pytest.raises(McpError):
+                await _validate_url_is_safe("http://router.local/")
+
+    @pytest.mark.asyncio
+    async def test_fails_closed_when_no_resolved_ip_parses(self):
+        """If resolution yields no address we can parse, fail closed rather than
+        fall through to an empty (allow-all) IP check."""
+        loop = asyncio.get_running_loop()
+        fake = [(0, 0, 0, "", ("not-an-ip-at-all", 80, 0, 0))]
+        with patch.object(loop, "getaddrinfo", new=AsyncMock(return_value=fake)):
+            with pytest.raises(McpError):
+                await _validate_url_is_safe("http://weird.example/")
+
+    @pytest.mark.asyncio
+    async def test_allow_internal_ips_still_blocks_non_http_scheme(self):
+        """--allow-internal-ips relaxes only the private-IP check; the scheme
+        lock stays on, so file:// is still refused and no request is made."""
+        with patch("httpx.AsyncClient") as mock_client_class:
+            mock_client = AsyncMock()
+            mock_client.get = AsyncMock()
+            mock_client_class.return_value.__aenter__ = AsyncMock(return_value=mock_client)
+            mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None)
+
+            with pytest.raises(McpError):
+                await fetch_url(
+                    "file:///etc/passwd",
+                    DEFAULT_USER_AGENT_AUTONOMOUS,
+                    allow_internal_ips=True,
+                )
+
+            mock_client.get.assert_not_called()
+
+
+class TestProxyDetection:
+    """Tests for detecting proxy settings that the SSRF guard cannot enforce."""
+
+    def test_no_proxy_configured(self, monkeypatch):
+        for name in ("HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY"):
+            monkeypatch.delenv(name, raising=False)
+            monkeypatch.delenv(name.lower(), raising=False)
+        assert _configured_proxy_sources(None) == []
+
+    def test_detects_proxy_url_argument(self, monkeypatch):
+        for name in ("HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY"):
+            monkeypatch.delenv(name, raising=False)
+            monkeypatch.delenv(name.lower(), raising=False)
+        assert _configured_proxy_sources("http://proxy:8080") == ["--proxy-url"]
+
+    def test_detects_uppercase_env_proxy(self, monkeypatch):
+        monkeypatch.setenv("HTTPS_PROXY", "http://proxy:8080")
+        assert "HTTPS_PROXY" in _configured_proxy_sources(None)
+
+    def test_detects_lowercase_env_proxy(self, monkeypatch):
+        # httpx honors the lowercase spellings too, via trust_env.
+        for name in ("HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY"):
+            monkeypatch.delenv(name, raising=False)
+        monkeypatch.setenv("all_proxy", "socks5://proxy:1080")
+        assert "ALL_PROXY" in _configured_proxy_sources(None)
+
+    def test_empty_env_proxy_is_not_a_proxy(self, monkeypatch):
+        for name in ("HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY"):
+            monkeypatch.delenv(name, raising=False)
+            monkeypatch.delenv(name.lower(), raising=False)
+        monkeypatch.setenv("HTTP_PROXY", "")
+        assert _configured_proxy_sources(None) == []