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 17b73b6043..51ee14edfb 100644 --- a/python/packages/foundry_hosting/agent_framework_foundry_hosting/_responses.py +++ b/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,7 +679,10 @@ 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 @@ -599,7 +690,13 @@ async def _handle_inner_workflow( # 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,6 +1701,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. @@ -1611,6 +1709,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. @@ -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.") diff --git a/python/packages/foundry_hosting/tests/test_responses.py b/python/packages/foundry_hosting/tests/test_responses.py index b12497e530..6b289874f6 100644 --- a/python/packages/foundry_hosting/tests/test_responses.py +++ b/python/packages/foundry_hosting/tests/test_responses.py @@ -15,6 +15,7 @@ import uuid from collections.abc import AsyncGenerator, AsyncIterator, Awaitable, Callable, Mapping, Sequence from dataclasses import dataclass +from types import SimpleNamespace from typing import Literal, cast, overload from unittest.mock import AsyncMock, MagicMock, patch @@ -3748,6 +3749,191 @@ 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 + assert "2 tool(s)" in json.dumps(incomplete[0]["data"]) + + 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=[ + 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 + # 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 # region Error handling (response.failed surfacing) @@ -4043,13 +4229,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 @@ -4091,19 +4282,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: @@ -4178,6 +4369,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)