From b452dd73d24806070d97cbbae3db97a7baa95f2f Mon Sep 17 00:00:00 2001
From: Giles Odigwe <gilesodigwe@microsoft.com>
Date: Thu, 13 Aug 2026 14:26:22 -0700
Subject: [PATCH 1/5] Python: surface mid-run oauth_consent_request items from
 ResponsesHostServer

`_to_outputs` had no branch for `oauth_consent_request` content, so a consent
link produced after the agent was entered (for example by an on-behalf-of MCP
server that needs a per-user token at tool-invocation time) fell into the
catch-all and was dropped with "Content type 'oauth_consent_request' is not
supported yet". The client saw a completed response with no consent prompt.

Only connect-time consent failures raised by `_ensure_agent_ready` were
surfaced as `oauth_consent_request` output items, and the inbound conversion
(`_output_item_to_message`) already handled the item type, so the outbound
direction was the missing half.

- Emit `oauth_consent_request` added/done output items from `_to_outputs`,
  validating the link is an absolute HTTPS URL and reading `server_label` from
  the content's additional properties.
- End the response as `incomplete` (instead of `completed`) when a consent
  request was emitted mid-run, in both the agent and workflow handlers,
  matching the connect-time path.
- Factor the item emission, link validation, and incomplete reason into shared
  helpers reused by the connect-time path.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 992057d8-78a8-45a7-9201-35af0919b071
---
 .../_responses.py                             |  99 +++++++++++++---
 .../foundry_hosting/tests/test_responses.py   | 110 ++++++++++++++++++
 2 files changed, 195 insertions(+), 14 deletions(-)

diff --git a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
index 6ff4550f7e0..8206c34e6bc 100644
--- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
+++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
@@ -10,6 +10,7 @@
 from contextlib import AbstractAsyncContextManager, AsyncExitStack
 from dataclasses import asdict, dataclass, is_dataclass
 from typing import Literal, cast
+from urllib.parse import urlparse
 
 from agent_framework import (
     ChatOptions,
@@ -164,6 +165,55 @@ def consent_url_from_error(exc: BaseException) -> list[ConsentError] | None:
     return None
 
 
+def _consent_link_from_content(content: Content) -> str | None:
+    """Return a validated consent link for an ``oauth_consent_request`` content.
+
+    Returns ``None`` when *content* is not an OAuth consent request, when it carries
+    no consent link, or when the link is not an absolute HTTPS URL.
+    """
+    if content.type != "oauth_consent_request":
+        return None
+    consent_link = content.consent_link
+    if not consent_link:
+        logger.warning("Received oauth_consent_request content without a consent_link; skipping.")
+        return None
+    parsed = urlparse(consent_link)
+    if parsed.scheme.lower() != "https" or not parsed.netloc:
+        logger.warning("Skipping oauth_consent_request with non-HTTPS consent_link.")
+        return None
+    return consent_link
+
+
+def _emit_oauth_consent_item(
+    stream: ResponseEventStream,
+    response_id: str,
+    consent_link: str,
+    server_label: str,
+) -> Generator[ResponseStreamEvent]:
+    """Yield the added/done events for an ``oauth_consent_request`` output item."""
+    oauth_item = OAuthConsentRequestOutputItem(
+        id=IdGenerator.new_id("oacr"),
+        response_id=response_id,
+        type="oauth_consent_request",
+        consent_link=consent_link,
+        server_label=server_label,
+    )
+    builder = stream.add_output_item(oauth_item["id"])
+    yield builder.emit_added(oauth_item)
+    yield builder.emit_done(oauth_item)
+
+
+def _consent_incomplete_reason(count: int) -> str:
+    """Return the ``response.incomplete`` reason for *count* pending consent requests."""
+    return f"OAuth consent required for {count} tool(s)."
+
+
+def _consent_server_label(content: Content) -> str:
+    """Return the server label to report for an ``oauth_consent_request`` content."""
+    label = content.additional_properties.get("server_label") if content.additional_properties else None
+    return label if isinstance(label, str) and label else "agent_framework"
+
+
 # endregion Foundry Toolbox Auth integration
 
 
@@ -353,20 +403,15 @@ async def _handle_inner_agent(
 
             for consent_error in consent_errors_to_emit:
                 logger.warning("Consent URL for tool '%s': %s", consent_error.name, consent_error.consent_url)
-                oauth_item = OAuthConsentRequestOutputItem(
-                    id=IdGenerator.new_id("oacr"),
-                    response_id=context.response_id,
-                    type="oauth_consent_request",
-                    consent_link=consent_error.consent_url,
-                    server_label=consent_error.name,
-                )
-                builder = response_event_stream.add_output_item(oauth_item["id"])
-                yield builder.emit_added(oauth_item)
-                yield builder.emit_done(oauth_item)
+                for event in _emit_oauth_consent_item(
+                    response_event_stream,
+                    context.response_id,
+                    consent_error.consent_url,
+                    consent_error.name,
+                ):
+                    yield event
 
-            yield response_event_stream.emit_incomplete(
-                reason=f"OAuth consent required for {len(consent_errors_to_emit)} tool(s)."
-            )
+            yield response_event_stream.emit_incomplete(reason=_consent_incomplete_reason(len(consent_errors_to_emit)))
             return
 
         try:
@@ -406,6 +451,7 @@ async def _handle_inner_agent(
         request_failure: Exception | None = None
         save_failure: Exception | None = None
         request_interrupted = False
+        pending_consent_count = 0
 
         try:
             if self._uses_hosted_responses_history:
@@ -431,6 +477,8 @@ async def _handle_inner_agent(
 
             async for update in self._agent.run(stream=True, **run_kwargs):  # type: ignore[reportUnknownMemberType]
                 for content in update.contents:
+                    if _consent_link_from_content(content) is not None:
+                        pending_consent_count += 1
                     for event in tracker.handle(content):
                         yield event
                     if tracker.needs_async:
@@ -480,6 +528,10 @@ async def _handle_inner_agent(
         elif save_failure is not None:
             for event in self._emit_failure(response_event_stream, tracker, save_failure):
                 yield event
+        elif pending_consent_count > 0:
+            # The turn cannot finish until the user completes OAuth consent, so the response
+            # ends as `incomplete` rather than `completed`, matching the connect-time path.
+            yield response_event_stream.emit_incomplete(reason=_consent_incomplete_reason(pending_consent_count))
         else:
             yield response_event_stream.emit_completed()
 
@@ -580,6 +632,7 @@ async def _handle_inner_workflow(
                     pass
 
             tracker = _OutputItemTracker(response_event_stream)
+            pending_consent_count = 0
 
             # Run the workflow agent in streaming mode with the new user input.
             async for update in self._agent.run(
@@ -588,6 +641,8 @@ async def _handle_inner_workflow(
                 checkpoint_storage=write_storage,
             ):
                 for content in update.contents:
+                    if _consent_link_from_content(content) is not None:
+                        pending_consent_count += 1
                     for event in tracker.handle(content):
                         yield event
                     if tracker.needs_async:
@@ -602,7 +657,10 @@ async def _handle_inner_workflow(
                 yield event
 
             await self._delete_not_latest_checkpoints(write_storage, self._agent.workflow.name)
-            yield response_event_stream.emit_completed()
+            if pending_consent_count > 0:
+                yield response_event_stream.emit_incomplete(reason=_consent_incomplete_reason(pending_consent_count))
+            else:
+                yield response_event_stream.emit_completed()
         except Exception as ex:
             logger.exception("Failed to produce response for workflow agent")
             for event in self._emit_failure(response_event_stream, tracker, ex):
@@ -1735,6 +1793,19 @@ async def _to_outputs(
                 "Approval request was not saved to approval storage because the approval request ID "
                 "could not be extracted from the stream event."
             )
+    elif content.type == "oauth_consent_request":
+        # An OBO/on-behalf-of tool can require consent mid-run, after the agent has already
+        # been entered. Surface the link as an `oauth_consent_request` output item so the
+        # client can render a consent prompt instead of an empty assistant turn.
+        consent_link = _consent_link_from_content(content)
+        if consent_link is not None:
+            for event in _emit_oauth_consent_item(
+                stream,
+                str(stream.response["id"]),
+                consent_link,
+                _consent_server_label(content),
+            ):
+                yield event
     else:
         # Log a warning for unsupported content types instead of raising an error to avoid breaking the response stream.
         logger.warning(f"Content type '{content.type}' is not supported yet. This is usually safe to ignore.")
diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py
index 5b0dfda65c0..103f0a61e70 100644
--- a/python/packages/foundry_hosting/tests/test_responses.py
+++ b/python/packages/foundry_hosting/tests/test_responses.py
@@ -3833,6 +3833,116 @@ async def test_retry_after_consent_succeeds(self) -> None:
         agent.run.assert_called_once()
 
 
+class TestMidRunOAuthConsentSurfacing:
+    """A tool can require consent after the agent has been entered (e.g. an on-behalf-of
+    MCP server needing a per-user token), in which case the consent link arrives as
+    ``oauth_consent_request`` content in the agent's stream rather than as a connect-time error.
+    """
+
+    async def test_streaming_mid_run_consent_content_emits_oauth_output_item(self) -> None:
+        agent = _make_agent(
+            stream_updates=[
+                AgentResponseUpdate(contents=[Content.from_text("one moment")], role="assistant"),
+                AgentResponseUpdate(
+                    contents=[
+                        Content.from_oauth_consent_request(
+                            consent_link="https://consent.example.com/obo",
+                            additional_properties={"server_label": "obo-mcp"},
+                        )
+                    ],
+                    role="assistant",
+                ),
+            ]
+        )
+        server = _make_server(agent)
+
+        resp = await _post(server, input_text="hello", stream=True)
+        assert resp.status_code == 200
+        events = _parse_sse_events(resp.text)
+        types = _sse_event_types(events)
+
+        assert types[-1] == "response.incomplete"
+
+        added = [e for e in events if e["event"] == "response.output_item.added"]
+        oauth_added = [e for e in added if e["data"]["item"]["type"] == "oauth_consent_request"]
+        assert len(oauth_added) == 1
+        assert oauth_added[0]["data"]["item"]["consent_link"] == "https://consent.example.com/obo"
+        assert oauth_added[0]["data"]["item"]["server_label"] == "obo-mcp"
+
+        done = [e for e in events if e["event"] == "response.output_item.done"]
+        assert any(e["data"]["item"]["type"] == "oauth_consent_request" for e in done)
+
+    async def test_non_streaming_mid_run_consent_content_emits_oauth_output_item(self) -> None:
+        agent = _make_agent(
+            response=AgentResponse(
+                messages=[
+                    Message(
+                        role="assistant",
+                        contents=[Content.from_oauth_consent_request(consent_link="https://consent.example.com/obo")],
+                    )
+                ]
+            )
+        )
+        server = _make_server(agent)
+
+        resp = await _post(server, input_text="hello", stream=False)
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["status"] == "incomplete"
+
+        oauth_items = [it for it in body["output"] if it["type"] == "oauth_consent_request"]
+        assert len(oauth_items) == 1
+        assert oauth_items[0]["consent_link"] == "https://consent.example.com/obo"
+        assert oauth_items[0]["server_label"] == "agent_framework"
+
+    async def test_multiple_consent_contents_each_emit_an_item(self) -> None:
+        agent = _make_agent(
+            stream_updates=[
+                AgentResponseUpdate(
+                    contents=[
+                        Content.from_oauth_consent_request(consent_link="https://consent.example.com/one"),
+                        Content.from_oauth_consent_request(consent_link="https://consent.example.com/two"),
+                    ],
+                    role="assistant",
+                )
+            ]
+        )
+        server = _make_server(agent)
+
+        resp = await _post(server, input_text="hello", stream=True)
+        events = _parse_sse_events(resp.text)
+        added = [e for e in events if e["event"] == "response.output_item.added"]
+        oauth_added = [e for e in added if e["data"]["item"]["type"] == "oauth_consent_request"]
+        assert len(oauth_added) == 2
+        assert {e["data"]["item"]["id"] for e in oauth_added} != {""}
+        assert len({e["data"]["item"]["id"] for e in oauth_added}) == 2
+
+        incomplete = [e for e in events if e["event"] == "response.incomplete"]
+        assert len(incomplete) == 1
+
+    @pytest.mark.parametrize("consent_link", ["", "http://consent.example.com/obo", "not-a-url"])
+    async def test_invalid_consent_link_is_skipped(self, consent_link: str) -> None:
+        agent = _make_agent(
+            stream_updates=[
+                AgentResponseUpdate(
+                    contents=[Content(type="oauth_consent_request", consent_link=consent_link or None)],
+                    role="assistant",
+                ),
+                AgentResponseUpdate(contents=[Content.from_text("done")], role="assistant"),
+            ]
+        )
+        server = _make_server(agent)
+
+        resp = await _post(server, input_text="hello", stream=True)
+        events = _parse_sse_events(resp.text)
+        types = _sse_event_types(events)
+
+        added = [e for e in events if e["event"] == "response.output_item.added"]
+        assert not any(e["data"]["item"]["type"] == "oauth_consent_request" for e in added)
+        # A link we cannot render is not a consent prompt, so the turn still completes.
+        assert types[-1] == "response.completed"
+
+
 # endregion
 
 # region Error handling (response.failed surfacing)

From 9b33db988a055c06fff56ba83c93e973421b5e15 Mon Sep 17 00:00:00 2001
From: Giles Odigwe <gilesodigwe@microsoft.com>
Date: Fri, 14 Aug 2026 12:12:29 -0700
Subject: [PATCH 2/5] Python: address review feedback on mid-run OAuth consent
 surfacing

- Harden consent link validation. `urlparse` raises `ValueError` for malformed
  authorities such as `https://[broken`, which turned an unrenderable link into
  a failed response instead of skipping the item, and a non-empty `netloc` is
  not sufficient on its own (`https://@` has one but no host). Validation now
  catches the parse error and requires a hostname, in both the hosting layer
  and `agent_framework_foundry._oauth_helpers`, which had the same defect.

- Preserve the server label. `try_parse_oauth_consent_event` only kept the
  upstream item in `raw_representation`, so every real Foundry consent event
  was re-emitted with the fallback label. The parser now copies `server_label`
  into `additional_properties`, and hosting falls back to the raw item's label
  before defaulting.

- Validate connect-time consent links too. An entry-time consent error with no
  renderable link now produces `response.failed` rather than an `incomplete`
  carrying no link the user can act on, and the reported count reflects the
  items actually emitted.

- Suppress duplicate consent prompts. `WorkflowAgent` replays the inner agent's
  content as workflow output, so the same consent request reached the host
  twice and produced two prompts. `_to_outputs` now takes the set of emitted
  `(consent_link, server_label)` pairs and skips repeats.

- Cover the workflow hosting path, which was previously exercised only through
  the regular agent handler.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 992057d8-78a8-45a7-9201-35af0919b071
---
 .../agent_framework_foundry/_oauth_helpers.py |  27 +++-
 .../tests/foundry/test_oauth_helpers.py       |  62 ++++++++
 .../_responses.py                             | 113 ++++++++++----
 .../foundry_hosting/tests/test_responses.py   | 143 +++++++++++++++++-
 4 files changed, 302 insertions(+), 43 deletions(-)

diff --git a/python/packages/foundry/agent_framework_foundry/_oauth_helpers.py b/python/packages/foundry/agent_framework_foundry/_oauth_helpers.py
index 873d42c3d9c..c8d70fe97fb 100644
--- a/python/packages/foundry/agent_framework_foundry/_oauth_helpers.py
+++ b/python/packages/foundry/agent_framework_foundry/_oauth_helpers.py
@@ -12,12 +12,22 @@
 
 
 def _validate_consent_link(consent_link: str, item_id: str) -> str:
-    """Validate a consent link is HTTPS with a valid netloc.
+    """Validate a consent link is HTTPS with a valid host.
 
-    Returns the link unchanged if valid, or an empty string if not.
+    Returns the link unchanged if valid, or an empty string if not. ``urlparse`` raises
+    ``ValueError`` for malformed authorities (for example ``https://[broken``), and a
+    non-empty ``netloc`` is not sufficient on its own (``https://@`` has one but no host).
     """
-    parsed = urlparse(consent_link)
-    if parsed.scheme.lower() != "https" or not parsed.netloc:
+    try:
+        parsed = urlparse(consent_link)
+        hostname = parsed.hostname
+    except ValueError:
+        logger.warning(
+            "Skipping oauth_consent_request with malformed consent_link (item id=%s)",
+            item_id,
+        )
+        return ""
+    if parsed.scheme.lower() != "https" or not hostname:
         logger.warning(
             "Skipping oauth_consent_request with non-HTTPS consent_link (item id=%s)",
             item_id,
@@ -55,9 +65,18 @@ def try_parse_oauth_consent_event(event: Any, model: str) -> ChatResponseUpdate
 
     contents: list[Content] = []
     if consent_link:
+        # ``server_label`` identifies the MCP server that needs consent and is required by
+        # downstream Responses output items. It is copied into ``additional_properties``
+        # because ``raw_representation`` is provider specific and does not survive a
+        # session round trip.
+        server_label = getattr(raw_item, "server_label", None)
+        additional_properties = (
+            {"server_label": server_label} if isinstance(server_label, str) and server_label else None
+        )
         contents.append(
             Content.from_oauth_consent_request(
                 consent_link=consent_link,
+                additional_properties=additional_properties,
                 raw_representation=raw_item,
             )
         )
diff --git a/python/packages/foundry/tests/foundry/test_oauth_helpers.py b/python/packages/foundry/tests/foundry/test_oauth_helpers.py
index 2ab209e141e..46087e595d7 100644
--- a/python/packages/foundry/tests/foundry/test_oauth_helpers.py
+++ b/python/packages/foundry/tests/foundry/test_oauth_helpers.py
@@ -44,6 +44,23 @@ def test_validate_consent_link_rejects_non_url(caplog: pytest.LogCaptureFixture)
     assert result == ""
 
 
+def test_validate_consent_link_rejects_malformed_authority(caplog: pytest.LogCaptureFixture) -> None:
+    """A malformed authority makes urlparse raise ValueError; it must be rejected, not propagated."""
+    with caplog.at_level(logging.WARNING):
+        result = _validate_consent_link("https://[broken", "item-5")
+    assert result == ""
+    assert "malformed" in caplog.text
+    assert "item-5" in caplog.text
+
+
+def test_validate_consent_link_rejects_netloc_without_host(caplog: pytest.LogCaptureFixture) -> None:
+    """``https://@`` has a netloc but no host, so it is rejected."""
+    with caplog.at_level(logging.WARNING):
+        result = _validate_consent_link("https://@", "item-6")
+    assert result == ""
+    assert "non-HTTPS" in caplog.text
+
+
 # endregion
 
 # region try_parse_oauth_consent_event tests
@@ -54,6 +71,7 @@ def _make_output_item_event(
     item_type: str = "oauth_consent_request",
     consent_link: Any = "https://consent.example.com/auth",
     item_id: str = "oauth-item-1",
+    server_label: Any = "obo-mcp",
 ) -> MagicMock:
     """Create a mock ``response.output_item.added`` event."""
     event = MagicMock()
@@ -62,6 +80,7 @@ def _make_output_item_event(
     item.type = item_type
     item.consent_link = consent_link
     item.id = item_id
+    item.server_label = server_label
     event.item = item
     return event
 
@@ -70,12 +89,14 @@ def _make_top_level_event(
     *,
     consent_link: Any = "https://consent.example.com/authorize",
     event_id: str = "consent-event-1",
+    server_label: Any = "obo-mcp",
 ) -> MagicMock:
     """Create a mock ``response.oauth_consent_requested`` event."""
     event = MagicMock()
     event.type = "response.oauth_consent_requested"
     event.consent_link = consent_link
     event.id = event_id
+    event.server_label = server_label
     return event
 
 
@@ -161,4 +182,45 @@ def test_empty_contents_for_https_empty_netloc(caplog: pytest.LogCaptureFixture)
     assert "non-HTTPS" in caplog.text
 
 
+def test_empty_contents_for_malformed_authority(caplog: pytest.LogCaptureFixture) -> None:
+    """A malformed authority is rejected instead of raising out of the parser."""
+    event = _make_output_item_event(consent_link="https://[broken", item_id="item-malformed")
+    with caplog.at_level(logging.WARNING):
+        update = try_parse_oauth_consent_event(event, "test-model")
+
+    assert update is not None
+    assert len(update.contents) == 0
+    assert "malformed" in caplog.text
+
+
+def test_server_label_is_preserved_in_additional_properties() -> None:
+    """The upstream item's server_label is carried forward so hosting can re-emit it."""
+    event = _make_output_item_event(server_label="work-iq-connection")
+    update = try_parse_oauth_consent_event(event, "test-model")
+
+    assert update is not None
+    consent = [c for c in update.contents if c.type == "oauth_consent_request"]
+    assert consent[0].additional_properties["server_label"] == "work-iq-connection"
+
+
+def test_top_level_event_server_label_is_preserved() -> None:
+    """The top-level consent event's server_label is carried forward too."""
+    event = _make_top_level_event(server_label="work-iq-connection")
+    update = try_parse_oauth_consent_event(event, "test-model")
+
+    assert update is not None
+    consent = [c for c in update.contents if c.type == "oauth_consent_request"]
+    assert consent[0].additional_properties["server_label"] == "work-iq-connection"
+
+
+def test_missing_server_label_leaves_additional_properties_empty() -> None:
+    """A non-string server_label is ignored rather than stored."""
+    event = _make_output_item_event(server_label=None)
+    update = try_parse_oauth_consent_event(event, "test-model")
+
+    assert update is not None
+    consent = [c for c in update.contents if c.type == "oauth_consent_request"]
+    assert "server_label" not in consent[0].additional_properties
+
+
 # endregion
diff --git a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
index 8206c34e6bc..33d5f03884d 100644
--- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
+++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
@@ -165,23 +165,41 @@ def consent_url_from_error(exc: BaseException) -> list[ConsentError] | None:
     return None
 
 
+def _validated_consent_link(consent_link: str | None) -> str | None:
+    """Return *consent_link* when it is an absolute HTTPS URL with a host, else ``None``.
+
+    A consent link is rendered as a clickable prompt by the client, so anything that is
+    not an absolute ``https`` URL is dropped rather than surfaced. ``urlparse`` raises
+    ``ValueError`` for malformed authorities (for example ``https://[broken``), and a
+    non-empty ``netloc`` is not sufficient on its own (``https://@`` has one but no host),
+    so both conditions are handled here.
+    """
+    if not consent_link:
+        return None
+    try:
+        parsed = urlparse(consent_link)
+        hostname = parsed.hostname
+    except ValueError:
+        logger.warning("Skipping oauth_consent_request with a malformed consent_link.")
+        return None
+    if parsed.scheme.lower() != "https" or not hostname:
+        logger.warning("Skipping oauth_consent_request with a non-HTTPS consent_link.")
+        return None
+    return consent_link
+
+
 def _consent_link_from_content(content: Content) -> str | None:
     """Return a validated consent link for an ``oauth_consent_request`` content.
 
     Returns ``None`` when *content* is not an OAuth consent request, when it carries
-    no consent link, or when the link is not an absolute HTTPS URL.
+    no consent link, or when the link fails :func:`_validated_consent_link`.
     """
     if content.type != "oauth_consent_request":
         return None
-    consent_link = content.consent_link
-    if not consent_link:
+    if not content.consent_link:
         logger.warning("Received oauth_consent_request content without a consent_link; skipping.")
         return None
-    parsed = urlparse(consent_link)
-    if parsed.scheme.lower() != "https" or not parsed.netloc:
-        logger.warning("Skipping oauth_consent_request with non-HTTPS consent_link.")
-        return None
-    return consent_link
+    return _validated_consent_link(content.consent_link)
 
 
 def _emit_oauth_consent_item(
@@ -209,8 +227,16 @@ def _consent_incomplete_reason(count: int) -> str:
 
 
 def _consent_server_label(content: Content) -> str:
-    """Return the server label to report for an ``oauth_consent_request`` content."""
+    """Return the server label to report for an ``oauth_consent_request`` content.
+
+    Prefers the label carried in ``additional_properties``; falls back to the provider's
+    raw output item, which carries ``server_label`` as a required field. The raw
+    representation is provider specific and does not survive a session round trip, so it
+    is only a fallback.
+    """
     label = content.additional_properties.get("server_label") if content.additional_properties else None
+    if not isinstance(label, str) or not label:
+        label = getattr(content.raw_representation, "server_label", None)
     return label if isinstance(label, str) and label else "agent_framework"
 
 
@@ -394,19 +420,27 @@ async def _handle_inner_agent(
         try:
             await self._ensure_agent_ready()
         except AgentFrameworkException as ex:
-            consent_errors_to_emit = consent_url_from_error(ex)
-            if consent_errors_to_emit is None or len(consent_errors_to_emit) == 0:
+            consent_errors = consent_url_from_error(ex)
+            # A consent link the client cannot render is not an actionable consent prompt, so
+            # invalid links are dropped here and an entry-time failure with no usable link left
+            # is reported as ``response.failed`` rather than an ``incomplete`` the user cannot act on.
+            consent_errors_to_emit = [
+                (consent_error, link)
+                for consent_error in consent_errors or []
+                if (link := _validated_consent_link(consent_error.consent_url)) is not None
+            ]
+            if not consent_errors_to_emit:
                 logger.error("Failed to prepare agent: %s", ex, exc_info=(type(ex), ex, ex.__traceback__))
                 for event in self._emit_failure(response_event_stream, None, ex):
                     yield event
                 return
 
-            for consent_error in consent_errors_to_emit:
-                logger.warning("Consent URL for tool '%s': %s", consent_error.name, consent_error.consent_url)
+            for consent_error, consent_link in consent_errors_to_emit:
+                logger.warning("Consent URL for tool '%s': %s", consent_error.name, consent_link)
                 for event in _emit_oauth_consent_item(
                     response_event_stream,
                     context.response_id,
-                    consent_error.consent_url,
+                    consent_link,
                     consent_error.name,
                 ):
                     yield event
@@ -451,7 +485,7 @@ async def _handle_inner_agent(
         request_failure: Exception | None = None
         save_failure: Exception | None = None
         request_interrupted = False
-        pending_consent_count = 0
+        emitted_consent_requests: set[tuple[str, str]] = set()
 
         try:
             if self._uses_hosted_responses_history:
@@ -477,8 +511,6 @@ async def _handle_inner_agent(
 
             async for update in self._agent.run(stream=True, **run_kwargs):  # type: ignore[reportUnknownMemberType]
                 for content in update.contents:
-                    if _consent_link_from_content(content) is not None:
-                        pending_consent_count += 1
                     for event in tracker.handle(content):
                         yield event
                     if tracker.needs_async:
@@ -486,6 +518,7 @@ async def _handle_inner_agent(
                             response_event_stream,
                             content,
                             approval_storage=approval_storage,
+                            emitted_consent_requests=emitted_consent_requests,
                         ):
                             yield item
                         tracker.needs_async = False
@@ -528,10 +561,12 @@ async def _handle_inner_agent(
         elif save_failure is not None:
             for event in self._emit_failure(response_event_stream, tracker, save_failure):
                 yield event
-        elif pending_consent_count > 0:
+        elif emitted_consent_requests:
             # The turn cannot finish until the user completes OAuth consent, so the response
             # ends as `incomplete` rather than `completed`, matching the connect-time path.
-            yield response_event_stream.emit_incomplete(reason=_consent_incomplete_reason(pending_consent_count))
+            yield response_event_stream.emit_incomplete(
+                reason=_consent_incomplete_reason(len(emitted_consent_requests))
+            )
         else:
             yield response_event_stream.emit_completed()
 
@@ -632,7 +667,7 @@ async def _handle_inner_workflow(
                     pass
 
             tracker = _OutputItemTracker(response_event_stream)
-            pending_consent_count = 0
+            emitted_consent_requests: set[tuple[str, str]] = set()
 
             # Run the workflow agent in streaming mode with the new user input.
             async for update in self._agent.run(
@@ -641,13 +676,14 @@ async def _handle_inner_workflow(
                 checkpoint_storage=write_storage,
             ):
                 for content in update.contents:
-                    if _consent_link_from_content(content) is not None:
-                        pending_consent_count += 1
                     for event in tracker.handle(content):
                         yield event
                     if tracker.needs_async:
                         async for item in _to_outputs(
-                            response_event_stream, content, approval_storage=approval_storage
+                            response_event_stream,
+                            content,
+                            approval_storage=approval_storage,
+                            emitted_consent_requests=emitted_consent_requests,
                         ):
                             yield item
                         tracker.needs_async = False
@@ -657,8 +693,10 @@ async def _handle_inner_workflow(
                 yield event
 
             await self._delete_not_latest_checkpoints(write_storage, self._agent.workflow.name)
-            if pending_consent_count > 0:
-                yield response_event_stream.emit_incomplete(reason=_consent_incomplete_reason(pending_consent_count))
+            if emitted_consent_requests:
+                yield response_event_stream.emit_incomplete(
+                    reason=_consent_incomplete_reason(len(emitted_consent_requests))
+                )
             else:
                 yield response_event_stream.emit_completed()
         except Exception as ex:
@@ -1678,6 +1716,7 @@ async def _to_outputs(
     content: Content,
     *,
     approval_storage: FunctionApprovalStore | None = None,
+    emitted_consent_requests: set[tuple[str, str]] | None = None,
 ) -> AsyncIterator[ResponseStreamEvent]:
     """Converts a Content object to an async sequence of ResponseStreamEvent objects.
 
@@ -1685,6 +1724,8 @@ async def _to_outputs(
         stream: The ResponseEventStream to use for building events.
         content: The Content to convert.
         approval_storage: An optional ApprovalStorage instance to use for saving and loading function approval requests.
+        emitted_consent_requests: An optional set of ``(consent_link, server_label)`` pairs already emitted for
+            this response. It is updated in place and used to suppress duplicate OAuth consent prompts.
 
     Yields:
         ResponseStreamEvent: The converted event objects.
@@ -1799,13 +1840,21 @@ async def _to_outputs(
         # client can render a consent prompt instead of an empty assistant turn.
         consent_link = _consent_link_from_content(content)
         if consent_link is not None:
-            for event in _emit_oauth_consent_item(
-                stream,
-                str(stream.response["id"]),
-                consent_link,
-                _consent_server_label(content),
-            ):
-                yield event
+            server_label = _consent_server_label(content)
+            # A `WorkflowAgent` replays the inner agent's content as workflow output, so the
+            # same consent request can arrive more than once in one response. Emitting it
+            # twice would show the user duplicate consent prompts.
+            consent_key = (consent_link, server_label)
+            if emitted_consent_requests is None or consent_key not in emitted_consent_requests:
+                if emitted_consent_requests is not None:
+                    emitted_consent_requests.add(consent_key)
+                for event in _emit_oauth_consent_item(
+                    stream,
+                    str(stream.response["id"]),
+                    consent_link,
+                    server_label,
+                ):
+                    yield event
     else:
         # Log a warning for unsupported content types instead of raising an error to avoid breaking the response stream.
         logger.warning(f"Content type '{content.type}' is not supported yet. This is usually safe to ignore.")
diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py
index 103f0a61e70..aa200e6cacc 100644
--- a/python/packages/foundry_hosting/tests/test_responses.py
+++ b/python/packages/foundry_hosting/tests/test_responses.py
@@ -16,6 +16,7 @@
 from collections.abc import AsyncGenerator, AsyncIterator, Awaitable, Callable, Generator, Mapping, Sequence
 from contextlib import contextmanager
 from dataclasses import dataclass
+from types import SimpleNamespace
 from typing import Literal, cast, overload
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -3919,8 +3920,38 @@ async def test_multiple_consent_contents_each_emit_an_item(self) -> None:
 
         incomplete = [e for e in events if e["event"] == "response.incomplete"]
         assert len(incomplete) == 1
+        assert "2 tool(s)" in json.dumps(incomplete[0]["data"])
 
-    @pytest.mark.parametrize("consent_link", ["", "http://consent.example.com/obo", "not-a-url"])
+    async def test_repeated_consent_content_emits_one_item(self) -> None:
+        """The same consent request arriving twice must not produce duplicate prompts."""
+        agent = _make_agent(
+            stream_updates=[
+                AgentResponseUpdate(
+                    contents=[Content.from_oauth_consent_request(consent_link="https://consent.example.com/obo")],
+                    role="assistant",
+                ),
+                AgentResponseUpdate(
+                    contents=[Content.from_oauth_consent_request(consent_link="https://consent.example.com/obo")],
+                    role="assistant",
+                ),
+            ]
+        )
+        server = _make_server(agent)
+
+        resp = await _post(server, input_text="hello", stream=True)
+        events = _parse_sse_events(resp.text)
+        added = [e for e in events if e["event"] == "response.output_item.added"]
+        oauth_added = [e for e in added if e["data"]["item"]["type"] == "oauth_consent_request"]
+        assert len(oauth_added) == 1
+
+        incomplete = [e for e in events if e["event"] == "response.incomplete"]
+        assert len(incomplete) == 1
+        assert "1 tool(s)" in json.dumps(incomplete[0]["data"])
+
+    @pytest.mark.parametrize(
+        "consent_link",
+        ["", "http://consent.example.com/obo", "not-a-url", "https:///path", "https://[broken", "https://@"],
+    )
     async def test_invalid_consent_link_is_skipped(self, consent_link: str) -> None:
         agent = _make_agent(
             stream_updates=[
@@ -3939,8 +3970,53 @@ async def test_invalid_consent_link_is_skipped(self, consent_link: str) -> None:
 
         added = [e for e in events if e["event"] == "response.output_item.added"]
         assert not any(e["data"]["item"]["type"] == "oauth_consent_request" for e in added)
-        # A link we cannot render is not a consent prompt, so the turn still completes.
+        # A link we cannot render is not a consent prompt, so the turn still completes
+        # rather than failing the whole response.
         assert types[-1] == "response.completed"
+        assert "response.failed" not in types
+
+    async def test_server_label_falls_back_to_raw_representation(self) -> None:
+        """The Foundry parser carries ``server_label`` in additional properties, but a
+        content that only has the provider's raw item must still report its label.
+        """
+        raw_item = SimpleNamespace(server_label="raw-obo-mcp")
+        agent = _make_agent(
+            stream_updates=[
+                AgentResponseUpdate(
+                    contents=[
+                        Content.from_oauth_consent_request(
+                            consent_link="https://consent.example.com/obo",
+                            raw_representation=raw_item,
+                        )
+                    ],
+                    role="assistant",
+                )
+            ]
+        )
+        server = _make_server(agent)
+
+        resp = await _post(server, input_text="hello", stream=False)
+        body = resp.json()
+        oauth_items = [it for it in body["output"] if it["type"] == "oauth_consent_request"]
+        assert len(oauth_items) == 1
+        assert oauth_items[0]["server_label"] == "raw-obo-mcp"
+
+    async def test_connect_time_invalid_consent_link_fails_the_response(self) -> None:
+        """An entry-time consent error whose link cannot be rendered leaves nothing for the
+        user to act on, so the response fails instead of reporting ``incomplete``.
+        """
+        agent = _make_agent(
+            response=AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text("hi")])])
+        )
+        agent.__aenter__.side_effect = _make_consent_error("http://insecure.example.com/consent")
+        server = _make_server(agent)
+
+        resp = await _post(server, input_text="hello", stream=False)
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["status"] == "failed"
+        assert not any(it["type"] == "oauth_consent_request" for it in body.get("output", []))
+        agent.run.assert_not_called()
 
 
 # endregion
@@ -4238,13 +4314,18 @@ async def _iter() -> AsyncIterator[AgentResponseUpdate]:
 
 def _build_text_workflow_agent(text: str) -> WorkflowAgent:
     """Build a minimal ``WorkflowAgent`` whose inner agent emits a fixed text."""
+    return _build_contents_workflow_agent([Content.from_text(text=text)])
+
+
+def _build_contents_workflow_agent(contents: list[Content]) -> WorkflowAgent:
+    """Build a minimal ``WorkflowAgent`` whose inner agent emits fixed contents."""
 
     class _TextAgent(SupportsAgentRun):
-        def __init__(self, name: str, text: str) -> None:
+        def __init__(self, name: str, contents: list[Content]) -> None:
             self.id = str(uuid.uuid4())
             self.name = name
             self.description: str | None = None
-            self._text = text
+            self._contents = contents
 
         def create_session(self, **kwargs: Any) -> AgentSession:
             del kwargs
@@ -4286,19 +4367,19 @@ def run(
         ) -> Awaitable[AgentResponse] | ResponseStream[AgentResponseUpdate, AgentResponse]:
             del messages, session, kwargs
             assert stream is True, "The inner agent only runs in stream mode in Foundry Hosted Agents."
-            text = self._text
+            agent_contents = self._contents
             name = self.name
 
             async def _aiter() -> AsyncIterator[AgentResponseUpdate]:
                 yield AgentResponseUpdate(
-                    contents=[Content.from_text(text=text)],
+                    contents=agent_contents,
                     role="assistant",
                     author_name=name,
                 )
 
             return ResponseStream(_aiter(), finalizer=AgentResponse.from_updates)
 
-    inner = _TextAgent("text-agent", text)
+    inner = _TextAgent("text-agent", contents)
 
     @executor
     async def start(messages: list[Message], ctx: WorkflowContext[AgentExecutorRequest]) -> None:
@@ -4373,6 +4454,54 @@ async def test_basic_text_response_streaming(self) -> None:
         text_done = [e for e in events if e["event"] == "response.output_text.done"]
         assert any(e["data"]["text"] == "hello stream" for e in text_done)
 
+    async def test_mid_run_consent_emits_oauth_item_and_incomplete(self) -> None:
+        """A workflow that surfaces a consent request must emit the output item and end
+        ``incomplete``, after its checkpoint finalization, just like the regular agent path.
+        """
+        workflow_agent = _build_contents_workflow_agent([
+            Content.from_oauth_consent_request(
+                consent_link="https://consent.example.com/obo",
+                additional_properties={"server_label": "obo-mcp"},
+            )
+        ])
+        server = _make_server(workflow_agent)
+
+        resp = await _post(server, input_text="hi", stream=True)
+        assert resp.status_code == 200
+        events = _parse_sse_events(resp.text)
+        types = _sse_event_types(events)
+
+        assert types[-1] == "response.incomplete"
+
+        added = [e for e in events if e["event"] == "response.output_item.added"]
+        oauth_added = [e for e in added if e["data"]["item"]["type"] == "oauth_consent_request"]
+        assert len(oauth_added) == 1
+        assert oauth_added[0]["data"]["item"]["consent_link"] == "https://consent.example.com/obo"
+        assert oauth_added[0]["data"]["item"]["server_label"] == "obo-mcp"
+
+        done = [e for e in events if e["event"] == "response.output_item.done"]
+        assert any(e["data"]["item"]["type"] == "oauth_consent_request" for e in done)
+
+        # A WorkflowAgent replays the inner agent's content as workflow output, so the same
+        # consent request reaches the host twice and must not produce two consent prompts.
+        incomplete = [e for e in events if e["event"] == "response.incomplete"]
+        assert len(incomplete) == 1
+        assert "1 tool(s)" in json.dumps(incomplete[0]["data"])
+
+    async def test_mid_run_invalid_consent_link_still_completes(self) -> None:
+        workflow_agent = _build_contents_workflow_agent([
+            Content(type="oauth_consent_request", consent_link="https://[broken")
+        ])
+        server = _make_server(workflow_agent)
+
+        resp = await _post(server, input_text="hi", stream=True)
+        events = _parse_sse_events(resp.text)
+        types = _sse_event_types(events)
+
+        added = [e for e in events if e["event"] == "response.output_item.added"]
+        assert not any(e["data"]["item"]["type"] == "oauth_consent_request" for e in added)
+        assert types[-1] == "response.completed"
+
     async def test_non_streaming_emits_mcp_approval_request_and_persists_to_storage(self) -> None:
         workflow_agent, mock_agent = _build_approval_workflow_agent(approval_request_id="apr_wf_ns")
         server = _make_server(workflow_agent)

From 8f775204aa80706b6ce38a863f88a752c29674fe Mon Sep 17 00:00:00 2001
From: Giles Odigwe <gilesodigwe@microsoft.com>
Date: Mon, 17 Aug 2026 13:46:40 -0700
Subject: [PATCH 3/5] Python: harden consent link validation and fail on
 unusable consent links

Addresses two review findings on the mid-run OAuth consent surfacing.

Consent link validation was incomplete. `urlparse` only validates the port
when it is read, so `https://host:bad` and `https://host:99999` passed the
previous check, and a non-empty hostname was accepted even when it contained
characters no URL client can resolve (`https://exa mple.com`). `urlparse` also
silently strips tab and newline, letting control characters through. The
validator now reads `port` inside the guarded block, checks the hostname
against a permitted character set for both registered names and IPv6 literals,
and rejects whitespace and control characters before parsing. The same rules
are applied in the foundry parser and the hosting layer so the two agree.

Dropping an unusable link also erased the consent requirement: nothing was
recorded, so both host paths emitted `response.completed` and a blocked turn
looked successful, reproducing the silent drop this feature exists to fix.
Consent requests are now tracked in a `_ConsentTracker` holding the requests
that were emitted and the ones whose link could not be surfaced. A response
with at least one usable link still terminates as `incomplete`, and a response
where consent was required but no link could be shown terminates as
`response.failed`, matching the connect-time path.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 992057d8-78a8-45a7-9201-35af0919b071
---
 .../agent_framework_foundry/_oauth_helpers.py |  40 +++++-
 .../tests/foundry/test_oauth_helpers.py       |  60 +++++++++
 .../_responses.py                             | 117 ++++++++++++++----
 .../foundry_hosting/tests/test_responses.py   |  62 ++++++++--
 4 files changed, 245 insertions(+), 34 deletions(-)

diff --git a/python/packages/foundry/agent_framework_foundry/_oauth_helpers.py b/python/packages/foundry/agent_framework_foundry/_oauth_helpers.py
index c8d70fe97fb..6af6878d7a3 100644
--- a/python/packages/foundry/agent_framework_foundry/_oauth_helpers.py
+++ b/python/packages/foundry/agent_framework_foundry/_oauth_helpers.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import logging
+import re
 from typing import Any
 from urllib.parse import urlparse
 
@@ -10,17 +11,44 @@
 
 logger = logging.getLogger(__name__)
 
+# Characters allowed in a registered host name, and in the bracketed IPv6 form that
+# ``urlparse`` reports with its brackets already stripped.
+_HOST_PATTERN = re.compile(r"^[A-Za-z0-9._~%-]+$")
+_IPV6_HOST_PATTERN = re.compile(r"^[0-9A-Fa-f:.%-]+$")
+
+
+def _is_valid_consent_host(hostname: str) -> bool:
+    """Return whether *hostname* is syntactically usable by a standard URL client.
+
+    ``urlparse`` does not reject hosts that contain illegal characters, so values such as
+    ``exa mple.com`` are reported as a hostname even though no client can resolve them.
+    """
+    pattern = _IPV6_HOST_PATTERN if ":" in hostname else _HOST_PATTERN
+    return bool(pattern.match(hostname))
+
 
 def _validate_consent_link(consent_link: str, item_id: str) -> str:
-    """Validate a consent link is HTTPS with a valid host.
+    """Validate a consent link is HTTPS with a valid host and port.
 
     Returns the link unchanged if valid, or an empty string if not. ``urlparse`` raises
-    ``ValueError`` for malformed authorities (for example ``https://[broken``), and a
-    non-empty ``netloc`` is not sufficient on its own (``https://@`` has one but no host).
+    ``ValueError`` for malformed authorities (for example ``https://[broken``) and for
+    invalid ports, but only when ``port`` is read, so it is accessed here. A non-empty
+    ``netloc`` is not sufficient on its own (``https://@`` has one but no host), and a
+    non-empty ``hostname`` is not either (``https://exa mple.com`` reports one).
     """
+    if any(char.isspace() or ord(char) < 0x20 or ord(char) == 0x7F for char in consent_link):
+        # ``urlparse`` silently strips tab and newline, which would let a link carrying
+        # control characters through even though it is not safe to render or log.
+        logger.warning(
+            "Skipping oauth_consent_request with whitespace or control characters in consent_link (item id=%s)",
+            item_id,
+        )
+        return ""
     try:
         parsed = urlparse(consent_link)
         hostname = parsed.hostname
+        # Reading ``port`` is what validates it; ``https://host:bad`` raises here.
+        _ = parsed.port
     except ValueError:
         logger.warning(
             "Skipping oauth_consent_request with malformed consent_link (item id=%s)",
@@ -33,6 +61,12 @@ def _validate_consent_link(consent_link: str, item_id: str) -> str:
             item_id,
         )
         return ""
+    if not _is_valid_consent_host(hostname):
+        logger.warning(
+            "Skipping oauth_consent_request with an invalid consent_link host (item id=%s)",
+            item_id,
+        )
+        return ""
     return consent_link
 
 
diff --git a/python/packages/foundry/tests/foundry/test_oauth_helpers.py b/python/packages/foundry/tests/foundry/test_oauth_helpers.py
index 46087e595d7..8e98e9f94b7 100644
--- a/python/packages/foundry/tests/foundry/test_oauth_helpers.py
+++ b/python/packages/foundry/tests/foundry/test_oauth_helpers.py
@@ -61,6 +61,66 @@ def test_validate_consent_link_rejects_netloc_without_host(caplog: pytest.LogCap
     assert "non-HTTPS" in caplog.text
 
 
+@pytest.mark.parametrize(
+    "consent_link",
+    [
+        "https://consent.example.com:bad/obo",
+        "https://consent.example.com:99999/obo",
+    ],
+)
+def test_validate_consent_link_rejects_invalid_port(consent_link: str, caplog: pytest.LogCaptureFixture) -> None:
+    """``urlparse`` only validates the port when it is read, so an invalid port must be caught."""
+    with caplog.at_level(logging.WARNING):
+        result = _validate_consent_link(consent_link, "item-7")
+    assert result == ""
+    assert "malformed" in caplog.text
+
+
+@pytest.mark.parametrize(
+    "consent_link",
+    [
+        "https://cons|ent.example.com/obo",
+        "https://exa^mple.com/obo",
+    ],
+)
+def test_validate_consent_link_rejects_invalid_host_characters(
+    consent_link: str, caplog: pytest.LogCaptureFixture
+) -> None:
+    """``urlparse`` reports a hostname for values that no URL client can resolve."""
+    with caplog.at_level(logging.WARNING):
+        result = _validate_consent_link(consent_link, "item-8")
+    assert result == ""
+    assert "invalid consent_link host" in caplog.text
+
+
+@pytest.mark.parametrize(
+    "consent_link",
+    [
+        "https://cons\tent.example.com/obo",
+        "https://consent.example.com/obo\n",
+    ],
+)
+def test_validate_consent_link_rejects_control_characters(consent_link: str, caplog: pytest.LogCaptureFixture) -> None:
+    """``urlparse`` strips tab and newline, so they must be rejected before parsing."""
+    with caplog.at_level(logging.WARNING):
+        result = _validate_consent_link(consent_link, "item-10")
+    assert result == ""
+    assert "control characters" in caplog.text
+
+
+@pytest.mark.parametrize(
+    "consent_link",
+    [
+        "https://consent.example.com/obo",
+        "https://consent.example.com:8443/obo",
+        "https://[2001:db8::1]/obo",
+    ],
+)
+def test_validate_consent_link_accepts_usable_links(consent_link: str) -> None:
+    """Valid ports and IPv6 literals stay usable and must not be dropped."""
+    assert _validate_consent_link(consent_link, "item-9") == consent_link
+
+
 # endregion
 
 # region try_parse_oauth_consent_event tests
diff --git a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
index 51ee14edfb0..e6652d03973 100644
--- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
+++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
@@ -7,9 +7,10 @@
 import json
 import logging
 import os
+import re
 from collections.abc import AsyncIterable, AsyncIterator, Generator, Mapping, Sequence
 from contextlib import AbstractAsyncContextManager, AsyncExitStack
-from dataclasses import asdict, dataclass, is_dataclass
+from dataclasses import asdict, dataclass, field, is_dataclass
 from typing import Literal, cast
 from urllib.parse import urlparse
 
@@ -176,26 +177,53 @@ def consent_url_from_error(exc: BaseException) -> list[ConsentError] | None:
     return None
 
 
+# Characters allowed in a registered host name, and in the bracketed IPv6 form that
+# ``urlparse`` reports with its brackets already stripped.
+_HOST_PATTERN = re.compile(r"^[A-Za-z0-9._~%-]+$")
+_IPV6_HOST_PATTERN = re.compile(r"^[0-9A-Fa-f:.%-]+$")
+
+
+def _is_valid_consent_host(hostname: str) -> bool:
+    """Return whether *hostname* is syntactically usable by a standard URL client.
+
+    ``urlparse`` does not reject hosts that contain illegal characters, so values such as
+    ``exa mple.com`` are reported as a hostname even though no client can resolve them.
+    """
+    pattern = _IPV6_HOST_PATTERN if ":" in hostname else _HOST_PATTERN
+    return bool(pattern.match(hostname))
+
+
 def _validated_consent_link(consent_link: str | None) -> str | None:
-    """Return *consent_link* when it is an absolute HTTPS URL with a host, else ``None``.
+    """Return *consent_link* when it is an absolute HTTPS URL a client can open, else ``None``.
 
     A consent link is rendered as a clickable prompt by the client, so anything that is
     not an absolute ``https`` URL is dropped rather than surfaced. ``urlparse`` raises
-    ``ValueError`` for malformed authorities (for example ``https://[broken``), and a
-    non-empty ``netloc`` is not sufficient on its own (``https://@`` has one but no host),
-    so both conditions are handled here.
+    ``ValueError`` for malformed authorities (for example ``https://[broken``) and for
+    invalid ports, but only when ``port`` is read, so it is accessed here. A non-empty
+    ``netloc`` is not sufficient on its own (``https://@`` has one but no host), and a
+    non-empty ``hostname`` is not either (``https://exa mple.com`` reports one).
     """
     if not consent_link:
         return None
+    if any(char.isspace() or ord(char) < 0x20 or ord(char) == 0x7F for char in consent_link):
+        # ``urlparse`` silently strips tab and newline, which would let a link carrying
+        # control characters through even though it is not safe to render or log.
+        logger.warning("Skipping oauth_consent_request with whitespace in the consent_link.")
+        return None
     try:
         parsed = urlparse(consent_link)
         hostname = parsed.hostname
+        # Reading ``port`` is what validates it; ``https://host:bad`` raises here.
+        _ = parsed.port
     except ValueError:
         logger.warning("Skipping oauth_consent_request with a malformed consent_link.")
         return None
     if parsed.scheme.lower() != "https" or not hostname:
         logger.warning("Skipping oauth_consent_request with a non-HTTPS consent_link.")
         return None
+    if not _is_valid_consent_host(hostname):
+        logger.warning("Skipping oauth_consent_request with an invalid consent_link host.")
+        return None
     return consent_link
 
 
@@ -237,6 +265,30 @@ def _consent_incomplete_reason(count: int) -> str:
     return f"OAuth consent required for {count} tool(s)."
 
 
+def _consent_unusable_link_error(count: int) -> RuntimeError:
+    """Return the error reported when consent is required but no link can be surfaced."""
+    return RuntimeError(
+        f"OAuth consent is required for {count} tool(s), but no usable HTTPS consent link was "
+        "provided, so the request cannot be completed."
+    )
+
+
+@dataclass
+class _ConsentTracker:
+    """Tracks the OAuth consent requests seen while converting content for one response.
+
+    A consent request means the turn is blocked until the user grants access, so the
+    outcome has to be recorded even when the link cannot be shown. ``emitted`` drives the
+    terminal ``response.incomplete``, while ``dropped`` records requests whose link was
+    unusable so that the response fails instead of silently reporting success. Both are
+    keyed by ``(consent_link, server_label)`` because a ``WorkflowAgent`` replays the
+    inner agent's content as workflow output, so the same request can arrive twice.
+    """
+
+    emitted: set[tuple[str, str]] = field(default_factory=set[tuple[str, str]])
+    dropped: set[tuple[str, str]] = field(default_factory=set[tuple[str, str]])
+
+
 def _consent_server_label(content: Content) -> str:
     """Return the server label to report for an ``oauth_consent_request`` content.
 
@@ -485,7 +537,7 @@ async def _handle_inner_agent(
         request_failure: Exception | None = None
         save_failure: Exception | None = None
         request_interrupted = False
-        emitted_consent_requests: set[tuple[str, str]] = set()
+        consent_tracker = _ConsentTracker()
 
         try:
             if self._uses_hosted_responses_history:
@@ -518,7 +570,7 @@ async def _handle_inner_agent(
                             response_event_stream,
                             content,
                             approval_storage=approval_storage,
-                            emitted_consent_requests=emitted_consent_requests,
+                            consent_tracker=consent_tracker,
                         ):
                             yield item
                         tracker.needs_async = False
@@ -561,12 +613,17 @@ async def _handle_inner_agent(
         elif save_failure is not None:
             for event in self._emit_failure(response_event_stream, tracker, save_failure):
                 yield event
-        elif emitted_consent_requests:
+        elif consent_tracker.emitted:
             # The turn cannot finish until the user completes OAuth consent, so the response
             # ends as `incomplete` rather than `completed`, matching the connect-time path.
-            yield response_event_stream.emit_incomplete(
-                reason=_consent_incomplete_reason(len(emitted_consent_requests))
-            )
+            yield response_event_stream.emit_incomplete(reason=_consent_incomplete_reason(len(consent_tracker.emitted)))
+        elif consent_tracker.dropped:
+            # Consent was required but no link could be surfaced, so there is nothing for the
+            # user to act on. Failing is the honest outcome; `completed` would hide the block.
+            for event in self._emit_failure(
+                response_event_stream, tracker, _consent_unusable_link_error(len(consent_tracker.dropped))
+            ):
+                yield event
         else:
             yield response_event_stream.emit_completed()
 
@@ -666,7 +723,7 @@ async def _handle_inner_workflow(
                     pass
 
             tracker = _OutputItemTracker(response_event_stream)
-            emitted_consent_requests: set[tuple[str, str]] = set()
+            consent_tracker = _ConsentTracker()
 
             # Run the workflow agent in streaming mode with the new user input.
             async for update in self._agent.run(
@@ -682,7 +739,7 @@ async def _handle_inner_workflow(
                             response_event_stream,
                             content,
                             approval_storage=approval_storage,
-                            emitted_consent_requests=emitted_consent_requests,
+                            consent_tracker=consent_tracker,
                         ):
                             yield item
                         tracker.needs_async = False
@@ -691,10 +748,17 @@ async def _handle_inner_workflow(
             for event in tracker.close():
                 yield event
 
-            if emitted_consent_requests:
+            if consent_tracker.emitted:
                 yield response_event_stream.emit_incomplete(
-                    reason=_consent_incomplete_reason(len(emitted_consent_requests))
+                    reason=_consent_incomplete_reason(len(consent_tracker.emitted))
                 )
+            elif consent_tracker.dropped:
+                # Consent was required but no link could be surfaced, so there is nothing for
+                # the user to act on. Failing is the honest outcome; `completed` would hide it.
+                for event in self._emit_failure(
+                    response_event_stream, tracker, _consent_unusable_link_error(len(consent_tracker.dropped))
+                ):
+                    yield event
             else:
                 yield response_event_stream.emit_completed()
         except Exception as ex:
@@ -1701,7 +1765,7 @@ async def _to_outputs(
     content: Content,
     *,
     approval_storage: FunctionApprovalStore | None = None,
-    emitted_consent_requests: set[tuple[str, str]] | None = None,
+    consent_tracker: _ConsentTracker | None = None,
 ) -> AsyncIterator[ResponseStreamEvent]:
     """Converts a Content object to an async sequence of ResponseStreamEvent objects.
 
@@ -1709,8 +1773,9 @@ async def _to_outputs(
         stream: The ResponseEventStream to use for building events.
         content: The Content to convert.
         approval_storage: An optional ApprovalStorage instance to use for saving and loading function approval requests.
-        emitted_consent_requests: An optional set of ``(consent_link, server_label)`` pairs already emitted for
-            this response. It is updated in place and used to suppress duplicate OAuth consent prompts.
+        consent_tracker: An optional :class:`_ConsentTracker` recording the OAuth consent requests seen for this
+            response. It is updated in place, used to suppress duplicate consent prompts, and read by the caller to
+            pick the terminal event.
 
     Yields:
         ResponseStreamEvent: The converted event objects.
@@ -1823,16 +1888,22 @@ async def _to_outputs(
         # An OBO/on-behalf-of tool can require consent mid-run, after the agent has already
         # been entered. Surface the link as an `oauth_consent_request` output item so the
         # client can render a consent prompt instead of an empty assistant turn.
+        server_label = _consent_server_label(content)
         consent_link = _consent_link_from_content(content)
-        if consent_link is not None:
-            server_label = _consent_server_label(content)
+        if consent_link is None:
+            # The turn is still blocked on consent even though the link cannot be rendered,
+            # so record it: reporting success here would look exactly like the silent drop
+            # this branch exists to fix.
+            if consent_tracker is not None:
+                consent_tracker.dropped.add((content.consent_link or "", server_label))
+        else:
             # A `WorkflowAgent` replays the inner agent's content as workflow output, so the
             # same consent request can arrive more than once in one response. Emitting it
             # twice would show the user duplicate consent prompts.
             consent_key = (consent_link, server_label)
-            if emitted_consent_requests is None or consent_key not in emitted_consent_requests:
-                if emitted_consent_requests is not None:
-                    emitted_consent_requests.add(consent_key)
+            if consent_tracker is None or consent_key not in consent_tracker.emitted:
+                if consent_tracker is not None:
+                    consent_tracker.emitted.add(consent_key)
                 for event in _emit_oauth_consent_item(
                     stream,
                     str(stream.response["id"]),
diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py
index 6b289874f6d..ce7bc55bab4 100644
--- a/python/packages/foundry_hosting/tests/test_responses.py
+++ b/python/packages/foundry_hosting/tests/test_responses.py
@@ -3865,9 +3865,20 @@ async def test_repeated_consent_content_emits_one_item(self) -> None:
 
     @pytest.mark.parametrize(
         "consent_link",
-        ["", "http://consent.example.com/obo", "not-a-url", "https:///path", "https://[broken", "https://@"],
+        [
+            "",
+            "http://consent.example.com/obo",
+            "not-a-url",
+            "https:///path",
+            "https://[broken",
+            "https://@",
+            "https://consent.example.com:bad/obo",
+            "https://consent.example.com:99999/obo",
+            "https://cons ent.example.com/obo",
+            "https://cons|ent.example.com/obo",
+        ],
     )
-    async def test_invalid_consent_link_is_skipped(self, consent_link: str) -> None:
+    async def test_unusable_consent_link_fails_the_response(self, consent_link: str) -> None:
         agent = _make_agent(
             stream_updates=[
                 AgentResponseUpdate(
@@ -3885,10 +3896,41 @@ async def test_invalid_consent_link_is_skipped(self, consent_link: str) -> None:
 
         added = [e for e in events if e["event"] == "response.output_item.added"]
         assert not any(e["data"]["item"]["type"] == "oauth_consent_request" for e in added)
-        # A link we cannot render is not a consent prompt, so the turn still completes
-        # rather than failing the whole response.
-        assert types[-1] == "response.completed"
-        assert "response.failed" not in types
+        # The turn is still blocked on consent, so reporting success would repeat the silent
+        # drop this feature exists to fix. There is no link to act on, hence `failed`.
+        assert types[-1] == "response.failed"
+        assert "response.completed" not in types
+        assert "response.incomplete" not in types
+        failed = [e for e in events if e["event"] == "response.failed"]
+        assert "OAuth consent is required" in failed[0]["data"]["response"]["error"]["message"]
+
+    async def test_usable_consent_link_still_wins_over_an_unusable_one(self) -> None:
+        """A response with one renderable link stays actionable, so it ends `incomplete`."""
+        agent = _make_agent(
+            stream_updates=[
+                AgentResponseUpdate(
+                    contents=[
+                        Content(type="oauth_consent_request", consent_link="http://insecure.example.com/obo"),
+                        Content(type="oauth_consent_request", consent_link="https://consent.example.com/obo"),
+                    ],
+                    role="assistant",
+                ),
+            ]
+        )
+        server = _make_server(agent)
+
+        resp = await _post(server, input_text="hello", stream=True)
+        events = _parse_sse_events(resp.text)
+        types = _sse_event_types(events)
+
+        added = [
+            e
+            for e in events
+            if e["event"] == "response.output_item.added" and e["data"]["item"]["type"] == "oauth_consent_request"
+        ]
+        assert len(added) == 1
+        assert added[0]["data"]["item"]["consent_link"] == "https://consent.example.com/obo"
+        assert types[-1] == "response.incomplete"
 
     async def test_server_label_falls_back_to_raw_representation(self) -> None:
         """The Foundry parser carries ``server_label`` in additional properties, but a
@@ -4403,7 +4445,7 @@ async def test_mid_run_consent_emits_oauth_item_and_incomplete(self) -> None:
         assert len(incomplete) == 1
         assert "1 tool(s)" in json.dumps(incomplete[0]["data"])
 
-    async def test_mid_run_invalid_consent_link_still_completes(self) -> None:
+    async def test_mid_run_unusable_consent_link_fails_the_workflow_response(self) -> None:
         workflow_agent = _build_contents_workflow_agent([
             Content(type="oauth_consent_request", consent_link="https://[broken")
         ])
@@ -4415,7 +4457,11 @@ async def test_mid_run_invalid_consent_link_still_completes(self) -> None:
 
         added = [e for e in events if e["event"] == "response.output_item.added"]
         assert not any(e["data"]["item"]["type"] == "oauth_consent_request" for e in added)
-        assert types[-1] == "response.completed"
+        # The workflow is blocked on consent with no link to show, so it must not report success.
+        assert types[-1] == "response.failed"
+        assert "response.completed" not in types
+        failed = [e for e in events if e["event"] == "response.failed"]
+        assert "OAuth consent is required" in failed[0]["data"]["response"]["error"]["message"]
 
     async def test_non_streaming_emits_mcp_approval_request_and_persists_to_storage(self) -> None:
         workflow_agent, mock_agent = _build_approval_workflow_agent(approval_request_id="apr_wf_ns")

From 62b83ce80f4f19c75f57246a60ef50fe81bfdc67 Mon Sep 17 00:00:00 2001
From: Giles Odigwe <gilesodigwe@microsoft.com>
Date: Wed, 19 Aug 2026 12:54:34 -0700
Subject: [PATCH 4/5] Address maintainer review on OAuth consent surfacing

Share the consent-link validator and stop dropping unusable consent
requests at the parser layer.

- Add `agent_framework/_oauth.py` with `validate_oauth_consent_link`,
  the single definition of what makes a consent link renderable. The
  Foundry parser and the Foundry hosting layer had drifted copies of
  these rules; both now delegate to it while keeping their own empty
  string vs `None` return contracts. `foundry_hosting` depends on
  `agent-framework-core` but not on `agent-framework-foundry`, so core
  is the only module both packages can reach.

- Extract `_finish_consent_response` so the agent and workflow paths
  share one definition of the terminal status precedence: a surfaced
  consent link ends the turn `incomplete`, an unusable one ends it
  `failed`, otherwise `completed`. Higher precedence request and
  session persistence failures still apply before it.

- Always surface an `oauth_consent_request` marker from
  `try_parse_oauth_consent_event`, even when the link is missing or
  unusable. Returning empty contents meant the host never recorded the
  request as dropped, so the exact unusable link case this change
  exists to fail was reported as `response.completed` instead. Link
  validation stays in the parser for diagnostics, and the host remains
  the single authority on whether a link is renderable.

Tests cover the shared validator in core, the preserved marker in the
parser, and the existing host side failure path.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 992057d8-78a8-45a7-9201-35af0919b071
---
 .../packages/core/agent_framework/_oauth.py   |  79 +++++++++++++
 python/packages/core/tests/test_oauth.py      |  56 +++++++++
 .../agent_framework_foundry/_oauth_helpers.py | 102 +++++------------
 .../tests/foundry/test_foundry_agent.py       |  17 +--
 .../tests/foundry/test_foundry_chat_client.py |  17 +--
 .../tests/foundry/test_oauth_helpers.py       |  73 +++++-------
 .../_responses.py                             | 106 ++++++------------
 7 files changed, 244 insertions(+), 206 deletions(-)
 create mode 100644 python/packages/core/agent_framework/_oauth.py
 create mode 100644 python/packages/core/tests/test_oauth.py

diff --git a/python/packages/core/agent_framework/_oauth.py b/python/packages/core/agent_framework/_oauth.py
new file mode 100644
index 00000000000..7982b5cd055
--- /dev/null
+++ b/python/packages/core/agent_framework/_oauth.py
@@ -0,0 +1,79 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from __future__ import annotations
+
+import logging
+import re
+from urllib.parse import urlparse
+
+logger = logging.getLogger("agent_framework")
+
+__all__ = ["validate_oauth_consent_link"]
+
+# Characters allowed in a registered host name, and in the bracketed IPv6 form that
+# ``urlparse`` reports with its brackets already stripped.
+_HOST_PATTERN = re.compile(r"^[A-Za-z0-9._~%-]+$")
+_IPV6_HOST_PATTERN = re.compile(r"^[0-9A-Fa-f:.%-]+$")
+
+
+def _is_valid_host(hostname: str) -> bool:
+    """Return whether *hostname* is syntactically usable by a standard URL client.
+
+    ``urlparse`` does not reject hosts containing illegal characters, so values such as
+    ``exa mple.com`` are reported as a hostname even though no client can resolve them.
+    """
+    pattern = _IPV6_HOST_PATTERN if ":" in hostname else _HOST_PATTERN
+    return bool(pattern.match(hostname))
+
+
+def validate_oauth_consent_link(consent_link: str | None, *, item_id: str | None = None) -> str | None:
+    """Return *consent_link* when it is an absolute HTTPS URL a client can open, else ``None``.
+
+    A consent link is rendered as a clickable prompt by the client, so anything that is not
+    an absolute ``https`` URL is dropped rather than surfaced. Validation is shared by every
+    package that parses or re-emits ``oauth_consent_request`` content so the accepted shape
+    cannot drift between the provider that parses a link and the host that renders it.
+
+    ``urlparse`` is permissive in three ways that matter here, all handled below:
+
+    * it raises ``ValueError`` for malformed authorities (``https://[broken``) and for invalid
+      ports, but the port is only validated when it is read;
+    * a non-empty ``netloc`` does not imply a host (``https://@`` has one but no host), and a
+      non-empty ``hostname`` does not imply a usable one (``https://exa mple.com`` reports one);
+    * it silently strips tab and newline, so control characters must be rejected up front.
+
+    Args:
+        consent_link: The candidate consent URL, which may be ``None`` or empty.
+
+    Keyword Args:
+        item_id: Optional identifier of the source item, included in warning logs.
+
+    Returns:
+        The link unchanged when it is usable, otherwise ``None``.
+    """
+    if not consent_link:
+        return None
+
+    log_id = item_id or "<unknown>"
+
+    if any(char.isspace() or ord(char) < 0x20 or ord(char) == 0x7F for char in consent_link):
+        logger.warning(
+            "Skipping oauth_consent_request with whitespace or control characters in consent_link (item id=%s)",
+            log_id,
+        )
+        return None
+    try:
+        parsed = urlparse(consent_link)
+        hostname = parsed.hostname
+        # Reading ``port`` is what validates it; ``https://host:bad`` raises here.
+        _ = parsed.port
+    except ValueError:
+        logger.warning("Skipping oauth_consent_request with malformed consent_link (item id=%s)", log_id)
+        return None
+    if parsed.scheme.lower() != "https" or not hostname:
+        logger.warning("Skipping oauth_consent_request with non-HTTPS consent_link (item id=%s)", log_id)
+        return None
+    if not _is_valid_host(hostname):
+        logger.warning("Skipping oauth_consent_request with an invalid consent_link host (item id=%s)", log_id)
+        return None
+    return consent_link
diff --git a/python/packages/core/tests/test_oauth.py b/python/packages/core/tests/test_oauth.py
new file mode 100644
index 00000000000..4aeb3fe282d
--- /dev/null
+++ b/python/packages/core/tests/test_oauth.py
@@ -0,0 +1,56 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import pytest
+
+from agent_framework._oauth import validate_oauth_consent_link
+
+
+@pytest.mark.parametrize(
+    "link",
+    [
+        "https://login.example.com/consent",
+        "https://login.example.com:8443/consent?state=abc#frag",
+        "https://192.0.2.10/consent",
+        "https://[2001:db8::1]:8443/consent",
+        "HTTPS://login.example.com/consent",
+    ],
+)
+def test_usable_links_are_returned_unchanged(link: str) -> None:
+    assert validate_oauth_consent_link(link) == link
+
+
+@pytest.mark.parametrize(
+    ("link", "reason"),
+    [
+        (None, "missing"),
+        ("", "empty"),
+        ("   ", "whitespace only"),
+        ("http://login.example.com/consent", "non-HTTPS scheme"),
+        ("ftp://login.example.com/consent", "non-HTTPS scheme"),
+        ("/consent", "relative, so no scheme or host"),
+        ("https://", "no host"),
+        ("https://@", "netloc present but no host"),
+        ("https://[broken", "malformed authority, urlparse raises"),
+        ("https://login.example.com:bad/consent", "port only validated when read"),
+        ("https://login.example.com:99999/consent", "port out of range"),
+        ("https://exa mple.com/consent", "space in host, unresolvable"),
+        ("https://cons|ent.example.com/", "illegal character in host"),
+        ("https://login.example.com/consent\n", "trailing newline, silently stripped by urlparse"),
+        ("https://login.example.com/\tconsent", "embedded tab, silently stripped by urlparse"),
+    ],
+)
+def test_unusable_links_are_rejected(link: str | None, reason: str) -> None:
+    assert validate_oauth_consent_link(link) is None, reason
+
+
+def test_rejection_is_logged_with_the_item_id(caplog: pytest.LogCaptureFixture) -> None:
+    with caplog.at_level("WARNING"):
+        assert validate_oauth_consent_link("http://login.example.com", item_id="item-5") is None
+    assert "non-HTTPS" in caplog.text
+    assert "item-5" in caplog.text
+
+
+def test_rejection_without_an_item_id_still_logs(caplog: pytest.LogCaptureFixture) -> None:
+    with caplog.at_level("WARNING"):
+        assert validate_oauth_consent_link("http://login.example.com") is None
+    assert "non-HTTPS" in caplog.text
diff --git a/python/packages/foundry/agent_framework_foundry/_oauth_helpers.py b/python/packages/foundry/agent_framework_foundry/_oauth_helpers.py
index 6af6878d7a3..c06146644b6 100644
--- a/python/packages/foundry/agent_framework_foundry/_oauth_helpers.py
+++ b/python/packages/foundry/agent_framework_foundry/_oauth_helpers.py
@@ -3,71 +3,22 @@
 from __future__ import annotations
 
 import logging
-import re
 from typing import Any
-from urllib.parse import urlparse
 
 from agent_framework import ChatResponseUpdate, Content
+from agent_framework._oauth import validate_oauth_consent_link
 
 logger = logging.getLogger(__name__)
 
-# Characters allowed in a registered host name, and in the bracketed IPv6 form that
-# ``urlparse`` reports with its brackets already stripped.
-_HOST_PATTERN = re.compile(r"^[A-Za-z0-9._~%-]+$")
-_IPV6_HOST_PATTERN = re.compile(r"^[0-9A-Fa-f:.%-]+$")
-
-
-def _is_valid_consent_host(hostname: str) -> bool:
-    """Return whether *hostname* is syntactically usable by a standard URL client.
-
-    ``urlparse`` does not reject hosts that contain illegal characters, so values such as
-    ``exa mple.com`` are reported as a hostname even though no client can resolve them.
-    """
-    pattern = _IPV6_HOST_PATTERN if ":" in hostname else _HOST_PATTERN
-    return bool(pattern.match(hostname))
-
 
 def _validate_consent_link(consent_link: str, item_id: str) -> str:
     """Validate a consent link is HTTPS with a valid host and port.
 
-    Returns the link unchanged if valid, or an empty string if not. ``urlparse`` raises
-    ``ValueError`` for malformed authorities (for example ``https://[broken``) and for
-    invalid ports, but only when ``port`` is read, so it is accessed here. A non-empty
-    ``netloc`` is not sufficient on its own (``https://@`` has one but no host), and a
-    non-empty ``hostname`` is not either (``https://exa mple.com`` reports one).
+    Thin wrapper over the shared core validator that keeps this module's empty-string
+    contract. The rules live in ``agent_framework._oauth`` so the parser here and the
+    Foundry hosting layer that re-emits the link cannot drift apart.
     """
-    if any(char.isspace() or ord(char) < 0x20 or ord(char) == 0x7F for char in consent_link):
-        # ``urlparse`` silently strips tab and newline, which would let a link carrying
-        # control characters through even though it is not safe to render or log.
-        logger.warning(
-            "Skipping oauth_consent_request with whitespace or control characters in consent_link (item id=%s)",
-            item_id,
-        )
-        return ""
-    try:
-        parsed = urlparse(consent_link)
-        hostname = parsed.hostname
-        # Reading ``port`` is what validates it; ``https://host:bad`` raises here.
-        _ = parsed.port
-    except ValueError:
-        logger.warning(
-            "Skipping oauth_consent_request with malformed consent_link (item id=%s)",
-            item_id,
-        )
-        return ""
-    if parsed.scheme.lower() != "https" or not hostname:
-        logger.warning(
-            "Skipping oauth_consent_request with non-HTTPS consent_link (item id=%s)",
-            item_id,
-        )
-        return ""
-    if not _is_valid_consent_host(hostname):
-        logger.warning(
-            "Skipping oauth_consent_request with an invalid consent_link host (item id=%s)",
-            item_id,
-        )
-        return ""
-    return consent_link
+    return validate_oauth_consent_link(consent_link, item_id=item_id) or ""
 
 
 def try_parse_oauth_consent_event(event: Any, model: str) -> ChatResponseUpdate | None:
@@ -77,6 +28,10 @@ def try_parse_oauth_consent_event(event: Any, model: str) -> ChatResponseUpdate
     ``response.output_item.added`` carrying an ``oauth_consent_request`` item
     or a top-level ``response.oauth_consent_requested`` event,
     or ``None`` so the caller can fall through to the base implementation.
+
+    The consent request is surfaced even when its link is missing or unusable, so that a
+    turn which cannot proceed is never reported as a silent success. Link validation is
+    applied for diagnostics here and enforced by the host that renders the link.
     """
     consent_link: str = ""
     raw_item: Any = None
@@ -95,31 +50,32 @@ def try_parse_oauth_consent_event(event: Any, model: str) -> ChatResponseUpdate
     item_id = getattr(raw_item, "id", "<unknown>")
 
     if consent_link:
-        consent_link = _validate_consent_link(consent_link, item_id)
-
-    contents: list[Content] = []
-    if consent_link:
-        # ``server_label`` identifies the MCP server that needs consent and is required by
-        # downstream Responses output items. It is copied into ``additional_properties``
-        # because ``raw_representation`` is provider specific and does not survive a
-        # session round trip.
-        server_label = getattr(raw_item, "server_label", None)
-        additional_properties = (
-            {"server_label": server_label} if isinstance(server_label, str) and server_label else None
-        )
-        contents.append(
-            Content.from_oauth_consent_request(
-                consent_link=consent_link,
-                additional_properties=additional_properties,
-                raw_representation=raw_item,
-            )
-        )
+        # Validation here is diagnostic only. The provider has signalled that the turn
+        # cannot proceed without consent, so the request is always surfaced: dropping it
+        # would let a blocked turn finish as a silent success. The host re-validates and
+        # is the single authority on whether a link is renderable, failing the response
+        # when it is not.
+        _validate_consent_link(consent_link, item_id)
     else:
         logger.warning(
             "Received oauth_consent_request output without valid consent_link (item id=%s)",
             item_id,
         )
 
+    # ``server_label`` identifies the MCP server that needs consent and is required by
+    # downstream Responses output items. It is copied into ``additional_properties``
+    # because ``raw_representation`` is provider specific and does not survive a
+    # session round trip.
+    server_label = getattr(raw_item, "server_label", None)
+    additional_properties = {"server_label": server_label} if isinstance(server_label, str) and server_label else None
+    contents: list[Content] = [
+        Content.from_oauth_consent_request(
+            consent_link=consent_link,
+            additional_properties=additional_properties,
+            raw_representation=raw_item,
+        )
+    ]
+
     return ChatResponseUpdate(
         contents=contents,
         role="assistant",
diff --git a/python/packages/foundry/tests/foundry/test_foundry_agent.py b/python/packages/foundry/tests/foundry/test_foundry_agent.py
index 9ae2938ed62..8568ee4593a 100644
--- a/python/packages/foundry/tests/foundry/test_foundry_agent.py
+++ b/python/packages/foundry/tests/foundry/test_foundry_agent.py
@@ -1655,8 +1655,8 @@ def test_parse_chunk_surfaces_oauth_consent_request() -> None:
     assert update.raw_representation is mock_event
 
 
-def test_parse_chunk_skips_non_https_oauth_consent() -> None:
-    """An oauth_consent_request with a non-HTTPS link is rejected."""
+def test_parse_chunk_surfaces_non_https_oauth_consent() -> None:
+    """A non-HTTPS link is still surfaced so the host can fail the response."""
 
     mock_project = MagicMock()
     mock_project.get_openai_client.return_value = MagicMock()
@@ -1678,11 +1678,12 @@ def test_parse_chunk_skips_non_https_oauth_consent() -> None:
     update = client._parse_chunk_from_openai(mock_event, {}, {})
 
     consent_contents = [c for c in update.contents if c.type == "oauth_consent_request"]
-    assert len(consent_contents) == 0
+    assert len(consent_contents) == 1
+    assert consent_contents[0].consent_link == "http://insecure.example.com/login"
 
 
 def test_parse_chunk_handles_missing_consent_link() -> None:
-    """An oauth_consent_request without a consent_link produces no content."""
+    """A missing consent_link still surfaces the request, with an empty link."""
 
     mock_project = MagicMock()
     mock_project.get_openai_client.return_value = MagicMock()
@@ -1704,11 +1705,12 @@ def test_parse_chunk_handles_missing_consent_link() -> None:
     update = client._parse_chunk_from_openai(mock_event, {}, {})
 
     consent_contents = [c for c in update.contents if c.type == "oauth_consent_request"]
-    assert len(consent_contents) == 0
+    assert len(consent_contents) == 1
+    assert consent_contents[0].consent_link == ""
 
 
 def test_parse_chunk_handles_empty_string_consent_link() -> None:
-    """An oauth_consent_request with empty-string consent_link produces no content."""
+    """An empty-string consent_link still surfaces the request."""
 
     mock_project = MagicMock()
     mock_project.get_openai_client.return_value = MagicMock()
@@ -1730,7 +1732,8 @@ def test_parse_chunk_handles_empty_string_consent_link() -> None:
     update = client._parse_chunk_from_openai(mock_event, {}, {})
 
     consent_contents = [c for c in update.contents if c.type == "oauth_consent_request"]
-    assert len(consent_contents) == 0
+    assert len(consent_contents) == 1
+    assert consent_contents[0].consent_link == ""
 
 
 def test_parse_chunk_delegates_non_oauth_events_to_super() -> None:
diff --git a/python/packages/foundry/tests/foundry/test_foundry_chat_client.py b/python/packages/foundry/tests/foundry/test_foundry_chat_client.py
index db03045dbef..3ade3f94209 100644
--- a/python/packages/foundry/tests/foundry/test_foundry_chat_client.py
+++ b/python/packages/foundry/tests/foundry/test_foundry_chat_client.py
@@ -1453,8 +1453,8 @@ def test_parse_chunk_surfaces_oauth_consent_request() -> None:
     assert update.model == "test-model"
 
 
-def test_parse_chunk_skips_non_https_oauth_consent() -> None:
-    """An oauth_consent_request with a non-HTTPS link is rejected."""
+def test_parse_chunk_surfaces_non_https_oauth_consent() -> None:
+    """A non-HTTPS link is still surfaced so the host can fail the response."""
 
     mock_project = MagicMock()
     mock_openai = _make_mock_openai_client()
@@ -1477,11 +1477,12 @@ def test_parse_chunk_skips_non_https_oauth_consent() -> None:
     update = client._parse_chunk_from_openai(mock_event, {}, {})
 
     consent_contents = [c for c in update.contents if c.type == "oauth_consent_request"]
-    assert len(consent_contents) == 0
+    assert len(consent_contents) == 1
+    assert consent_contents[0].consent_link == "http://insecure.example.com/login"
 
 
 def test_parse_chunk_handles_missing_consent_link() -> None:
-    """An oauth_consent_request without a consent_link produces no content."""
+    """A missing consent_link still surfaces the request, with an empty link."""
 
     mock_project = MagicMock()
     mock_openai = _make_mock_openai_client()
@@ -1504,11 +1505,12 @@ def test_parse_chunk_handles_missing_consent_link() -> None:
     update = client._parse_chunk_from_openai(mock_event, {}, {})
 
     consent_contents = [c for c in update.contents if c.type == "oauth_consent_request"]
-    assert len(consent_contents) == 0
+    assert len(consent_contents) == 1
+    assert consent_contents[0].consent_link == ""
 
 
 def test_parse_chunk_handles_empty_string_consent_link() -> None:
-    """An oauth_consent_request with empty-string consent_link produces no content."""
+    """An empty-string consent_link still surfaces the request."""
 
     mock_project = MagicMock()
     mock_openai = _make_mock_openai_client()
@@ -1531,7 +1533,8 @@ def test_parse_chunk_handles_empty_string_consent_link() -> None:
     update = client._parse_chunk_from_openai(mock_event, {}, {})
 
     consent_contents = [c for c in update.contents if c.type == "oauth_consent_request"]
-    assert len(consent_contents) == 0
+    assert len(consent_contents) == 1
+    assert consent_contents[0].consent_link == ""
 
 
 def test_parse_chunk_delegates_non_oauth_events_to_super() -> None:
diff --git a/python/packages/foundry/tests/foundry/test_oauth_helpers.py b/python/packages/foundry/tests/foundry/test_oauth_helpers.py
index 8e98e9f94b7..70904af9f2b 100644
--- a/python/packages/foundry/tests/foundry/test_oauth_helpers.py
+++ b/python/packages/foundry/tests/foundry/test_oauth_helpers.py
@@ -198,59 +198,36 @@ def test_parses_top_level_consent_requested_event() -> None:
     assert consent[0].consent_link == "https://consent.example.com/authorize"
 
 
-def test_empty_contents_for_non_https_link(caplog: pytest.LogCaptureFixture) -> None:
-    """A non-HTTPS consent_link produces an update with empty contents and logs a warning."""
-    event = _make_output_item_event(consent_link="http://bad.example.com/login", item_id="item-http")
-    with caplog.at_level(logging.WARNING):
-        update = try_parse_oauth_consent_event(event, "test-model")
-
-    assert update is not None
-    assert len(update.contents) == 0
-    assert "non-HTTPS" in caplog.text
-
-
-def test_empty_contents_for_missing_consent_link(caplog: pytest.LogCaptureFixture) -> None:
-    """A None consent_link produces an update with empty contents and logs a warning."""
-    event = _make_output_item_event(consent_link=None, item_id="item-none")
-    with caplog.at_level(logging.WARNING):
-        update = try_parse_oauth_consent_event(event, "test-model")
-
-    assert update is not None
-    assert len(update.contents) == 0
-    assert "without valid consent_link" in caplog.text
-
-
-def test_empty_contents_for_empty_string_consent_link(caplog: pytest.LogCaptureFixture) -> None:
-    """An empty-string consent_link produces an update with empty contents and logs a warning."""
-    event = _make_output_item_event(consent_link="", item_id="item-empty")
-    with caplog.at_level(logging.WARNING):
-        update = try_parse_oauth_consent_event(event, "test-model")
-
-    assert update is not None
-    assert len(update.contents) == 0
-    assert "without valid consent_link" in caplog.text
-
-
-def test_empty_contents_for_https_empty_netloc(caplog: pytest.LogCaptureFixture) -> None:
-    """An HTTPS URL with empty netloc (https:///path) is rejected."""
-    event = _make_output_item_event(consent_link="https:///path", item_id="item-no-netloc")
-    with caplog.at_level(logging.WARNING):
-        update = try_parse_oauth_consent_event(event, "test-model")
-
-    assert update is not None
-    assert len(update.contents) == 0
-    assert "non-HTTPS" in caplog.text
-
+@pytest.mark.parametrize(
+    ("consent_link", "expected_link", "expected_log"),
+    [
+        pytest.param("http://bad.example.com/login", "http://bad.example.com/login", "non-HTTPS", id="non-https"),
+        pytest.param(None, "", "without valid consent_link", id="missing"),
+        pytest.param("", "", "without valid consent_link", id="empty-string"),
+        pytest.param("https:///path", "https:///path", "non-HTTPS", id="empty-netloc"),
+        pytest.param("https://[broken", "https://[broken", "malformed", id="malformed-authority"),
+    ],
+)
+def test_unusable_consent_link_is_still_surfaced(
+    consent_link: str | None,
+    expected_link: str,
+    expected_log: str,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """An unusable link is logged but still surfaced so the host can fail the response.
 
-def test_empty_contents_for_malformed_authority(caplog: pytest.LogCaptureFixture) -> None:
-    """A malformed authority is rejected instead of raising out of the parser."""
-    event = _make_output_item_event(consent_link="https://[broken", item_id="item-malformed")
+    Dropping the content here would leave the host with nothing to record, so a turn that
+    cannot proceed without consent would be reported as a silent success.
+    """
+    event = _make_output_item_event(consent_link=consent_link, item_id="item-bad")
     with caplog.at_level(logging.WARNING):
         update = try_parse_oauth_consent_event(event, "test-model")
 
     assert update is not None
-    assert len(update.contents) == 0
-    assert "malformed" in caplog.text
+    consent = [c for c in update.contents if c.type == "oauth_consent_request"]
+    assert len(consent) == 1
+    assert consent[0].consent_link == expected_link
+    assert expected_log in caplog.text
 
 
 def test_server_label_is_preserved_in_additional_properties() -> None:
diff --git a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
index e6652d03973..5131a814a89 100644
--- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
+++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
@@ -7,12 +7,10 @@
 import json
 import logging
 import os
-import re
 from collections.abc import AsyncIterable, AsyncIterator, Generator, Mapping, Sequence
 from contextlib import AbstractAsyncContextManager, AsyncExitStack
 from dataclasses import asdict, dataclass, field, is_dataclass
 from typing import Literal, cast
-from urllib.parse import urlparse
 
 from agent_framework import (
     ChatOptions,
@@ -27,6 +25,7 @@
     SupportsAgentRun,
     WorkflowAgent,
 )
+from agent_framework._oauth import validate_oauth_consent_link
 from agent_framework._telemetry import mark_feature_used
 from agent_framework.exceptions import AgentFrameworkException
 from azure.ai.agentserver.core import get_request_context
@@ -177,54 +176,13 @@ def consent_url_from_error(exc: BaseException) -> list[ConsentError] | None:
     return None
 
 
-# Characters allowed in a registered host name, and in the bracketed IPv6 form that
-# ``urlparse`` reports with its brackets already stripped.
-_HOST_PATTERN = re.compile(r"^[A-Za-z0-9._~%-]+$")
-_IPV6_HOST_PATTERN = re.compile(r"^[0-9A-Fa-f:.%-]+$")
-
-
-def _is_valid_consent_host(hostname: str) -> bool:
-    """Return whether *hostname* is syntactically usable by a standard URL client.
-
-    ``urlparse`` does not reject hosts that contain illegal characters, so values such as
-    ``exa mple.com`` are reported as a hostname even though no client can resolve them.
-    """
-    pattern = _IPV6_HOST_PATTERN if ":" in hostname else _HOST_PATTERN
-    return bool(pattern.match(hostname))
-
-
 def _validated_consent_link(consent_link: str | None) -> str | None:
     """Return *consent_link* when it is an absolute HTTPS URL a client can open, else ``None``.
 
-    A consent link is rendered as a clickable prompt by the client, so anything that is
-    not an absolute ``https`` URL is dropped rather than surfaced. ``urlparse`` raises
-    ``ValueError`` for malformed authorities (for example ``https://[broken``) and for
-    invalid ports, but only when ``port`` is read, so it is accessed here. A non-empty
-    ``netloc`` is not sufficient on its own (``https://@`` has one but no host), and a
-    non-empty ``hostname`` is not either (``https://exa mple.com`` reports one).
+    Thin wrapper over the shared core validator. The rules live in ``agent_framework._oauth``
+    so this host and the Foundry parser that produces the link cannot drift apart.
     """
-    if not consent_link:
-        return None
-    if any(char.isspace() or ord(char) < 0x20 or ord(char) == 0x7F for char in consent_link):
-        # ``urlparse`` silently strips tab and newline, which would let a link carrying
-        # control characters through even though it is not safe to render or log.
-        logger.warning("Skipping oauth_consent_request with whitespace in the consent_link.")
-        return None
-    try:
-        parsed = urlparse(consent_link)
-        hostname = parsed.hostname
-        # Reading ``port`` is what validates it; ``https://host:bad`` raises here.
-        _ = parsed.port
-    except ValueError:
-        logger.warning("Skipping oauth_consent_request with a malformed consent_link.")
-        return None
-    if parsed.scheme.lower() != "https" or not hostname:
-        logger.warning("Skipping oauth_consent_request with a non-HTTPS consent_link.")
-        return None
-    if not _is_valid_consent_host(hostname):
-        logger.warning("Skipping oauth_consent_request with an invalid consent_link host.")
-        return None
-    return consent_link
+    return validate_oauth_consent_link(consent_link)
 
 
 def _consent_link_from_content(content: Content) -> str | None:
@@ -613,19 +571,9 @@ async def _handle_inner_agent(
         elif save_failure is not None:
             for event in self._emit_failure(response_event_stream, tracker, save_failure):
                 yield event
-        elif consent_tracker.emitted:
-            # The turn cannot finish until the user completes OAuth consent, so the response
-            # ends as `incomplete` rather than `completed`, matching the connect-time path.
-            yield response_event_stream.emit_incomplete(reason=_consent_incomplete_reason(len(consent_tracker.emitted)))
-        elif consent_tracker.dropped:
-            # Consent was required but no link could be surfaced, so there is nothing for the
-            # user to act on. Failing is the honest outcome; `completed` would hide the block.
-            for event in self._emit_failure(
-                response_event_stream, tracker, _consent_unusable_link_error(len(consent_tracker.dropped))
-            ):
-                yield event
         else:
-            yield response_event_stream.emit_completed()
+            for event in self._finish_consent_response(response_event_stream, tracker, consent_tracker):
+                yield event
 
     async def _handle_inner_workflow(
         self,
@@ -748,24 +696,40 @@ async def _handle_inner_workflow(
             for event in tracker.close():
                 yield event
 
-            if consent_tracker.emitted:
-                yield response_event_stream.emit_incomplete(
-                    reason=_consent_incomplete_reason(len(consent_tracker.emitted))
-                )
-            elif consent_tracker.dropped:
-                # Consent was required but no link could be surfaced, so there is nothing for
-                # the user to act on. Failing is the honest outcome; `completed` would hide it.
-                for event in self._emit_failure(
-                    response_event_stream, tracker, _consent_unusable_link_error(len(consent_tracker.dropped))
-                ):
-                    yield event
-            else:
-                yield response_event_stream.emit_completed()
+            for event in self._finish_consent_response(response_event_stream, tracker, consent_tracker):
+                yield event
         except Exception as ex:
             logger.exception("Failed to produce response for workflow agent")
             for event in self._emit_failure(response_event_stream, tracker, ex):
                 yield event
 
+    @classmethod
+    def _finish_consent_response(
+        cls,
+        response_event_stream: ResponseEventStream,
+        tracker: _OutputItemTracker | None,
+        consent_tracker: _ConsentTracker,
+    ) -> Generator[ResponseStreamEvent]:
+        """Yield the terminal event for a turn that may have requested OAuth consent.
+
+        Shared by the agent and workflow paths so both report the same outcome for the
+        same consent state. Callers must apply any higher-precedence failure (a request
+        or session-persistence error) before delegating here.
+
+        The precedence is: a surfaced consent link ends the turn as ``incomplete``,
+        because it cannot finish until the user completes consent; a consent request whose
+        link was unusable ends it as ``failed``, because there is nothing for the user to
+        act on and ``completed`` would hide the block; otherwise the turn ``completed``.
+        """
+        if consent_tracker.emitted:
+            yield response_event_stream.emit_incomplete(reason=_consent_incomplete_reason(len(consent_tracker.emitted)))
+        elif consent_tracker.dropped:
+            yield from cls._emit_failure(
+                response_event_stream, tracker, _consent_unusable_link_error(len(consent_tracker.dropped))
+            )
+        else:
+            yield response_event_stream.emit_completed()
+
     @staticmethod
     def _emit_failure(
         response_event_stream: ResponseEventStream,

From 5d95e482676f29745bb23f3bc8c869611da8f8eb Mon Sep 17 00:00:00 2001
From: Giles Odigwe <gilesodigwe@microsoft.com>
Date: Wed, 26 Aug 2026 09:32:27 -0700
Subject: [PATCH 5/5] Python: report workflow OAuth consent turns as not
 resumable

A `WorkflowAgent` surfaces an OAuth consent request through
`AgentExecutor.ctx.request_info(...)`, which parks the workflow on a pending
`request_info` event. OAuth consent has no response content type, so nothing a
later turn sends can answer that request. Ending the turn `incomplete` promised
a continuation the workflow cannot honor, and the next turn on the same
conversation restored the parked checkpoint and failed deep in the workflow
machinery with `Unexpected content type while awaiting request info responses`.

Report the block directly instead:

- `_finish_consent_response` takes a `resumable` flag. The agent path stays
  `incomplete`, because a plain agent re-runs on the next turn and picks up the
  newly granted access. The workflow path ends `failed` with a message telling
  the user to grant consent and start a new conversation.
- `_pending_consent_links` reads the consent requests a restored checkpoint is
  parked on straight off `WorkflowCheckpoint.pending_request_info_events`. The
  restore-only run does not replay them as agent response updates, so they are
  not observable from the update stream.
- `_handle_inner_workflow` raises before starting the run when the restored
  checkpoint is parked on consent, letting the caller emit the single terminal
  failure event the same way an unresumable `previous_response_id` does.

The consent link is still surfaced in both cases; only the terminal status
changes. This does not make workflow consent resumable, which needs a matching
input contract in the core workflow layer.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 992057d8-78a8-45a7-9201-35af0919b071
---
 .../_responses.py                             | 78 +++++++++++++++++--
 .../foundry_hosting/tests/test_responses.py   | 52 +++++++++++--
 2 files changed, 117 insertions(+), 13 deletions(-)

diff --git a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
index 932d74c1077..065df1be735 100644
--- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
+++ b/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py
@@ -26,6 +26,7 @@
     SupportsAgentRun,
     UsageDetails,
     WorkflowAgent,
+    WorkflowCheckpoint,
     add_usage_details,
 )
 from agent_framework._oauth import validate_oauth_consent_link
@@ -375,6 +376,40 @@ def _consent_unusable_link_error(count: int) -> RuntimeError:
     )
 
 
+def _consent_not_resumable_error(count: int) -> RuntimeError:
+    """Return the error reported when a workflow turn is blocked on consent it cannot resume from.
+
+    OAuth consent has no response content type, so a caller cannot answer the pending request
+    an ``AgentExecutor`` records for it. Reporting ``incomplete`` would promise a continuation
+    that the workflow cannot honor, so the turn fails with the consent link already surfaced
+    and the user is directed to a new conversation.
+    """
+    return RuntimeError(
+        f"OAuth consent is required for {count} tool(s). The consent link has been provided, but this "
+        "workflow turn cannot be resumed after consent is granted because OAuth consent has no response "
+        "type. Grant consent using the link above, then start a new conversation."
+    )
+
+
+def _pending_consent_links(checkpoint: WorkflowCheckpoint | None) -> set[str]:
+    """Return the consent links a *checkpoint* is parked on, if any.
+
+    A workflow that stops on an OAuth consent request records it as a pending
+    ``request_info`` event. Consent has no response content type, so nothing a later turn
+    sends can satisfy that request and the workflow can never be resumed from it. Reading
+    the pending events off the checkpoint lets the host report the block itself; the
+    restore-only run does not replay these requests as updates.
+    """
+    if checkpoint is None:
+        return set()
+    links: set[str] = set()
+    for event in checkpoint.pending_request_info_events.values():
+        data = getattr(event, "data", None)
+        if isinstance(data, Content) and data.type == "oauth_consent_request":
+            links.add(data.consent_link or "")
+    return links
+
+
 @dataclass
 class _ConsentTracker:
     """Tracks the OAuth consent requests seen while converting content for one response.
@@ -646,7 +681,9 @@ async def _handle_response(
             for event in tracker.close():
                 yield event
 
-            for event in self._finish_consent_response(response_event_stream, tracker):
+            for event in self._finish_consent_response(
+                response_event_stream, tracker, resumable=not self._is_workflow_agent
+            ):
                 yield event
         except Exception as ex:
             logger.error("Failed to produce response for agent", exc_info=(type(ex), ex, ex.__traceback__))
@@ -906,6 +943,17 @@ async def _handle_inner_workflow(
                         if cancellation_signal.is_set():
                             return
 
+                # OAuth consent is the exception to the pending-request handling above: it has
+                # no response content type, so a pending consent request can never be fulfilled
+                # by anything this turn carries. Those requests are read straight off the
+                # checkpoint, because the restore-only call does not replay them as updates, and
+                # reported directly instead of letting the run below fail with an internal
+                # "unexpected content type" error. Raising lets the caller emit the single
+                # terminal failure event, the same way an unresumable previous_response_id does.
+                pending_consent_links = _pending_consent_links(latest_checkpoint)
+                if pending_consent_links:
+                    raise _consent_not_resumable_error(len(pending_consent_links))
+
                 # A cancel signal that fired after the restore-only replay finished (or was never
                 # entered) must still preempt starting a brand new workflow run below.
                 if cancellation_signal.is_set():
@@ -993,6 +1041,8 @@ def _finish_consent_response(
         cls,
         response_event_stream: ResponseEventStream,
         tracker: _OutputItemTracker,
+        *,
+        resumable: bool,
     ) -> Generator[ResponseStreamEvent]:
         """Yield the terminal event for a turn that may have requested OAuth consent.
 
@@ -1000,14 +1050,30 @@ def _finish_consent_response(
         same consent state. Callers must apply any higher-precedence failure (a request
         or session-persistence error) before delegating here.
 
-        The precedence is: a surfaced consent link ends the turn as ``incomplete``,
-        because it cannot finish until the user completes consent; a consent request whose
-        link was unusable ends it as ``failed``, because there is nothing for the user to
-        act on and ``completed`` would hide the block; otherwise the turn ``completed``.
+        The precedence is: a surfaced consent link ends the turn as ``incomplete`` when the
+        turn can be continued, and as ``failed`` when it cannot, so the status never promises
+        a continuation the caller will not get; a consent request whose link was unusable ends
+        the turn as ``failed``, because there is nothing for the user to act on and
+        ``completed`` would hide the block; otherwise the turn ``completed``.
+
+        Args:
+            response_event_stream: The stream the terminal event is emitted on.
+            tracker: The active output item tracker, drained before a failure event, and the
+                owner of the consent requests seen while converting this response.
+
+        Keyword Args:
+            resumable: Whether the turn can continue once consent is granted. True for a
+                plain agent, which re-runs and picks up the newly granted access. False for
+                a workflow, where the blocked turn is recorded as a pending request that has
+                no matching response type and therefore cannot be answered.
         """
         consent_tracker = tracker.consent
         if consent_tracker.emitted:
-            yield response_event_stream.emit_incomplete(reason=_consent_incomplete_reason(len(consent_tracker.emitted)))
+            count = len(consent_tracker.emitted)
+            if resumable:
+                yield response_event_stream.emit_incomplete(reason=_consent_incomplete_reason(count))
+            else:
+                yield from cls._emit_failure(response_event_stream, tracker, _consent_not_resumable_error(count))
         elif consent_tracker.dropped:
             yield from cls._emit_failure(
                 response_event_stream, tracker, _consent_unusable_link_error(len(consent_tracker.dropped))
diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py
index 09d296f7a46..1ac671bdaf7 100644
--- a/python/packages/foundry_hosting/tests/test_responses.py
+++ b/python/packages/foundry_hosting/tests/test_responses.py
@@ -4818,9 +4818,13 @@ async def test_basic_text_response_streaming(self) -> None:
         text_done = [e for e in events if e["event"] == "response.output_text.done"]
         assert any(e["data"]["text"] == "hello stream" for e in text_done)
 
-    async def test_mid_run_consent_emits_oauth_item_and_incomplete(self) -> None:
-        """A workflow that surfaces a consent request must emit the output item and end
-        ``incomplete``, after its checkpoint finalization, just like the regular agent path.
+    async def test_mid_run_consent_fails_the_workflow_response_as_not_resumable(self) -> None:
+        """A workflow that surfaces a consent request must emit the output item, then fail.
+
+        Unlike the regular agent path, a consent-blocked workflow turn cannot be continued:
+        the ``AgentExecutor`` records a pending request that no input can answer, because
+        OAuth consent has no response content type. Reporting ``incomplete`` would promise a
+        continuation the workflow cannot honor, so the link is surfaced and the turn fails.
         """
         workflow_agent = _build_contents_workflow_agent([
             Content.from_oauth_consent_request(
@@ -4835,7 +4839,9 @@ async def test_mid_run_consent_emits_oauth_item_and_incomplete(self) -> None:
         events = _parse_sse_events(resp.text)
         types = _sse_event_types(events)
 
-        assert types[-1] == "response.incomplete"
+        assert types[-1] == "response.failed"
+        assert "response.incomplete" not in types
+        assert "response.completed" not in types
 
         added = [e for e in events if e["event"] == "response.output_item.added"]
         oauth_added = [e for e in added if e["data"]["item"]["type"] == "oauth_consent_request"]
@@ -4848,9 +4854,41 @@ async def test_mid_run_consent_emits_oauth_item_and_incomplete(self) -> None:
 
         # A WorkflowAgent replays the inner agent's content as workflow output, so the same
         # consent request reaches the host twice and must not produce two consent prompts.
-        incomplete = [e for e in events if e["event"] == "response.incomplete"]
-        assert len(incomplete) == 1
-        assert "1 tool(s)" in json.dumps(incomplete[0]["data"])
+        failed = [e for e in events if e["event"] == "response.failed"]
+        assert len(failed) == 1
+        message = failed[0]["data"]["response"]["error"]["message"]
+        assert "1 tool(s)" in message
+        assert "cannot be resumed" in message
+
+    async def test_second_turn_after_consent_reports_a_clear_failure(self) -> None:
+        """A conversation parked on consent must fail legibly on every later turn.
+
+        The consent request is checkpointed as a pending ``request_info`` that no input can
+        answer, so restoring it and sending ordinary text would otherwise surface an internal
+        ``Unexpected content type while awaiting request info responses`` error. The restored
+        consent links are detected up front so the block is reported directly instead.
+        """
+        workflow_agent = _build_contents_workflow_agent([
+            Content.from_oauth_consent_request(
+                consent_link="https://consent.example.com/obo",
+                additional_properties={"server_label": "obo-mcp"},
+            )
+        ])
+        server = _make_server(workflow_agent)
+        conversation_id = "conv-consent-resume"
+
+        first = await _post(server, input_text="hi", stream=True, conversation_id=conversation_id)
+        assert _sse_event_types(_parse_sse_events(first.text))[-1] == "response.failed"
+
+        second = await _post(server, input_text="thanks, granted", stream=True, conversation_id=conversation_id)
+        events = _parse_sse_events(second.text)
+        types = _sse_event_types(events)
+
+        assert types[-1] == "response.failed"
+        assert "response.completed" not in types
+        message = [e for e in events if e["event"] == "response.failed"][0]["data"]["response"]["error"]["message"]
+        assert "cannot be resumed" in message
+        assert "Unexpected content type" not in message
 
     async def test_mid_run_unusable_consent_link_fails_the_workflow_response(self) -> None:
         workflow_agent = _build_contents_workflow_agent([