From b452dd73d24806070d97cbbae3db97a7baa95f2f Mon Sep 17 00:00:00 2001 From: Giles Odigwe Date: Thu, 13 Aug 2026 14:26:22 -0700 Subject: [PATCH 1/2] 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 6ff4550f7e..8206c34e6b 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 5b0dfda65c..103f0a61e7 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 Date: Fri, 14 Aug 2026 12:12:29 -0700 Subject: [PATCH 2/2] 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 873d42c3d9..c8d70fe97f 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 2ab209e141..46087e595d 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 8206c34e6b..33d5f03884 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 103f0a61e7..aa200e6cac 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)