diff --git a/fastmcp_slim/fastmcp/client/auth/oauth.py b/fastmcp_slim/fastmcp/client/auth/oauth.py
index aca909e4a..a854501e6 100644
--- a/fastmcp_slim/fastmcp/client/auth/oauth.py
+++ b/fastmcp_slim/fastmcp/client/auth/oauth.py
@@ -415,6 +415,7 @@ class OAuth(OAuthClientProvider):
                     return AuthorizationCodeResult(
                         code=result.code,  # type: ignore[arg-type]  # ty:ignore[invalid-argument-type]
                         state=result.state,
+                        iss=result.iss,
                     )
             except TimeoutError as e:
                 raise TimeoutError(
diff --git a/fastmcp_slim/fastmcp/client/oauth_callback.py b/fastmcp_slim/fastmcp/client/oauth_callback.py
index 3d7a74223..ade1764dc 100644
--- a/fastmcp_slim/fastmcp/client/oauth_callback.py
+++ b/fastmcp_slim/fastmcp/client/oauth_callback.py
@@ -82,6 +82,11 @@ class CallbackResponse:
     state: str | None = None
     error: str | None = None
     error_description: str | None = None
+    # RFC 9207: the authorization server's issuer identifier, sent on both
+    # success and error redirects once OAuthProxy advertises
+    # `authorization_response_iss_parameter_supported`. Must be captured
+    # here or `from_dict`'s annotation filter silently drops it.
+    iss: str | None = None
 
     @classmethod
     def from_dict(cls, data: dict[str, str]) -> CallbackResponse:
@@ -98,6 +103,12 @@ class OAuthCallbackResult:
     code: str | None = None
     state: str | None = None
     error: Exception | None = None
+    # RFC 9207 issuer identifier, captured on both success and error
+    # callbacks. The MCP SDK's `validate_authorization_response_iss` only
+    # consumes this on the success path (via `AuthorizationCodeResult.iss`),
+    # but it is stored unconditionally here so the error path never silently
+    # drops it either.
+    iss: str | None = None
 
 
 def create_oauth_callback_server(
@@ -127,6 +138,7 @@ def create_oauth_callback_server(
         code: str | None = None,
         state: str | None = None,
         error: Exception | None = None,
+        iss: str | None = None,
     ) -> None:
         """Store the first callback result and ignore subsequent requests."""
         if result_container is None or result_ready is None or result_ready.is_set():
@@ -135,6 +147,7 @@ def create_oauth_callback_server(
         result_container.code = code
         result_container.state = state
         result_container.error = error
+        result_container.iss = iss
         result_ready.set()
 
     async def callback_handler(request: Request):
@@ -151,8 +164,13 @@ def create_oauth_callback_server(
             else:
                 user_message = f"Authorization failed: {error_desc}"
 
-            # Store error and signal completion if result tracking provided
-            store_result_once(error=RuntimeError(user_message))
+            # Store error and signal completion if result tracking provided.
+            # RFC 9207: `iss` is captured here too, even though the callback
+            # ultimately raises instead of returning a result, so it isn't
+            # silently dropped for callers that want to inspect it.
+            store_result_once(
+                error=RuntimeError(user_message), iss=callback_response.iss
+            )
 
             return create_secure_html_response(
                 create_callback_html(
@@ -166,7 +184,9 @@ def create_oauth_callback_server(
             user_message = "No authorization code was received from the server."
 
             # Store error and signal completion if result tracking provided
-            store_result_once(error=RuntimeError(user_message))
+            store_result_once(
+                error=RuntimeError(user_message), iss=callback_response.iss
+            )
 
             return create_secure_html_response(
                 create_callback_html(
@@ -183,7 +203,9 @@ def create_oauth_callback_server(
             )
 
             # Store error and signal completion if result tracking provided
-            store_result_once(error=RuntimeError(user_message))
+            store_result_once(
+                error=RuntimeError(user_message), iss=callback_response.iss
+            )
 
             return create_secure_html_response(
                 create_callback_html(
@@ -196,6 +218,7 @@ def create_oauth_callback_server(
         # Success case - store result and signal completion if result tracking provided
         store_result_once(
             code=callback_response.code,
+            iss=callback_response.iss,
             state=callback_response.state,
         )
 
diff --git a/fastmcp_slim/fastmcp/server/auth/handlers/authorize.py b/fastmcp_slim/fastmcp/server/auth/handlers/authorize.py
index f91b266b7..b5086ea75 100644
--- a/fastmcp_slim/fastmcp/server/auth/handlers/authorize.py
+++ b/fastmcp_slim/fastmcp/server/auth/handlers/authorize.py
@@ -15,6 +15,7 @@ from __future__ import annotations
 
 import json
 from typing import TYPE_CHECKING
+from urllib.parse import parse_qs, urlparse
 
 from mcp.server.auth.handlers.authorize import (
     AuthorizationHandler as SDKAuthorizationHandler,
@@ -23,6 +24,7 @@ from pydantic import AnyHttpUrl
 from starlette.requests import Request
 from starlette.responses import Response
 
+from fastmcp.server.auth.redirect_validation import build_client_redirect
 from fastmcp.utilities.logging import get_logger
 from fastmcp.utilities.ui import (
     INFO_BOX_STYLES,
@@ -189,6 +191,10 @@ class AuthorizationHandler(SDKAuthorizationHandler):
             server_icon_url: Optional server icon URL for branding
         """
         super().__init__(provider)
+        # Unnormalized on purpose: this must match the discovery document's
+        # `issuer` field byte-for-byte per RFC 9207, and that field is built
+        # from the same unmodified base_url (see OAuthProxy.get_routes()).
+        self._issuer = str(base_url)
         self._base_url = str(base_url).rstrip("/")
         self._server_name = server_name
         self._server_icon_url = server_icon_url
@@ -209,6 +215,31 @@ class AuthorizationHandler(SDKAuthorizationHandler):
         # Call the SDK handler
         response = await super().handle(request)
 
+        if 300 <= response.status_code < 400 and "location" in response.headers:
+            redirect_url = response.headers["location"]
+            redirect_params = parse_qs(urlparse(redirect_url).query)
+            # RFC 9207: any client-facing authorization response — success
+            # (`code`) or error (`error`) — must carry `iss`. The base SDK
+            # handler's redirect target is normally `/consent` or the
+            # upstream IdP (neither carries `code`/`error`), but a provider
+            # can override `authorize()` to redirect straight back to the
+            # client (e.g. when consent/upstream is skipped entirely), so
+            # this must not be gated on "error" alone.
+            if "error" in redirect_params or "code" in redirect_params:
+                # `build_client_redirect` owns the "set `iss` idempotently"
+                # invariant: a provider's `authorize()` override (or the
+                # client's own registered redirect_uri) may already carry an
+                # `iss` — matching or not — and RFC 6749 §3.1 forbids a
+                # response parameter from appearing more than once. A
+                # mismatched existing value is already unusable to a
+                # spec-compliant client (it validates `iss` against the
+                # discovery document's `issuer`, i.e. `self._issuer`), so
+                # the helper corrects it to the canonical value rather than
+                # leaving it broken or appending a duplicate.
+                response.headers["location"] = build_client_redirect(
+                    redirect_url, {}, iss=self._issuer
+                )
+
         # Check if this is a client not found error
         if response.status_code == 400:
             # Try to extract client_id from request for enhanced error
diff --git a/fastmcp_slim/fastmcp/server/auth/oauth_proxy/consent.py b/fastmcp_slim/fastmcp/server/auth/oauth_proxy/consent.py
index c25194936..a42e6dfab 100644
--- a/fastmcp_slim/fastmcp/server/auth/oauth_proxy/consent.py
+++ b/fastmcp_slim/fastmcp/server/auth/oauth_proxy/consent.py
@@ -23,7 +23,10 @@ from starlette.responses import HTMLResponse, RedirectResponse
 
 from fastmcp.server.auth.oauth_proxy.models import ProxyDCRClient
 from fastmcp.server.auth.oauth_proxy.ui import create_consent_html
-from fastmcp.server.auth.redirect_validation import validate_redirect_uri
+from fastmcp.server.auth.redirect_validation import (
+    build_client_redirect,
+    validate_redirect_uri,
+)
 from fastmcp.utilities.logging import get_logger
 from fastmcp.utilities.ui import create_secure_html_response
 
@@ -388,9 +391,12 @@ class ConsentMixin:
                         "error": "access_denied",
                         "state": txn.get("client_state") or "",
                     }
-                    sep = "&" if "?" in txn["client_redirect_uri"] else "?"
                     return RedirectResponse(
-                        url=f"{txn['client_redirect_uri']}{sep}{urlencode(callback_params)}",
+                        url=build_client_redirect(
+                            txn["client_redirect_uri"],
+                            callback_params,
+                            iss=str(self.base_url),
+                        ),
                         status_code=302,
                     )
             else:
@@ -563,9 +569,8 @@ class ConsentMixin:
                 "error": "access_denied",
                 "state": txn.get("client_state") or "",
             }
-            sep = "&" if "?" in txn["client_redirect_uri"] else "?"
-            client_callback_url = (
-                f"{txn['client_redirect_uri']}{sep}{urlencode(callback_params)}"
+            client_callback_url = build_client_redirect(
+                txn["client_redirect_uri"], callback_params, iss=str(self.base_url)
             )
             response = RedirectResponse(url=client_callback_url, status_code=302)
 
diff --git a/fastmcp_slim/fastmcp/server/auth/oauth_proxy/proxy.py b/fastmcp_slim/fastmcp/server/auth/oauth_proxy/proxy.py
index fd6a4f3dc..f8b3bd3dc 100644
--- a/fastmcp_slim/fastmcp/server/auth/oauth_proxy/proxy.py
+++ b/fastmcp_slim/fastmcp/server/auth/oauth_proxy/proxy.py
@@ -26,7 +26,6 @@ from collections import OrderedDict
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from typing import Any, Literal
-from urllib.parse import urlencode
 
 import anyio
 import httpx2
@@ -103,7 +102,10 @@ from fastmcp.server.auth.oauth_proxy.models import (
 )
 from fastmcp.server.auth.oauth_proxy.ui import create_error_html
 from fastmcp.server.auth.oauth_proxy.upstream import AsyncOAuth2Client
-from fastmcp.server.auth.redirect_validation import validate_redirect_uri
+from fastmcp.server.auth.redirect_validation import (
+    build_client_redirect,
+    validate_redirect_uri,
+)
 from fastmcp.utilities.auth import parse_scopes
 from fastmcp.utilities.logging import get_logger
 
@@ -2291,10 +2293,8 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
                         methods=["POST", "OPTIONS"],
                     )
                 )
-            elif (
-                (self._cimd_manager is not None or self._identity_assertion is not None)
-                and isinstance(route, Route)
-                and route.path.startswith("/.well-known/oauth-authorization-server")
+            elif isinstance(route, Route) and route.path.startswith(
+                "/.well-known/oauth-authorization-server"
             ):
                 client_registration_options = (
                     self.client_registration_options or ClientRegistrationOptions()
@@ -2307,6 +2307,11 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
                     revocation_options,
                     supports_identity_assertion=self._identity_assertion is not None,
                 )
+                # RFC 9207: every authorization response we issue carries an
+                # `iss` matching this issuer byte-for-byte, so this route must
+                # always be overridden to advertise support — not just when
+                # CIMD or identity assertion is also enabled.
+                metadata.authorization_response_iss_parameter_supported = True
                 if self._cimd_manager is not None:
                     metadata.client_id_metadata_document_supported = True
                     existing = metadata.token_endpoint_auth_methods_supported or []
@@ -2423,9 +2428,12 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
                     }
                     if error_description:
                         error_params["error_description"] = error_description
-                    separator = "&" if "?" in client_redirect_uri else "?"
                     return RedirectResponse(
-                        url=f"{client_redirect_uri}{separator}{urlencode(error_params)}",
+                        url=build_client_redirect(
+                            client_redirect_uri,
+                            error_params,
+                            iss=str(self.base_url),
+                        ),
                         status_code=302,
                     )
                 # No trusted redirect_uri available — show local error page
@@ -2591,10 +2599,8 @@ class OAuthProxy(OAuthProvider, ConsentMixin):
                 "state": client_state,
             }
 
-            # Add query parameters to client redirect URI
-            separator = "&" if "?" in client_redirect_uri else "?"
-            client_callback_url = (
-                f"{client_redirect_uri}{separator}{urlencode(callback_params)}"
+            client_callback_url = build_client_redirect(
+                client_redirect_uri, callback_params, iss=str(self.base_url)
             )
 
             logger.debug(f"Forwarding to client callback for transaction {txn_id}")
diff --git a/fastmcp_slim/fastmcp/server/auth/redirect_validation.py b/fastmcp_slim/fastmcp/server/auth/redirect_validation.py
index cdc62cc4f..fb7cba969 100644
--- a/fastmcp_slim/fastmcp/server/auth/redirect_validation.py
+++ b/fastmcp_slim/fastmcp/server/auth/redirect_validation.py
@@ -5,7 +5,7 @@ protecting against userinfo-based bypass attacks like http://localhost@evil.com.
 """
 
 import fnmatch
-from urllib.parse import unquote, urlparse
+from urllib.parse import unquote, urlencode, urlparse, urlunparse
 
 from pydantic import AnyUrl
 
@@ -19,6 +19,99 @@ UNSAFE_REDIRECT_URI_SCHEMES = frozenset(
 )
 
 
+def add_query_params(url: str, params: dict[str, str]) -> str:
+    """Append query parameters to a URL while preserving existing parameters.
+
+    The existing query string is appended to verbatim rather than decoded
+    and re-serialized, since registered redirect URIs may carry opaque or
+    signed query strings whose exact bytes matter to the receiving client
+    (for example, a valueless `?flag` must not become `?flag=`, and
+    non-UTF-8 percent-encoded sequences must not be replaced).
+    """
+    parsed = urlparse(url)
+    new_query = urlencode(params)
+    query = f"{parsed.query}&{new_query}" if parsed.query else new_query
+    return urlunparse(parsed._replace(query=query))
+
+
+def replace_query_param(url: str, key: str, value: str) -> str:
+    """Replace the first occurrence of `key` in a URL's query string in place.
+
+    Like `add_query_params`, this does not round-trip the query through
+    `parse_qsl`/`urlencode`: every segment other than the matched one is
+    passed through byte-for-byte, so opaque or non-UTF-8 percent-encoded
+    values elsewhere in the query are left untouched. Only the matched
+    segment's encoding is replaced (with `key=value`, freshly
+    `urlencode`d). If `key` is not present, it is appended, matching
+    `add_query_params`'s behavior.
+    """
+    parsed = urlparse(url)
+    segments = parsed.query.split("&") if parsed.query else []
+    new_segment = urlencode({key: value})
+
+    replaced = False
+    new_segments: list[str] = []
+    for segment in segments:
+        segment_key = segment.split("=", 1)[0]
+        if not replaced and unquote(segment_key) == key:
+            new_segments.append(new_segment)
+            replaced = True
+        else:
+            new_segments.append(segment)
+    if not replaced:
+        new_segments.append(new_segment)
+
+    return urlunparse(parsed._replace(query="&".join(new_segments)))
+
+
+def build_client_redirect(url: str, params: dict[str, str], *, iss: str) -> str:
+    """Build a client-facing authorization redirect that carries exactly one `iss`.
+
+    Every redirect this server sends back to an OAuth client from the
+    authorization endpoint -- success (carrying `code`) or error (carrying
+    `error`) -- must carry the proxy's RFC 9207 issuer exactly once (RFC
+    6749 §3.1 forbids a response parameter from appearing more than once).
+    A registered redirect_uri can legitimately carry its own `iss` query
+    parameter already (e.g. `https://client.example/callback?iss=tenant`),
+    so blindly appending the server's issuer on top of that would duplicate
+    it -- this is what every client-facing redirect site must get right,
+    and the reason this helper exists instead of five call sites each
+    reimplementing the same invariant by hand.
+
+    `params` is appended to `url` via `add_query_params` (verbatim, without
+    re-encoding the existing query -- see that function's docstring), and
+    `iss` is then set idempotently via `replace_query_param`: an existing
+    occurrence -- whether contributed by the registered redirect_uri or
+    already present in `url` -- is overwritten with the canonical value;
+    otherwise `iss` is appended.
+
+    `iss` is keyword-only and required so a caller cannot forget to pass
+    it. `params` must not itself contain an `"iss"` key -- pass it via the
+    `iss` keyword instead, so there is exactly one place the value can come
+    from.
+
+    Args:
+        url: The redirect target -- normally the client's registered
+            redirect_uri.
+        params: The response parameters to append (e.g. `code`/`state`, or
+            `error`/`error_description`). Must not include `"iss"`.
+        iss: The canonical RFC 9207 issuer, byte-for-byte equal to the
+            discovery document's `issuer` (`str(self.base_url)` /
+            `self._issuer` -- never the rstripped `self._base_url`).
+
+    Returns:
+        `url` with `params` appended and exactly one `iss` query parameter
+        set to `iss`.
+    """
+    if "iss" in params:
+        raise ValueError(
+            "params must not include 'iss' -- pass it via the 'iss' keyword"
+        )
+    if params:
+        url = add_query_params(url, params)
+    return replace_query_param(url, "iss", iss)
+
+
 def _parse_host_port(netloc: str) -> tuple[str | None, str | None]:
     """Parse host and port from netloc, handling wildcards.
 
diff --git a/fastmcp_slim/fastmcp/utilities/tests.py b/fastmcp_slim/fastmcp/utilities/tests.py
index 77b31ba36..30d354789 100644
--- a/fastmcp_slim/fastmcp/utilities/tests.py
+++ b/fastmcp_slim/fastmcp/utilities/tests.py
@@ -270,6 +270,7 @@ class HeadlessOAuth(OAuth):
 
             auth_code = query_params["code"][0]
             state = query_params.get("state", [None])[0]
-            return AuthorizationCodeResult(code=auth_code, state=state)
+            iss = query_params.get("iss", [None])[0]
+            return AuthorizationCodeResult(code=auth_code, state=state, iss=iss)
         else:
             raise RuntimeError(f"Authorization failed: {response.status_code}")
diff --git a/tests/client/auth/test_oauth_client.py b/tests/client/auth/test_oauth_client.py
index f538a89ca..89427fd24 100644
--- a/tests/client/auth/test_oauth_client.py
+++ b/tests/client/auth/test_oauth_client.py
@@ -3,6 +3,7 @@ import time
 from unittest.mock import patch
 from urllib.parse import urlparse
 
+import anyio
 import httpx2
 import pytest
 from key_value.aio.stores.memory import MemoryStore
@@ -177,6 +178,44 @@ async def test_expired_dynamic_registration_is_retried():
     assert provider.registration_count == 2
 
 
+async def test_oauth_callback_handler_propagates_iss_to_authorization_code_result():
+    """RFC 9207: `OAuth.callback_handler()` (the production, non-headless path)
+    must carry `iss` from the callback query string all the way into the
+    `AuthorizationCodeResult` handed back to the MCP SDK.
+
+    The MCP SDK's `validate_authorization_response_iss` raises when the
+    authorization server metadata advertises
+    `authorization_response_iss_parameter_supported` and the result it
+    receives has no `iss` -- so if this hop drops it, every production OAuth
+    login against an RFC 9207-compliant server (like OAuthProxy) fails, even
+    though the server sent `iss` correctly. `HeadlessOAuth` already carries
+    `iss` through for tests -- this test exercises the real `OAuth` class
+    that production clients actually use.
+    """
+    oauth = OAuth(mcp_url="http://127.0.0.1:9999")
+
+    async def send_callback():
+        await anyio.sleep(0.1)
+        async with httpx2.AsyncClient() as client:
+            response = await client.get(
+                f"http://{oauth._callback_host}:{oauth.redirect_port}/callback",
+                params={
+                    "code": "auth-code-123",
+                    "state": "state-xyz",
+                    "iss": "https://issuer.example.com",
+                },
+            )
+            assert response.status_code == 200
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(send_callback)
+        result = await oauth.callback_handler()
+
+    assert result.code == "auth-code-123"
+    assert result.state == "state-xyz"
+    assert result.iss == "https://issuer.example.com"
+
+
 class TestOAuthClientUrlHandling:
     """Tests for OAuth client URL handling (issue #2573)."""
 
diff --git a/tests/client/test_oauth_callback_race.py b/tests/client/test_oauth_callback_race.py
index 43efddee8..83622a86a 100644
--- a/tests/client/test_oauth_callback_race.py
+++ b/tests/client/test_oauth_callback_race.py
@@ -48,3 +48,86 @@ def test_oauth_callback_server_uses_configured_host():
     server = create_oauth_callback_server(port=find_available_port(), host="localhost")
 
     assert server.config.host == "localhost"
+
+
+async def test_oauth_callback_result_captures_iss():
+    """RFC 9207: the `iss` query parameter must survive from the raw callback
+    request through to `OAuthCallbackResult`, the same as `code` and `state`.
+
+    OAuthProxy advertises `authorization_response_iss_parameter_supported` and
+    includes `iss` on every authorization redirect. If the callback server's
+    query-parsing chain (CallbackResponse.from_dict -> store_result_once ->
+    OAuthCallbackResult) drops it, the MCP SDK's `validate_authorization_response_iss`
+    rejects an otherwise-successful callback.
+    """
+    port = find_available_port()
+    result = OAuthCallbackResult()
+    result_ready = anyio.Event()
+    server = create_oauth_callback_server(
+        port=port,
+        result_container=result,
+        result_ready=result_ready,
+    )
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(server.serve)
+
+        await anyio.sleep(0.05)
+
+        async with httpx2.AsyncClient() as client:
+            response = await client.get(
+                f"http://127.0.0.1:{port}/callback",
+                params={
+                    "code": "good",
+                    "state": "s1",
+                    "iss": "https://issuer.example.com",
+                },
+            )
+            assert response.status_code == 200
+
+        await result_ready.wait()
+
+        assert result.error is None
+        assert result.code == "good"
+        assert result.state == "s1"
+        assert result.iss == "https://issuer.example.com"
+
+        tg.cancel_scope.cancel()
+
+
+async def test_oauth_callback_result_captures_iss_on_error():
+    """RFC 9207 applies to error redirects too -- the server emits `iss` on
+    them, so the callback server must not silently drop it while building the
+    error result.
+    """
+    port = find_available_port()
+    result = OAuthCallbackResult()
+    result_ready = anyio.Event()
+    server = create_oauth_callback_server(
+        port=port,
+        result_container=result,
+        result_ready=result_ready,
+    )
+
+    async with anyio.create_task_group() as tg:
+        tg.start_soon(server.serve)
+
+        await anyio.sleep(0.05)
+
+        async with httpx2.AsyncClient() as client:
+            response = await client.get(
+                f"http://127.0.0.1:{port}/callback",
+                params={
+                    "error": "access_denied",
+                    "state": "s1",
+                    "iss": "https://issuer.example.com",
+                },
+            )
+            assert response.status_code == 400
+
+        await result_ready.wait()
+
+        assert result.error is not None
+        assert result.iss == "https://issuer.example.com"
+
+        tg.cancel_scope.cancel()
diff --git a/tests/server/auth/oauth_proxy/test_oauth_proxy.py b/tests/server/auth/oauth_proxy/test_oauth_proxy.py
index f3b962bc4..2a035bd5e 100644
--- a/tests/server/auth/oauth_proxy/test_oauth_proxy.py
+++ b/tests/server/auth/oauth_proxy/test_oauth_proxy.py
@@ -1,11 +1,14 @@
 """Tests for OAuth proxy initialization and configuration."""
 
 import time
+from unittest.mock import AsyncMock, patch
 from urllib.parse import parse_qs, urlparse
 
 import httpx2
 import pytest
 from key_value.aio.stores.memory import MemoryStore
+from mcp.shared.auth import OAuthClientInformationFull
+from pydantic import AnyUrl
 from starlette.applications import Starlette
 
 from fastmcp.server.auth.oauth_proxy import OAuthProxy
@@ -238,6 +241,34 @@ class TestOAuthProxyInitialization:
             "none",
         }
 
+    async def test_metadata_advertises_authorization_response_issuer_parameter(
+        self, jwt_verifier
+    ):
+        """OAuth metadata should advertise RFC 9207 authorization response issuers."""
+        proxy = OAuthProxy(
+            upstream_authorization_endpoint="https://auth.example.com/authorize",
+            upstream_token_endpoint="https://auth.example.com/token",
+            upstream_client_id="client-123",
+            upstream_client_secret="secret-456",
+            token_verifier=jwt_verifier,
+            base_url="https://api.example.com",
+            jwt_signing_key="test-secret",
+            client_storage=MemoryStore(),
+        )
+
+        app = Starlette(routes=proxy.get_routes())
+        transport = httpx2.ASGITransport(app=app)
+
+        async with httpx2.AsyncClient(
+            transport=transport, base_url="https://api.example.com"
+        ) as client:
+            response = await client.get("/.well-known/oauth-authorization-server")
+
+        assert response.status_code == 200
+        metadata = response.json()
+        assert metadata["issuer"] == "https://api.example.com/"
+        assert metadata["authorization_response_iss_parameter_supported"] is True
+
 
 class TestOptionalClientSecret:
     """Tests for OAuthProxy without upstream_client_secret."""
@@ -360,6 +391,61 @@ class TestIdpCallbackErrorForwarding:
         assert params["error"] == ["access_denied"]
         assert params["error_description"] == ["User denied access"]
         assert params["state"] == [client_state]
+        assert params["iss"] == ["https://myserver.com/"]
+
+    async def test_error_redirect_does_not_duplicate_iss_already_in_redirect_uri(
+        self, oauth_proxy
+    ):
+        """RFC 9207 P2 regression: a registered redirect_uri may already
+        carry its own `iss` query parameter (e.g. a multi-tenant client
+        encoding its tenant in the callback URL). Forwarding an IdP error
+        must not append a second `iss` on top of it -- RFC 6749 §3.1
+        forbids a response parameter appearing more than once -- and every
+        other query byte on the registered URI (a valueless `flag` and a
+        non-UTF-8 percent-encoded `sig`) must survive untouched.
+        """
+        txn_id = "test-txn-dup-iss"
+        client_redirect_uri = (
+            "http://localhost:12345/callback?iss=tenant&flag&sig=%FF%FE"
+        )
+        client_state = "client-state-abc"
+
+        transaction = OAuthTransaction(
+            txn_id=txn_id,
+            client_id="test-client",
+            client_redirect_uri=client_redirect_uri,
+            client_state=client_state,
+            code_challenge=None,
+            code_challenge_method="S256",
+            scopes=["read"],
+            created_at=time.time(),
+        )
+        await oauth_proxy._transaction_store.put(key=txn_id, value=transaction)
+
+        app = Starlette(routes=oauth_proxy.get_routes())
+        transport = httpx2.ASGITransport(app=app)
+
+        async with httpx2.AsyncClient(
+            transport=transport,
+            base_url="https://myserver.com",
+            follow_redirects=False,
+        ) as client:
+            response = await client.get(
+                f"/auth/callback?error=access_denied&state={txn_id}"
+            )
+
+        assert response.status_code == 302
+        location = response.headers["location"]
+        query = urlparse(location).query
+        params = parse_qs(query)
+
+        # Exactly one `iss`, corrected to the canonical value -- a
+        # duplicate would make this list have length 2.
+        assert params["iss"] == ["https://myserver.com/"]
+        # Other query bytes from the registered redirect_uri survive
+        # byte-for-byte.
+        assert "flag" in query
+        assert "sig=%FF%FE" in query
 
     async def test_error_with_unsafe_transaction_redirect_returns_html_error(
         self, oauth_proxy
@@ -412,3 +498,97 @@ class TestIdpCallbackErrorForwarding:
             )
 
         assert response.status_code == 400
+
+
+class TestIdpCallbackSuccessForwarding:
+    """Tests for the success (`code`) path in the IdP callback."""
+
+    async def test_success_redirect_does_not_duplicate_iss_already_in_redirect_uri(
+        self, jwt_verifier
+    ):
+        """RFC 9207 P2 regression at the success-redirect call site: a
+        registered redirect_uri already carrying `iss` must end up with
+        exactly one `iss` (the canonical value) after the proxy forwards
+        the exchanged authorization code, and every other query byte on the
+        registered URI must survive untouched.
+        """
+        # Consent is disabled here because this test exercises callback
+        # forwarding, not the consent-binding-cookie check that the
+        # standard consent flow additionally requires.
+        oauth_proxy = OAuthProxy(
+            upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
+            upstream_token_endpoint="https://github.com/login/oauth/access_token",
+            upstream_client_id="test-client-id",
+            upstream_client_secret="test-client-secret",
+            token_verifier=jwt_verifier,
+            base_url="https://myserver.com",
+            redirect_path="/auth/callback",
+            jwt_signing_key="test-secret",
+            client_storage=MemoryStore(),
+            require_authorization_consent=False,
+        )
+
+        client_id = "success-dup-iss-client"
+        client_redirect_uri = (
+            "http://localhost:12345/callback?iss=tenant&flag&sig=%FF%FE"
+        )
+        client_info = OAuthClientInformationFull(
+            client_id=client_id,
+            client_secret="test-secret",
+            redirect_uris=[AnyUrl(client_redirect_uri)],
+        )
+        await oauth_proxy.register_client(client_info)
+
+        txn_id = "test-txn-success-dup-iss"
+        transaction = OAuthTransaction(
+            txn_id=txn_id,
+            client_id=client_id,
+            client_redirect_uri=client_redirect_uri,
+            client_state="client-state-success",
+            code_challenge=None,
+            code_challenge_method="S256",
+            scopes=["read"],
+            created_at=time.time(),
+        )
+        await oauth_proxy._transaction_store.put(key=txn_id, value=transaction)
+
+        app = Starlette(routes=oauth_proxy.get_routes())
+        transport = httpx2.ASGITransport(app=app)
+
+        with patch(
+            "fastmcp.server.auth.oauth_proxy.proxy.AsyncOAuth2Client"
+        ) as MockClient:
+            mock_client = AsyncMock()
+            mock_client.fetch_token = AsyncMock(
+                return_value={
+                    "access_token": "upstream-access-token",
+                    "refresh_token": "upstream-refresh-token",
+                    "expires_in": 3600,
+                    "token_type": "Bearer",
+                }
+            )
+            MockClient.return_value = mock_client
+
+            async with httpx2.AsyncClient(
+                transport=transport,
+                base_url="https://myserver.com",
+                follow_redirects=False,
+            ) as client:
+                response = await client.get(
+                    f"/auth/callback?code=idp-authorization-code&state={txn_id}"
+                )
+
+        assert response.status_code == 302
+        location = response.headers["location"]
+        query = urlparse(location).query
+        params = parse_qs(query)
+
+        assert "code" in params
+        assert params["state"] == ["client-state-success"]
+        # Exactly one `iss`, corrected to the canonical value -- a
+        # duplicate would make this list have length 2.
+        assert params["iss"] == ["https://myserver.com/"]
+        # Other query bytes from the registered redirect_uri survive
+        # byte-for-byte.
+        assert "flag" in query
+        assert "sig=%FF%FE" in query
diff --git a/tests/server/auth/oauth_proxy/test_tokens.py b/tests/server/auth/oauth_proxy/test_tokens.py
index 0ecb42afe..0e552d615 100644
--- a/tests/server/auth/oauth_proxy/test_tokens.py
+++ b/tests/server/auth/oauth_proxy/test_tokens.py
@@ -3,6 +3,7 @@
 import logging
 import time
 from unittest.mock import AsyncMock, Mock, patch
+from urllib.parse import parse_qs, urlparse
 
 import pytest
 from key_value.aio.stores.memory import MemoryStore
@@ -249,6 +250,60 @@ class TestOAuthProxyTokenEndpointAuth:
         mock_client.fetch_token.assert_awaited_once()
         mock_client.aclose.assert_awaited_once()
 
+    async def test_callback_redirect_includes_proxy_issuer(self, jwt_verifier):
+        proxy = OAuthProxy(
+            upstream_authorization_endpoint="https://oauth.example.com/authorize",
+            upstream_token_endpoint="https://oauth.example.com/token",
+            upstream_client_id="client-id",
+            upstream_client_secret="client-secret",
+            token_verifier=jwt_verifier,
+            base_url="https://proxy.example.com",
+            require_authorization_consent=False,
+            jwt_signing_key="test-secret",
+            client_storage=MemoryStore(),
+        )
+
+        await proxy._transaction_store.put(
+            key="txn-id",
+            value=OAuthTransaction(
+                txn_id="txn-id",
+                client_id="test-client",
+                client_redirect_uri="http://localhost:12345/callback",
+                client_state="client-state",
+                code_challenge="",
+                code_challenge_method="S256",
+                scopes=["read"],
+                created_at=time.time(),
+            ),
+        )
+
+        mock_request = Mock()
+        mock_request.query_params = {"code": "idp-code", "state": "txn-id"}
+        mock_request.cookies = {}
+
+        mock_client = AsyncMock()
+        mock_client.fetch_token = AsyncMock(
+            return_value={
+                "access_token": "upstream-access-token",
+                "refresh_token": "upstream-refresh-token",
+                "expires_in": 3600,
+                "token_type": "Bearer",
+            }
+        )
+
+        with patch.object(
+            proxy, "_create_upstream_oauth_client", return_value=mock_client
+        ):
+            response = await proxy._handle_idp_callback(mock_request)
+
+        assert response.status_code == 302
+        location = response.headers["location"]
+        query_params = parse_qs(urlparse(location).query)
+        assert "code" in query_params
+        assert query_params["state"] == ["client-state"]
+        assert query_params["iss"] == ["https://proxy.example.com/"]
+        mock_client.aclose.assert_awaited_once()
+
     async def test_callback_rejects_unsafe_transaction_redirect(self, jwt_verifier):
         proxy = OAuthProxy(
             upstream_authorization_endpoint="https://oauth.example.com/authorize",
diff --git a/tests/server/auth/test_enhanced_error_responses.py b/tests/server/auth/test_enhanced_error_responses.py
index e75a3dde0..0ff82110b 100644
--- a/tests/server/auth/test_enhanced_error_responses.py
+++ b/tests/server/auth/test_enhanced_error_responses.py
@@ -7,7 +7,11 @@ This test suite covers:
 4. Server branding in error pages
 """
 
+import asyncio
+from urllib.parse import parse_qs, quote, urlparse
+
 import pytest
+from key_value.aio.stores.memory import MemoryStore
 from mcp.shared.auth import OAuthClientInformationFull
 from pydantic import AnyHttpUrl, AnyUrl
 from starlette.applications import Starlette
@@ -18,6 +22,7 @@ from fastmcp.server.auth import RemoteAuthProvider, TokenVerifier
 from fastmcp.server.auth.auth import AccessToken
 from fastmcp.server.auth.oauth_proxy import OAuthProxy
 from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair
+from fastmcp.server.http import create_streamable_http_app
 
 
 class _UnderScopedTokenVerifier(TokenVerifier):
@@ -33,6 +38,57 @@ class _UnderScopedOAuthProxy(OAuthProxy):
         return AccessToken(token=token, client_id="test-client", scopes=["other"])
 
 
+class _DirectClientRedirectOAuthProxy(OAuthProxy):
+    """Proxy whose `authorize()` bypasses consent/upstream entirely and
+    redirects straight back to the client with a `code` — the pattern used
+    by providers (or tests) that short-circuit the standard
+    consent -> upstream IdP -> callback flow. OAuthProxy's own `authorize()`
+    never does this itself, but a subclass legitimately can, and
+    `AuthorizationHandler.handle()` must still attach `iss` to whatever
+    redirect comes back.
+
+    Appends its own `code`/`state` with `&` rather than an unconditional
+    `?` so this still produces a well-formed URL when `redirect_uri` is a
+    registered redirect that already carries its own query string (e.g. a
+    client-supplied `iss`)."""
+
+    async def authorize(self, client, params):  # type: ignore[override]
+        separator = "&" if "?" in str(params.redirect_uri) else "?"
+        return (
+            f"{params.redirect_uri}{separator}code=test-auth-code&state={params.state}"
+        )
+
+
+class _DirectClientRedirectWithIssOAuthProxy(OAuthProxy):
+    """Like `_DirectClientRedirectOAuthProxy`, but the provider's
+    `authorize()` override already put its own `iss` on the redirect —
+    simulating a provider that is itself RFC 9207-aware (or, when
+    `redirect_iss` doesn't match this server's issuer, a provider bug).
+    `response_kind` selects whether the redirect looks like a success
+    (`code`) or error (`error`) response; `AuthorizationHandler.handle()`
+    must not duplicate `iss` on either."""
+
+    def __init__(
+        self,
+        *args,
+        redirect_iss: str,
+        response_kind: str = "code",
+        **kwargs,
+    ):
+        super().__init__(*args, **kwargs)
+        self._redirect_iss = redirect_iss
+        self._response_kind = response_kind
+
+    async def authorize(self, client, params):  # type: ignore[override]
+        payload = (
+            f"code=test-auth-code&state={params.state}"
+            if self._response_kind == "code"
+            else f"error=access_denied&state={params.state}"
+        )
+        iss = quote(self._redirect_iss, safe="")
+        return f"{params.redirect_uri}?{payload}&iss={iss}"
+
+
 class TestEnhancedAuthorizationHandler:
     """Tests for enhanced authorization handler error responses."""
 
@@ -44,8 +100,6 @@ class TestEnhancedAuthorizationHandler:
     @pytest.fixture
     def oauth_proxy(self, rsa_key_pair):
         """Create OAuth proxy for testing."""
-        from key_value.aio.stores.memory import MemoryStore
-
         return OAuthProxy(
             upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
             upstream_token_endpoint="https://github.com/login/oauth/access_token",
@@ -143,8 +197,6 @@ class TestEnhancedAuthorizationHandler:
         )
 
         # Need to register synchronously
-        import asyncio
-
         asyncio.run(oauth_proxy.register_client(client_info))
 
         with TestClient(app) as client:
@@ -165,6 +217,388 @@ class TestEnhancedAuthorizationHandler:
             assert response.status_code == 302
             assert "/consent" in response.headers["location"]
 
+    def test_redirect_error_includes_proxy_issuer(self, oauth_proxy):
+        """Authorization error redirects should include RFC 9207 issuer."""
+        app = Starlette(routes=oauth_proxy.get_routes())
+
+        client_info = OAuthClientInformationFull(
+            client_id="valid-client",
+            client_secret="valid-secret",
+            redirect_uris=[AnyUrl("http://localhost:12345/callback")],
+            scope="read",
+        )
+
+        asyncio.run(oauth_proxy.register_client(client_info))
+
+        with TestClient(app) as client:
+            response = client.get(
+                "/authorize",
+                params={
+                    "client_id": "valid-client",
+                    "redirect_uri": "http://localhost:12345/callback",
+                    "response_type": "code",
+                    "code_challenge": "test-challenge",
+                    "state": "test-state",
+                    "scope": "admin",
+                },
+                headers={"Accept": "text/html"},
+                follow_redirects=False,
+            )
+
+            assert response.status_code == 302
+            query_params = parse_qs(urlparse(response.headers["location"]).query)
+            assert query_params["error"] == ["invalid_scope"]
+            assert query_params["state"] == ["test-state"]
+            assert query_params["iss"] == ["https://myserver.com/"]
+
+    def test_redirect_error_matches_path_base_url_metadata_issuer(self, rsa_key_pair):
+        """Authorization error redirects should match the metadata issuer exactly."""
+        oauth_proxy = OAuthProxy(
+            upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
+            upstream_token_endpoint="https://github.com/login/oauth/access_token",
+            upstream_client_id="test-client-id",
+            upstream_client_secret="test-client-secret",
+            token_verifier=JWTVerifier(
+                public_key=rsa_key_pair.public_key,
+                issuer="https://test.com",
+                audience="https://test.com",
+                base_url="https://test.com",
+            ),
+            base_url="https://proxy.example.com/oauth",
+            jwt_signing_key="test-secret",
+            client_storage=MemoryStore(),
+        )
+        app = Starlette(routes=oauth_proxy.get_routes())
+
+        client_info = OAuthClientInformationFull(
+            client_id="valid-client",
+            client_secret="valid-secret",
+            redirect_uris=[AnyUrl("http://localhost:12345/callback")],
+            scope="read",
+        )
+
+        asyncio.run(oauth_proxy.register_client(client_info))
+
+        with TestClient(app) as client:
+            metadata_response = client.get("/.well-known/oauth-authorization-server")
+            metadata = metadata_response.json()
+
+            response = client.get(
+                "/authorize",
+                params={
+                    "client_id": "valid-client",
+                    "redirect_uri": "http://localhost:12345/callback",
+                    "response_type": "code",
+                    "code_challenge": "test-challenge",
+                    "state": "test-state",
+                    "scope": "admin",
+                },
+                headers={"Accept": "text/html"},
+                follow_redirects=False,
+            )
+
+            assert metadata["issuer"] == "https://proxy.example.com/oauth"
+            assert response.status_code == 302
+            query_params = parse_qs(urlparse(response.headers["location"]).query)
+            assert query_params["error"] == ["invalid_scope"]
+            assert query_params["state"] == ["test-state"]
+            assert query_params["iss"] == [metadata["issuer"]]
+
+    def test_success_redirect_from_authorize_override_includes_issuer(
+        self, rsa_key_pair
+    ):
+        """RFC 9207 regression: a `code` redirect returned directly by
+        `authorize()` (bypassing consent/upstream) must carry `iss` too, not
+        just `error` redirects.
+
+        `AuthorizationHandler.handle()` previously only attached `iss` when
+        the SDK's redirect contained an `error` parameter. The base
+        `OAuthProxy.authorize()` never redirects straight to the client, so
+        this gap was invisible until a provider override (or a test mock,
+        like the GitHub provider integration test) returned the client
+        redirect directly — at which point the server was advertising
+        `authorization_response_iss_parameter_supported: true` while
+        silently breaking RFC 9207-aware clients on this path.
+        """
+        oauth_proxy = _DirectClientRedirectOAuthProxy(
+            upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
+            upstream_token_endpoint="https://github.com/login/oauth/access_token",
+            upstream_client_id="test-client-id",
+            upstream_client_secret="test-client-secret",
+            token_verifier=JWTVerifier(
+                public_key=rsa_key_pair.public_key,
+                issuer="https://test.com",
+                audience="https://test.com",
+                base_url="https://test.com",
+            ),
+            base_url="https://myserver.com",
+            jwt_signing_key="test-secret",
+            client_storage=MemoryStore(),
+        )
+        app = Starlette(routes=oauth_proxy.get_routes())
+
+        client_info = OAuthClientInformationFull(
+            client_id="valid-client",
+            client_secret="valid-secret",
+            redirect_uris=[AnyUrl("http://localhost:12345/callback")],
+            scope="read",
+        )
+        asyncio.run(oauth_proxy.register_client(client_info))
+
+        with TestClient(app) as client:
+            metadata = client.get("/.well-known/oauth-authorization-server").json()
+            assert metadata["authorization_response_iss_parameter_supported"] is True
+
+            response = client.get(
+                "/authorize",
+                params={
+                    "client_id": "valid-client",
+                    "redirect_uri": "http://localhost:12345/callback",
+                    "response_type": "code",
+                    "code_challenge": "test-challenge",
+                    "state": "test-state",
+                },
+                follow_redirects=False,
+            )
+
+        assert response.status_code == 302
+        query_params = parse_qs(urlparse(response.headers["location"]).query)
+        assert query_params["code"] == ["test-auth-code"]
+        assert query_params["state"] == ["test-state"]
+        assert query_params["iss"] == [metadata["issuer"]]
+
+    def test_success_redirect_does_not_duplicate_iss_already_in_redirect_uri(
+        self, rsa_key_pair
+    ):
+        """RFC 9207 P2 regression: a registered redirect_uri may already
+        carry its own `iss` query parameter — distinct from the provider
+        adding one itself (covered by
+        `test_success_redirect_with_matching_iss_not_duplicated` below).
+        `AuthorizationHandler.handle()` must still land on exactly one
+        `iss` (the canonical value), with every other query byte on the
+        registered URI preserved untouched.
+        """
+        oauth_proxy = _DirectClientRedirectOAuthProxy(
+            upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
+            upstream_token_endpoint="https://github.com/login/oauth/access_token",
+            upstream_client_id="test-client-id",
+            upstream_client_secret="test-client-secret",
+            token_verifier=JWTVerifier(
+                public_key=rsa_key_pair.public_key,
+                issuer="https://test.com",
+                audience="https://test.com",
+                base_url="https://test.com",
+            ),
+            base_url="https://myserver.com",
+            jwt_signing_key="test-secret",
+            client_storage=MemoryStore(),
+        )
+        app = Starlette(routes=oauth_proxy.get_routes())
+
+        client_redirect_uri = (
+            "http://localhost:12345/callback?iss=tenant&flag&sig=%FF%FE"
+        )
+        client_info = OAuthClientInformationFull(
+            client_id="valid-client",
+            client_secret="valid-secret",
+            redirect_uris=[AnyUrl(client_redirect_uri)],
+            scope="read",
+        )
+        asyncio.run(oauth_proxy.register_client(client_info))
+
+        with TestClient(app) as client:
+            metadata = client.get("/.well-known/oauth-authorization-server").json()
+
+            response = client.get(
+                "/authorize",
+                params={
+                    "client_id": "valid-client",
+                    "redirect_uri": client_redirect_uri,
+                    "response_type": "code",
+                    "code_challenge": "test-challenge",
+                    "state": "test-state",
+                },
+                follow_redirects=False,
+            )
+
+        assert response.status_code == 302
+        location = response.headers["location"]
+        query = urlparse(location).query
+        query_params = parse_qs(query)
+        assert query_params["code"] == ["test-auth-code"]
+        assert query_params["state"] == ["test-state"]
+        # Exactly one `iss`, corrected to the canonical value -- a
+        # duplicate would make this list have length 2.
+        assert query_params["iss"] == [metadata["issuer"]]
+        # Other query bytes from the registered redirect_uri survive
+        # byte-for-byte.
+        assert "flag" in query
+        assert "sig=%FF%FE" in query
+
+    def test_success_redirect_with_matching_iss_not_duplicated(self, rsa_key_pair):
+        """If a provider's `authorize()` override already stamped the
+        correct `iss` on its redirect, `handle()` must not append a second
+        one — RFC 6749 §3.1 forbids a response parameter appearing twice.
+        """
+        oauth_proxy = _DirectClientRedirectWithIssOAuthProxy(
+            upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
+            upstream_token_endpoint="https://github.com/login/oauth/access_token",
+            upstream_client_id="test-client-id",
+            upstream_client_secret="test-client-secret",
+            token_verifier=JWTVerifier(
+                public_key=rsa_key_pair.public_key,
+                issuer="https://test.com",
+                audience="https://test.com",
+                base_url="https://test.com",
+            ),
+            base_url="https://myserver.com",
+            jwt_signing_key="test-secret",
+            client_storage=MemoryStore(),
+            redirect_iss="https://myserver.com/",
+        )
+        app = Starlette(routes=oauth_proxy.get_routes())
+
+        client_info = OAuthClientInformationFull(
+            client_id="valid-client",
+            client_secret="valid-secret",
+            redirect_uris=[AnyUrl("http://localhost:12345/callback")],
+            scope="read",
+        )
+        asyncio.run(oauth_proxy.register_client(client_info))
+
+        with TestClient(app) as client:
+            metadata = client.get("/.well-known/oauth-authorization-server").json()
+
+            response = client.get(
+                "/authorize",
+                params={
+                    "client_id": "valid-client",
+                    "redirect_uri": "http://localhost:12345/callback",
+                    "response_type": "code",
+                    "code_challenge": "test-challenge",
+                    "state": "test-state",
+                },
+                follow_redirects=False,
+            )
+
+        assert response.status_code == 302
+        query_params = parse_qs(urlparse(response.headers["location"]).query)
+        # Exactly one `iss` (a duplicate would make this list have length 2).
+        assert query_params["iss"] == [metadata["issuer"]]
+
+    def test_success_redirect_with_mismatched_iss_is_corrected(self, rsa_key_pair):
+        """A provider's `authorize()` override can put an `iss` on its
+        redirect that doesn't match what this server advertises in its own
+        discovery document (`self._issuer`). An RFC 9207 client validates
+        `iss` against that document, so the mismatched value is already
+        unusable to a spec-compliant client. `handle()` corrects it to the
+        canonical value rather than leaving the broken value in place or
+        appending a second `iss` (which RFC 6749 §3.1 forbids outright).
+
+        This is a deliberate policy choice, not the only defensible one —
+        see the comment in `AuthorizationHandler.handle()` for the
+        reasoning, and update this test if that policy changes.
+        """
+        oauth_proxy = _DirectClientRedirectWithIssOAuthProxy(
+            upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
+            upstream_token_endpoint="https://github.com/login/oauth/access_token",
+            upstream_client_id="test-client-id",
+            upstream_client_secret="test-client-secret",
+            token_verifier=JWTVerifier(
+                public_key=rsa_key_pair.public_key,
+                issuer="https://test.com",
+                audience="https://test.com",
+                base_url="https://test.com",
+            ),
+            base_url="https://myserver.com",
+            jwt_signing_key="test-secret",
+            client_storage=MemoryStore(),
+            redirect_iss="https://wrong-issuer.example.com/",
+        )
+        app = Starlette(routes=oauth_proxy.get_routes())
+
+        client_info = OAuthClientInformationFull(
+            client_id="valid-client",
+            client_secret="valid-secret",
+            redirect_uris=[AnyUrl("http://localhost:12345/callback")],
+            scope="read",
+        )
+        asyncio.run(oauth_proxy.register_client(client_info))
+
+        with TestClient(app) as client:
+            metadata = client.get("/.well-known/oauth-authorization-server").json()
+
+            response = client.get(
+                "/authorize",
+                params={
+                    "client_id": "valid-client",
+                    "redirect_uri": "http://localhost:12345/callback",
+                    "response_type": "code",
+                    "code_challenge": "test-challenge",
+                    "state": "test-state",
+                },
+                follow_redirects=False,
+            )
+
+        assert response.status_code == 302
+        query_params = parse_qs(urlparse(response.headers["location"]).query)
+        # Exactly one `iss`, corrected to the canonical value rather than
+        # left mismatched or duplicated.
+        assert query_params["iss"] == [metadata["issuer"]]
+        assert query_params["iss"] != ["https://wrong-issuer.example.com/"]
+
+    def test_error_redirect_with_existing_iss_not_duplicated(self, rsa_key_pair):
+        """The duplication guard applies to error redirects too, not just
+        success ones — a provider override can construct an `error`
+        redirect that already carries `iss`.
+        """
+        oauth_proxy = _DirectClientRedirectWithIssOAuthProxy(
+            upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
+            upstream_token_endpoint="https://github.com/login/oauth/access_token",
+            upstream_client_id="test-client-id",
+            upstream_client_secret="test-client-secret",
+            token_verifier=JWTVerifier(
+                public_key=rsa_key_pair.public_key,
+                issuer="https://test.com",
+                audience="https://test.com",
+                base_url="https://test.com",
+            ),
+            base_url="https://myserver.com",
+            jwt_signing_key="test-secret",
+            client_storage=MemoryStore(),
+            redirect_iss="https://myserver.com/",
+            response_kind="error",
+        )
+        app = Starlette(routes=oauth_proxy.get_routes())
+
+        client_info = OAuthClientInformationFull(
+            client_id="valid-client",
+            client_secret="valid-secret",
+            redirect_uris=[AnyUrl("http://localhost:12345/callback")],
+            scope="read",
+        )
+        asyncio.run(oauth_proxy.register_client(client_info))
+
+        with TestClient(app) as client:
+            metadata = client.get("/.well-known/oauth-authorization-server").json()
+
+            response = client.get(
+                "/authorize",
+                params={
+                    "client_id": "valid-client",
+                    "redirect_uri": "http://localhost:12345/callback",
+                    "response_type": "code",
+                    "code_challenge": "test-challenge",
+                    "state": "test-state",
+                },
+                follow_redirects=False,
+            )
+
+        assert response.status_code == 302
+        query_params = parse_qs(urlparse(response.headers["location"]).query)
+        assert query_params["error"] == ["access_denied"]
+        assert query_params["iss"] == [metadata["issuer"]]
+
     def test_html_error_includes_server_branding(self, oauth_proxy):
         """Test that HTML error page includes server branding from FastMCP instance."""
         from mcp_types import Icon
@@ -251,8 +685,6 @@ class TestEnhancedRequireAuthMiddleware:
 
     def test_missing_auth_no_error_attribute(self, jwt_verifier):
         """Test that missing auth returns 401 without error attribute (RFC 6750 §3.1)."""
-        from fastmcp.server.http import create_streamable_http_app
-
         server = FastMCP("Test Server")
 
         @server.tool
@@ -381,8 +813,6 @@ class TestEnhancedRequireAuthMiddleware:
 
     def test_invalid_token_enhanced_error_message(self, jwt_verifier):
         """Test that invalid_token errors have enhanced error messages."""
-        from fastmcp.server.http import create_streamable_http_app
-
         server = FastMCP("Test Server")
 
         @server.tool
@@ -413,8 +843,6 @@ class TestEnhancedRequireAuthMiddleware:
 
     def test_invalid_token_www_authenticate_header_format(self, jwt_verifier):
         """Test that invalid token WWW-Authenticate header includes error attribute."""
-        from fastmcp.server.http import create_streamable_http_app
-
         server = FastMCP("Test Server")
         app = create_streamable_http_app(
             server=server,
@@ -439,8 +867,6 @@ class TestEnhancedRequireAuthMiddleware:
     def test_insufficient_scope_not_enhanced(self, rsa_key_pair):
         """Test that insufficient_scope errors are not modified."""
         # Create a valid token with wrong scopes
-        from fastmcp.server.http import create_streamable_http_app
-
         jwt_verifier = JWTVerifier(
             public_key=rsa_key_pair.public_key,
             issuer="https://test.com",
@@ -475,8 +901,6 @@ class TestContentNegotiation:
     @pytest.fixture
     def oauth_proxy(self):
         """Create OAuth proxy for testing."""
-        from key_value.aio.stores.memory import MemoryStore
-
         return OAuthProxy(
             upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
             upstream_token_endpoint="https://github.com/login/oauth/access_token",
diff --git a/tests/server/auth/test_oauth_consent_flow.py b/tests/server/auth/test_oauth_consent_flow.py
index ed756c081..fae84068f 100644
--- a/tests/server/auth/test_oauth_consent_flow.py
+++ b/tests/server/auth/test_oauth_consent_flow.py
@@ -111,6 +111,26 @@ def oauth_proxy_https_remember():
     )
 
 
+@pytest.fixture
+def oauth_proxy_https_path():
+    """OAuthProxy with a path component in base_url (no trailing slash).
+
+    Exercises the RFC 9207 issuer consistency across a base_url shape where
+    naive normalization (e.g. force-appending a trailing slash) would produce
+    an `iss` value that no longer matches the discovery document's `issuer`.
+    """
+    return OAuthProxy(
+        upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
+        upstream_token_endpoint="https://github.com/login/oauth/access_token",
+        upstream_client_id="client-id",
+        upstream_client_secret="client-secret",
+        token_verifier=_Verifier(),
+        base_url="https://myserver.example/oauth",
+        client_storage=MemoryStore(),
+        jwt_signing_key="test-secret",
+    )
+
+
 async def _start_flow(
     proxy: OAuthProxy, client_id: str, redirect: str
 ) -> tuple[str, str]:
@@ -631,11 +651,180 @@ class TestConsentSecurity:
             q = parse_qs(parsed.query)
             assert q.get("error") == ["access_denied"]
             assert q.get("state") == ["client-state-xyz"]
+            assert q.get("iss") == ["https://myserver.example/"]
             # Signed denied cookie should be set
             assert "MCP_DENIED_CLIENTS" in ";\n".join(
                 r.headers.get("set-cookie", "").splitlines()
             )
 
+    async def test_deny_redirect_does_not_duplicate_iss_already_in_redirect_uri(
+        self, oauth_proxy_https_remember
+    ):
+        """RFC 9207 P2 regression: a registered redirect_uri may already
+        carry its own `iss` query parameter. Explicit consent denial must
+        not append a second `iss` on top of it -- RFC 6749 §3.1 forbids a
+        response parameter appearing more than once -- and every other
+        query byte on the registered URI (a valueless `flag` and a
+        non-UTF-8 percent-encoded `sig`) must survive untouched.
+        """
+        client_redirect = "http://localhost:5009/callback?iss=tenant&flag&sig=%FF%FE"
+        txn_id, _ = await _start_flow(
+            oauth_proxy_https_remember, "client-dup-iss", client_redirect
+        )
+        app = Starlette(routes=oauth_proxy_https_remember.get_routes())
+        with TestClient(app) as c:
+            consent = c.get(f"/consent?txn_id={txn_id}")
+            csrf = _extract_csrf(consent.text)
+            assert csrf
+            for k, v in consent.cookies.items():
+                c.cookies.set(k, v)
+            r = c.post(
+                "/consent",
+                data={"action": "deny", "txn_id": txn_id, "csrf_token": csrf},
+                follow_redirects=False,
+            )
+            assert r.status_code in (302, 303)
+            loc = r.headers.get("location", "")
+            query = urlparse(loc).query
+            q = parse_qs(query)
+            assert q.get("error") == ["access_denied"]
+            # Exactly one `iss`, corrected to the canonical value -- a
+            # duplicate would make this list have length 2.
+            assert q.get("iss") == ["https://myserver.example/"]
+            # Other query bytes from the registered redirect_uri survive
+            # byte-for-byte.
+            assert "flag" in query
+            assert "sig=%FF%FE" in query
+
+    async def test_deny_redirect_issuer_matches_path_base_url_metadata(
+        self, oauth_proxy_https_path
+    ):
+        """Consent-denial `iss` must match the discovery document exactly.
+
+        Regression test for a base_url with a path and no trailing slash
+        (`https://myserver.example/oauth`): the metadata `issuer` is the
+        unmodified base_url, so the denial redirect's `iss` must match it
+        byte-for-byte rather than force-appending a trailing slash.
+        """
+        client_redirect = "http://localhost:5008/callback"
+        txn_id, _ = await _start_flow(
+            oauth_proxy_https_path, "client-path", client_redirect
+        )
+        app = Starlette(routes=oauth_proxy_https_path.get_routes())
+        with TestClient(app) as c:
+            metadata = c.get("/.well-known/oauth-authorization-server").json()
+            assert metadata["issuer"] == "https://myserver.example/oauth"
+
+            consent = c.get(f"/consent?txn_id={txn_id}")
+            csrf = _extract_csrf(consent.text)
+            assert csrf
+            for k, v in consent.cookies.items():
+                c.cookies.set(k, v)
+            r = c.post(
+                "/consent",
+                data={"action": "deny", "txn_id": txn_id, "csrf_token": csrf},
+                follow_redirects=False,
+            )
+            assert r.status_code in (302, 303)
+            q = parse_qs(urlparse(r.headers.get("location", "")).query)
+            assert q.get("error") == ["access_denied"]
+            assert q.get("iss") == [metadata["issuer"]]
+
+    async def test_remembered_denial_redirects_with_issuer(
+        self, oauth_proxy_https_remember
+    ):
+        """Remembered consent denial redirects with RFC 9207 issuer."""
+        client_id = "client-denied"
+        redirect = "http://localhost:5007/callback"
+        txn_id, _ = await _start_flow(oauth_proxy_https_remember, client_id, redirect)
+        app = Starlette(routes=oauth_proxy_https_remember.get_routes())
+        with TestClient(app) as c:
+            consent = c.get(f"/consent?txn_id={txn_id}")
+            csrf = _extract_csrf(consent.text)
+            assert csrf
+            for k, v in consent.cookies.items():
+                c.cookies.set(k, v)
+            r = c.post(
+                "/consent",
+                data={"action": "deny", "txn_id": txn_id, "csrf_token": csrf},
+                follow_redirects=False,
+            )
+            set_cookie = ";\n".join(r.headers.get("set-cookie", "").splitlines())
+            m = re.search(r"__Host-MCP_DENIED_CLIENTS=([^;]+)", set_cookie)
+            assert m
+            denied_cookie = m.group(1)
+
+            new_txn, _ = await _start_flow(
+                oauth_proxy_https_remember, client_id, redirect
+            )
+            c.cookies.set("__Host-MCP_DENIED_CLIENTS", denied_cookie)
+            r2 = c.get(
+                f"/consent?txn_id={new_txn}",
+                headers={"Sec-Fetch-Site": "none"},
+                follow_redirects=False,
+            )
+
+            assert r2.status_code in (302, 303)
+            loc = r2.headers.get("location", "")
+            parsed = urlparse(loc)
+            assert parsed.scheme == "http" and parsed.netloc.startswith("localhost")
+            q = parse_qs(parsed.query)
+            assert q.get("error") == ["access_denied"]
+            assert q.get("state") == ["client-state-xyz"]
+            assert q.get("iss") == ["https://myserver.example/"]
+
+    async def test_remembered_denial_does_not_duplicate_iss_already_in_redirect_uri(
+        self, oauth_proxy_https_remember
+    ):
+        """RFC 9207 P2 regression: the *remembered/silent* denial path is a
+        separate call site from the explicit deny above, and must
+        independently avoid duplicating `iss` when the registered
+        redirect_uri already carries one.
+        """
+        client_id = "client-denied-dup-iss"
+        redirect = "http://localhost:5010/callback?iss=tenant&flag&sig=%FF%FE"
+        txn_id, _ = await _start_flow(oauth_proxy_https_remember, client_id, redirect)
+        app = Starlette(routes=oauth_proxy_https_remember.get_routes())
+        with TestClient(app) as c:
+            consent = c.get(f"/consent?txn_id={txn_id}")
+            csrf = _extract_csrf(consent.text)
+            assert csrf
+            for k, v in consent.cookies.items():
+                c.cookies.set(k, v)
+            r = c.post(
+                "/consent",
+                data={"action": "deny", "txn_id": txn_id, "csrf_token": csrf},
+                follow_redirects=False,
+            )
+            set_cookie = ";\n".join(r.headers.get("set-cookie", "").splitlines())
+            m = re.search(r"__Host-MCP_DENIED_CLIENTS=([^;]+)", set_cookie)
+            assert m
+            denied_cookie = m.group(1)
+
+            new_txn, _ = await _start_flow(
+                oauth_proxy_https_remember, client_id, redirect
+            )
+            c.cookies.set("__Host-MCP_DENIED_CLIENTS", denied_cookie)
+            r2 = c.get(
+                f"/consent?txn_id={new_txn}",
+                headers={"Sec-Fetch-Site": "none"},
+                follow_redirects=False,
+            )
+
+            assert r2.status_code in (302, 303)
+            loc = r2.headers.get("location", "")
+            query = urlparse(loc).query
+            q = parse_qs(query)
+            assert q.get("error") == ["access_denied"]
+            assert q.get("state") == ["client-state-xyz"]
+            # Exactly one `iss`, corrected to the canonical value -- a
+            # duplicate would make this list have length 2.
+            assert q.get("iss") == ["https://myserver.example/"]
+            # Other query bytes from the registered redirect_uri survive
+            # byte-for-byte.
+            assert "flag" in query
+            assert "sig=%FF%FE" in query
+
     async def test_approve_sets_cookie_and_redirects_to_upstream(
         self, oauth_proxy_https_remember
     ):
diff --git a/tests/server/auth/test_redirect_validation.py b/tests/server/auth/test_redirect_validation.py
index 58bb29568..f34f6097e 100644
--- a/tests/server/auth/test_redirect_validation.py
+++ b/tests/server/auth/test_redirect_validation.py
@@ -1,15 +1,246 @@
 """Tests for redirect URI validation in OAuth flows."""
 
+import re
+from pathlib import Path
+from urllib.parse import parse_qs, parse_qsl, urlparse
+
 import pytest
 from pydantic import AnyUrl
 
+import fastmcp.server.auth
 from fastmcp.server.auth.redirect_validation import (
     DEFAULT_LOCALHOST_PATTERNS,
+    add_query_params,
+    build_client_redirect,
     matches_allowed_pattern,
+    replace_query_param,
     validate_redirect_uri,
 )
 
 
+class TestAddQueryParams:
+    """Test that add_query_params preserves the registered callback's exact query bytes.
+
+    A registered redirect URI may carry an opaque or signed query string.
+    Decoding it with parse_qsl and re-serializing with urlencode mutates it
+    (a valueless `?flag` becomes `?flag=`, and non-UTF-8 percent-encoded
+    bytes get replaced) which breaks clients that route on, or
+    cryptographically validate, the raw callback query.
+    """
+
+    def test_preserves_valueless_param_and_non_utf8_bytes(self):
+        original_query = "flag&sig=%FF%FE"
+        url = f"https://client.example.com/callback?{original_query}"
+
+        result = add_query_params(
+            url,
+            {
+                "code": "abc123",
+                "state": "xyz state",
+                "iss": "https://issuer.example.com/",
+            },
+        )
+
+        result_query = urlparse(result).query
+
+        # The original query substring must survive byte-for-byte: the
+        # valueless `flag` must not become `flag=`, and the non-UTF-8
+        # percent-encoded `sig` value must not be decoded/replaced.
+        assert result_query.startswith(f"{original_query}&")
+
+        # New params are appended after a single `&`, correctly encoded.
+        appended = result_query[len(original_query) + 1 :]
+        assert dict(parse_qsl(appended)) == {
+            "code": "abc123",
+            "state": "xyz state",
+            "iss": "https://issuer.example.com/",
+        }
+
+    def test_empty_query_has_no_stray_ampersand(self):
+        url = "https://client.example.com/callback"
+
+        result = add_query_params(url, {"code": "abc123"})
+
+        assert result == "https://client.example.com/callback?code=abc123"
+
+    def test_appends_to_existing_ordinary_query(self):
+        url = "https://client.example.com/callback?foo=bar"
+
+        result = add_query_params(url, {"code": "abc123"})
+
+        assert result == "https://client.example.com/callback?foo=bar&code=abc123"
+
+
+class TestReplaceQueryParam:
+    """Direct tests for the idempotent replace-or-append primitive that
+    `build_client_redirect` relies on to guarantee exactly one `iss`.
+    """
+
+    def test_replaces_existing_value_in_place_preserving_other_bytes(self):
+        url = "https://client.example.com/callback?iss=tenant&sig=%FF%FE"
+
+        result = replace_query_param(url, "iss", "https://issuer.example.com/")
+
+        assert urlparse(result).query == (
+            "iss=https%3A%2F%2Fissuer.example.com%2F&sig=%FF%FE"
+        )
+
+    def test_appends_when_key_absent(self):
+        url = "https://client.example.com/callback?sig=%FF%FE"
+
+        result = replace_query_param(url, "iss", "https://issuer.example.com/")
+
+        assert urlparse(result).query == (
+            "sig=%FF%FE&iss=https%3A%2F%2Fissuer.example.com%2F"
+        )
+
+    def test_only_first_occurrence_is_replaced(self):
+        """A key appearing twice in the input is left with one replaced
+        occurrence and one untouched -- callers must not feed this function
+        an already-duplicated key and expect deduplication."""
+        url = "https://client.example.com/callback?iss=first&iss=second"
+
+        result = replace_query_param(url, "iss", "https://issuer.example.com/")
+
+        assert urlparse(result).query == (
+            "iss=https%3A%2F%2Fissuer.example.com%2F&iss=second"
+        )
+
+
+class TestBuildClientRedirect:
+    """Tests for the single helper that owns the client-facing-redirect
+    `iss` invariant: exactly one `iss`, set to the canonical value, with
+    every other query byte preserved verbatim.
+
+    This is the consolidation point for RFC 9207 support -- every redirect
+    the OAuth proxy sends back to a client (success or error, across all
+    five call sites that build one) must go through this function rather
+    than hand-building a params dict with its own `"iss"` key.
+    """
+
+    def test_appends_params_and_iss_when_absent(self):
+        url = "https://client.example.com/callback"
+
+        result = build_client_redirect(
+            url,
+            {"code": "abc", "state": "xyz"},
+            iss="https://issuer.example.com/",
+        )
+
+        assert dict(parse_qsl(urlparse(result).query)) == {
+            "code": "abc",
+            "state": "xyz",
+            "iss": "https://issuer.example.com/",
+        }
+
+    def test_replaces_iss_already_present_in_registered_redirect_uri(self):
+        """A registered redirect_uri may legitimately carry its own `iss`
+        query parameter (e.g. a multi-tenant client encoding its tenant in
+        the callback URL). Blindly appending the server's issuer on top of
+        that would yield two `iss` values -- RFC 6749 §3.1 forbids a
+        response parameter appearing more than once, so strict clients
+        reject the response or read the wrong value. This is the P2 defect
+        this helper exists to close off at every call site, not just one.
+        """
+        url = "https://client.example.com/callback?iss=tenant&sig=%FF%FE"
+
+        result = build_client_redirect(
+            url,
+            {"code": "abc", "state": "xyz"},
+            iss="https://issuer.example.com/",
+        )
+
+        result_query = urlparse(result).query
+        iss_values = parse_qs(result_query)["iss"]
+        assert len(iss_values) == 1
+        assert iss_values == ["https://issuer.example.com/"]
+
+    def test_preserves_valueless_param_and_non_utf8_bytes_alongside_existing_iss(
+        self,
+    ):
+        """Exact end-to-end reproduction of the worked example from the P2
+        review comment: registered redirect_uri already has `iss`, a
+        valueless `flag`, and a non-UTF-8 percent-encoded `sig` -- all three
+        must survive the round trip through `add_query_params` +
+        `replace_query_param` untouched, with only `iss` rewritten in
+        place.
+        """
+        url = "https://client.example.com/callback?iss=tenant&flag&sig=%FF%FE"
+
+        result = build_client_redirect(
+            url,
+            {"code": "abc", "state": "xyz state"},
+            iss="https://issuer.example.com/",
+        )
+
+        assert urlparse(result).query == (
+            "iss=https%3A%2F%2Fissuer.example.com%2F"
+            "&flag&sig=%FF%FE&code=abc&state=xyz+state"
+        )
+
+    def test_rejects_iss_hand_specified_in_params(self):
+        """`iss` must come from the keyword-only `iss` argument, never from
+        the `params` dict -- this keeps exactly one place a caller can set
+        it, rather than two that could disagree."""
+        with pytest.raises(ValueError, match="iss"):
+            build_client_redirect(
+                "https://client.example.com/callback",
+                {"code": "abc", "iss": "sneaky"},
+                iss="https://issuer.example.com/",
+            )
+
+    def test_empty_params_does_not_add_stray_ampersand(self):
+        """The authorize-handler call site passes no extra params (it only
+        needs to fix up `iss` on a URL the SDK already built) -- an empty
+        `params` dict must not introduce a trailing/stray `&`."""
+        url = "https://client.example.com/callback?code=abc&state=xyz"
+
+        result = build_client_redirect(url, {}, iss="https://issuer.example.com/")
+
+        assert urlparse(result).query == (
+            "code=abc&state=xyz&iss=https%3A%2F%2Fissuer.example.com%2F"
+        )
+        assert "&&" not in result
+        assert not result.endswith("&")
+
+
+class TestNoHandSpecifiedIssOutsideHelper:
+    """Guard against a future call site reintroducing the duplicate-`iss`
+    bug this PR consolidates away.
+
+    This is the sixth review round on the RFC 9207 `iss` work, and the last
+    two rounds were the same defect surfacing at different call sites: a
+    caller hand-building a params dict with its own `"iss"` key instead of
+    routing through `build_client_redirect`. Rather than trust that every
+    future redirect site remembers to do this, scan the directories that
+    build client-facing authorization redirects (`oauth_proxy/`,
+    `handlers/`) for a dict-literal `"iss"` key. `jwt_issuer.py` and the
+    JWT/Clerk providers legitimately use `"iss"` as a JWT claim name, but
+    those live outside these two directories, so this scan does not need to
+    special-case them.
+    """
+
+    def test_no_dict_literal_iss_key_in_redirect_building_modules(self):
+        auth_root = Path(fastmcp.server.auth.__file__).parent
+        scan_dirs = [auth_root / "oauth_proxy", auth_root / "handlers"]
+        iss_dict_key = re.compile(r"""["']iss["']\s*:""")
+
+        offenders = [
+            str(path)
+            for scan_dir in scan_dirs
+            for path in scan_dir.rglob("*.py")
+            if iss_dict_key.search(path.read_text())
+        ]
+
+        assert not offenders, (
+            "Found a hand-specified 'iss' dict key outside "
+            "build_client_redirect() in: "
+            f"{offenders}. Route this redirect through "
+            "fastmcp.server.auth.redirect_validation.build_client_redirect "
+            "instead so the duplicate-iss invariant stays centralized."
+        )
+
+
 class TestMatchesAllowedPattern:
     """Test wildcard pattern matching for redirect URIs."""