Python: surface mid-run oauth_consent_request items from ResponsesHostServer

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,
             )
         )

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

python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py

@@ -11,6 +11,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,
@@ -175,6 +176,81 @@ 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 fails :func:`_validated_consent_link`.
+    """
+    if content.type != "oauth_consent_request":
+        return None
+    if not content.consent_link:
+        logger.warning("Received oauth_consent_request content without a consent_link; skipping.")
+        return None
+    return _validated_consent_link(content.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.
+
+    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"
+
+
 # endregion Foundry Toolbox Auth integration
 
 
@@ -352,29 +428,32 @@ 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)
-                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 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_link,
+                    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 +485,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()
 
         try:
             if self._uses_hosted_responses_history:
@@ -438,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
@@ -480,6 +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 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(len(emitted_consent_requests))
+            )
         else:
             yield response_event_stream.emit_completed()
 
@@ -579,6 +666,7 @@ async def _handle_inner_workflow(
                     pass
 
             tracker = _OutputItemTracker(response_event_stream)
+            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(
@@ -591,15 +679,24 @@ async def _handle_inner_workflow(
                         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
 
             # Close any remaining active builder
             for event in tracker.close():
                 yield event
-            yield response_event_stream.emit_completed()
+
+            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:
             logger.exception("Failed to produce response for workflow agent")
             for event in self._emit_failure(response_event_stream, tracker, ex):
@@ -1604,13 +1701,16 @@ 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.
 
     Args:
         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.
@@ -1719,6 +1819,27 @@ 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:
+            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.")