From 2727909a37e8d3b74303871f8e5cc69c0379c265 Mon Sep 17 00:00:00 2001 From: Shivani767 <101629653+Shivani767@users.noreply.github.com> Date: Thu, 20 Aug 2026 23:54:18 +0530 Subject: [PATCH 1/4] Python: Preserve structured output on streaming harness tool-approval path Forward response_format through ToolApprovalMiddleware's stream finalizer so AgentResponse.value is parsed the same way as the non-streaming path. --- .../_harness/_tool_approval.py | 27 +++- .../tests/core/test_harness_tool_approval.py | 135 ++++++++++++++++++ 2 files changed, 161 insertions(+), 1 deletion(-) diff --git a/python/packages/core/agent_framework/_harness/_tool_approval.py b/python/packages/core/agent_framework/_harness/_tool_approval.py index 4e2db951c77..bffbb342bea 100644 --- a/python/packages/core/agent_framework/_harness/_tool_approval.py +++ b/python/packages/core/agent_framework/_harness/_tool_approval.py @@ -7,6 +7,7 @@ import json from asyncio import sleep from collections.abc import AsyncIterable, Awaitable, Callable, Iterable, Mapping, MutableMapping, Sequence +from functools import partial from typing import Any, Literal, cast from .._middleware import AgentContext, AgentMiddleware @@ -77,6 +78,24 @@ def _contents_from_state(values: Any) -> list[Content]: return [_content_from_state(value) for value in state_items] +def _structured_response_format(context: AgentContext) -> Any | None: + """Return the structured-output schema for this invocation, if any. + + Run ``options`` take precedence over the agent's ``default_options``. The + streaming re-wrap in :meth:`ToolApprovalMiddleware._process_stream` must + forward this to ``AgentResponse.from_updates`` or ``response.value`` is + dropped even when the inner stream already parsed it. + """ + if context.options is not None: + response_format = context.options.get("response_format") + if response_format is not None: + return response_format + default_options = getattr(context.agent, "default_options", None) + if isinstance(default_options, Mapping): + return default_options.get("response_format") + return None + + def _content_to_state(content: Content) -> dict[str, Any]: return content.to_dict() @@ -509,7 +528,13 @@ async def _stream() -> AsyncIterable[AgentResponseUpdate]: context.messages = [] context.result = None - return ResponseStream(_stream(), finalizer=AgentResponse.from_updates) + return ResponseStream( + _stream(), + finalizer=partial( + AgentResponse.from_updates, + output_format_type=_structured_response_format(context), + ), + ) def _prepare_inbound_messages( self, diff --git a/python/packages/core/tests/core/test_harness_tool_approval.py b/python/packages/core/tests/core/test_harness_tool_approval.py index 1a96bd97088..937b2c22b44 100644 --- a/python/packages/core/tests/core/test_harness_tool_approval.py +++ b/python/packages/core/tests/core/test_harness_tool_approval.py @@ -3092,3 +3092,138 @@ def optional_args_tool(value: str = "default") -> str: requests = _approval_requests(second_response.messages) assert [_function_call(request).arguments for request in requests] == ['{"value": "custom"}'] assert calls == 1 + + +@pytest.mark.parametrize("via", ["run_options", "default_options"], ids=["run-options", "default-options"]) +async def test_streaming_tool_approval_preserves_structured_value( + chat_client_base: MockBaseChatClient, + via: str, +) -> None: + """Streaming ToolApprovalMiddleware must forward response_format to the outer finalizer. + + Regression for https://github.com/microsoft/agent-framework/issues/7418: + re-wrapping with ``AgentResponse.from_updates`` dropped ``output_format_type``, + so ``response.value`` was None even when the inner stream had parsed it. + """ + from pydantic import BaseModel + + class Answer(BaseModel): + answer: str + + json_text = '{"answer": "42"}' + + @tool(name="echo", approval_mode="always_require") + def echo(text: str) -> str: + return text + + default_options = {"response_format": Answer} if via == "default_options" else None + run_options = {"response_format": Answer} if via == "run_options" else None + agent = Agent( + client=chat_client_base, + tools=[echo], + middleware=[ToolApprovalMiddleware()], + default_options=default_options, # type: ignore[arg-type] + ) + session = AgentSession(session_id=f"structured-stream-{via}") + chat_client_base.streaming_responses = [ + [ + ChatResponseUpdate( + role="assistant", + contents=[Content.from_text(json_text)], + finish_reason="stop", + ) + ] + ] + + stream = agent.run("return an Answer", stream=True, session=session, options=run_options) + async for _update in stream: + pass + response = await stream.get_final_response() + + assert response.text == json_text + assert isinstance(response.value, Answer) + assert response.value.answer == "42" + + +async def test_streaming_auto_approved_tool_preserves_structured_value( + chat_client_base: MockBaseChatClient, +) -> None: + """Auto-approved tool calls must still parse structured output on the streaming path.""" + from pydantic import BaseModel + + class Answer(BaseModel): + answer: str + + json_text = '{"answer": "42"}' + calls = 0 + + @tool(name="echo", approval_mode="always_require") + def echo(text: str) -> str: + nonlocal calls + calls += 1 + return text + + agent = Agent( + client=chat_client_base, + tools=[echo], + middleware=[ToolApprovalMiddleware(auto_approval_rules=[lambda function_call: True])], + ) + session = AgentSession(session_id="structured-stream-auto-approve") + function_call = Content.from_function_call(call_id="call_echo", name="echo", arguments='{"text": "hi"}') + chat_client_base.streaming_responses = [ + [ChatResponseUpdate(role="assistant", contents=[function_call])], + [ + ChatResponseUpdate( + role="assistant", + contents=[Content.from_text(json_text)], + finish_reason="stop", + ) + ], + ] + + stream = agent.run( + "Call echo and return an Answer.", + stream=True, + session=session, + options={"response_format": Answer}, + ) + async for _update in stream: + pass + response = await stream.get_final_response() + + assert calls == 1 + assert isinstance(response.value, Answer) + assert response.value.answer == "42" + + +async def test_non_streaming_tool_approval_preserves_structured_value( + chat_client_base: MockBaseChatClient, +) -> None: + """Non-streaming ToolApprovalMiddleware already returns the inner AgentResponse.""" + from pydantic import BaseModel + + class Answer(BaseModel): + answer: str + + json_text = '{"answer": "42"}' + + @tool(name="echo", approval_mode="always_require") + def echo(text: str) -> str: + return text + + agent = Agent( + client=chat_client_base, + tools=[echo], + middleware=[ToolApprovalMiddleware()], + ) + session = AgentSession(session_id="structured-non-stream") + chat_client_base.run_responses = [ChatResponse(messages=Message(role="assistant", contents=[json_text]))] + + response = await agent.run( + "return an Answer", + session=session, + options={"response_format": Answer}, + ) + + assert isinstance(response.value, Answer) + assert response.value.answer == "42" From c4ac2987c17753f0986bda5f48c72f83c73aeda1 Mon Sep 17 00:00:00 2001 From: Shivani767 <101629653+Shivani767@users.noreply.github.com> Date: Fri, 21 Aug 2026 00:08:23 +0530 Subject: [PATCH 2/4] Python: Keep terminal structured value after auto-approved tool preamble Return the last inner AgentResponse from the streaming tool-approval finalizer so coalesced preamble text cannot clobber response.value. --- .../_harness/_tool_approval.py | 30 ++++++--- .../tests/core/test_harness_tool_approval.py | 61 ++++++++++++++++++- 2 files changed, 81 insertions(+), 10 deletions(-) diff --git a/python/packages/core/agent_framework/_harness/_tool_approval.py b/python/packages/core/agent_framework/_harness/_tool_approval.py index bffbb342bea..abe9c4bd501 100644 --- a/python/packages/core/agent_framework/_harness/_tool_approval.py +++ b/python/packages/core/agent_framework/_harness/_tool_approval.py @@ -7,7 +7,6 @@ import json from asyncio import sleep from collections.abc import AsyncIterable, Awaitable, Callable, Iterable, Mapping, MutableMapping, Sequence -from functools import partial from typing import Any, Literal, cast from .._middleware import AgentContext, AgentMiddleware @@ -461,6 +460,12 @@ def _process_stream( call_next: Callable[[], Awaitable[None]], state: ToolApprovalState, ) -> ResponseStream[AgentResponseUpdate, AgentResponse]: + # Last inner AgentResponse. Its already-parsed structured value is carried + # over by the outer finalizer so an auto-approved preamble that + # AgentResponse.from_updates coalesces into the JSON message cannot mask it. + holder: dict[str, AgentResponse | None] = {"final": None} + response_format = _structured_response_format(context) + async def _stream() -> AsyncIterable[AgentResponseUpdate]: if context.session is None: raise RuntimeError("ToolApprovalMiddleware requires an AgentSession.") @@ -496,7 +501,7 @@ async def _stream() -> AsyncIterable[AgentResponseUpdate]: buffered_update = copy.copy(update) buffered_update.contents = list(update.contents) buffered_approval_updates.append(buffered_update) - await context.result.get_final_response() + holder["final"] = await context.result.get_final_response() if not approval_requests: return @@ -528,13 +533,20 @@ async def _stream() -> AsyncIterable[AgentResponseUpdate]: context.messages = [] context.result = None - return ResponseStream( - _stream(), - finalizer=partial( - AgentResponse.from_updates, - output_format_type=_structured_response_format(context), - ), - ) + def _finalize(updates: Sequence[AgentResponseUpdate]) -> AgentResponse: + # Build the response from the streamed updates so the middleware's + # approval / user-input handling is preserved, then carry over the + # structured value already parsed by the inner response. The coalesced + # update text can include preamble from auto-approved turns, which would + # otherwise mask the parsed value (#7418). + response = AgentResponse.from_updates(updates, output_format_type=response_format) + final = holder["final"] + if final is not None and final._value_parsed: # pyright: ignore[reportPrivateUsage] + response._value = final._value # pyright: ignore[reportPrivateUsage] + response._value_parsed = True # pyright: ignore[reportPrivateUsage] + return response + + return ResponseStream(_stream(), finalizer=_finalize) def _prepare_inbound_messages( self, diff --git a/python/packages/core/tests/core/test_harness_tool_approval.py b/python/packages/core/tests/core/test_harness_tool_approval.py index 937b2c22b44..1ff1655c1c0 100644 --- a/python/packages/core/tests/core/test_harness_tool_approval.py +++ b/python/packages/core/tests/core/test_harness_tool_approval.py @@ -3122,7 +3122,7 @@ def echo(text: str) -> str: client=chat_client_base, tools=[echo], middleware=[ToolApprovalMiddleware()], - default_options=default_options, # type: ignore[arg-type] + default_options=default_options, # type: ignore[arg-type, typeddict-item] # pyrefly: ignore[bad-argument-type] # ty: ignore[invalid-argument-type] ) session = AgentSession(session_id=f"structured-stream-{via}") chat_client_base.streaming_responses = [ @@ -3196,6 +3196,65 @@ def echo(text: str) -> str: assert response.value.answer == "42" +async def test_streaming_auto_approved_tool_preserves_value_when_preamble_text_coalesces( + chat_client_base: MockBaseChatClient, +) -> None: + """Preamble assistant text plus a tool call must not clobber the parsed structured value. + + Updates without ``message_id`` are coalesced by ``AgentResponse.from_updates``. If the + outer finalizer rebuilt from every yielded update, ``Calling echo…{"answer":"42"}`` + would fail to parse. The terminal inner response's value must be kept instead. + """ + from pydantic import BaseModel + + class Answer(BaseModel): + answer: str + + json_text = '{"answer": "42"}' + calls = 0 + + @tool(name="echo", approval_mode="always_require") + def echo(text: str) -> str: + nonlocal calls + calls += 1 + return text + + agent = Agent( + client=chat_client_base, + tools=[echo], + middleware=[ToolApprovalMiddleware(auto_approval_rules=[lambda function_call: True])], + ) + session = AgentSession(session_id="structured-stream-auto-approve-preamble") + function_call = Content.from_function_call(call_id="call_echo", name="echo", arguments='{"text": "hi"}') + chat_client_base.streaming_responses = [ + [ + ChatResponseUpdate(role="assistant", contents=[Content.from_text("Calling echo…")]), + ChatResponseUpdate(role="assistant", contents=[function_call]), + ], + [ + ChatResponseUpdate( + role="assistant", + contents=[Content.from_text(json_text)], + finish_reason="stop", + ) + ], + ] + + stream = agent.run( + "Call echo and return an Answer.", + stream=True, + session=session, + options={"response_format": Answer}, + ) + updates = [update async for update in stream] + response = await stream.get_final_response() + + assert any("Calling echo" in (content.text or "") for update in updates for content in update.contents) + assert calls == 1 + assert isinstance(response.value, Answer) + assert response.value.answer == "42" + + async def test_non_streaming_tool_approval_preserves_structured_value( chat_client_base: MockBaseChatClient, ) -> None: From 4a8f19b8d71adecc1f1c9cd538e2821429ffcf7e Mon Sep 17 00:00:00 2001 From: "Shivani ." Date: Mon, 21 Sep 2026 17:08:07 +0530 Subject: [PATCH 3/4] Potential fix for pull request finding 'Preserve terminal JSON value before checking lazy parse state' Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- .../core/agent_framework/_harness/_tool_approval.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/python/packages/core/agent_framework/_harness/_tool_approval.py b/python/packages/core/agent_framework/_harness/_tool_approval.py index abe9c4bd501..407576b3031 100644 --- a/python/packages/core/agent_framework/_harness/_tool_approval.py +++ b/python/packages/core/agent_framework/_harness/_tool_approval.py @@ -541,9 +541,11 @@ def _finalize(updates: Sequence[AgentResponseUpdate]) -> AgentResponse: # otherwise mask the parsed value (#7418). response = AgentResponse.from_updates(updates, output_format_type=response_format) final = holder["final"] - if final is not None and final._value_parsed: # pyright: ignore[reportPrivateUsage] - response._value = final._value # pyright: ignore[reportPrivateUsage] - response._value_parsed = True # pyright: ignore[reportPrivateUsage] + if final is not None: + value = final.value + if final._value_parsed: # pyright: ignore[reportPrivateUsage] + response._value = value # pyright: ignore[reportPrivateUsage] + response._value_parsed = True # pyright: ignore[reportPrivateUsage] return response return ResponseStream(_stream(), finalizer=_finalize) From f2e874f481c273c5dc564eb0539db73886238c7b Mon Sep 17 00:00:00 2001 From: Shivani Bhandari Date: Mon, 21 Sep 2026 17:11:56 +0530 Subject: [PATCH 4/4] Python: Address review on streaming structured-output finalizer Follow-up to #8548 review feedback: - Forward the terminal inner response's structured value through the public AgentResponse.from_updates(value=...) contract instead of mutating the private _value/_value_parsed fields. Those fields were only copied when the inner response had already parsed its value, so the normal lazy-parse case still reparsed the coalesced preamble and JSON as one invalid string. - Resolve final.value eagerly (which completes the lazy parse) and suppress the expected parse failures (ValueError/TypeError) so a parse error still surfaces lazily on value access, matching a middleware-free streaming run. - Cast default_options to Mapping[str, Any] in _structured_response_format so strict pyright no longer reports reportUnknownMemberType and reportUnknownVariableType (Package Checks). - Keep the parametrized regression test's option dicts out of Agent's generic client/options inference so mypy, pyrefly, ty, zuban, and pyright all pass (Test Typing Checks). --- .../_harness/_tool_approval.py | 36 ++++++++----- .../tests/core/test_harness_tool_approval.py | 51 +++++++++++++++++-- 2 files changed, 70 insertions(+), 17 deletions(-) diff --git a/python/packages/core/agent_framework/_harness/_tool_approval.py b/python/packages/core/agent_framework/_harness/_tool_approval.py index 407576b3031..6e33b205149 100644 --- a/python/packages/core/agent_framework/_harness/_tool_approval.py +++ b/python/packages/core/agent_framework/_harness/_tool_approval.py @@ -2,6 +2,7 @@ from __future__ import annotations +import contextlib import copy import inspect import json @@ -91,7 +92,8 @@ def _structured_response_format(context: AgentContext) -> Any | None: return response_format default_options = getattr(context.agent, "default_options", None) if isinstance(default_options, Mapping): - return default_options.get("response_format") + typed_default_options = cast("Mapping[str, Any]", default_options) + return typed_default_options.get("response_format") return None @@ -460,9 +462,11 @@ def _process_stream( call_next: Callable[[], Awaitable[None]], state: ToolApprovalState, ) -> ResponseStream[AgentResponseUpdate, AgentResponse]: - # Last inner AgentResponse. Its already-parsed structured value is carried - # over by the outer finalizer so an auto-approved preamble that - # AgentResponse.from_updates coalesces into the JSON message cannot mask it. + # Last inner AgentResponse. The outer finalizer resolves its structured + # value (completing its lazy parse) and forwards it through the public + # ``from_updates(value=...)`` contract, so an auto-approved preamble that + # the outer aggregation coalesces into the JSON message cannot mask it + # (#7418). holder: dict[str, AgentResponse | None] = {"final": None} response_format = _structured_response_format(context) @@ -535,18 +539,22 @@ async def _stream() -> AsyncIterable[AgentResponseUpdate]: def _finalize(updates: Sequence[AgentResponseUpdate]) -> AgentResponse: # Build the response from the streamed updates so the middleware's - # approval / user-input handling is preserved, then carry over the - # structured value already parsed by the inner response. The coalesced - # update text can include preamble from auto-approved turns, which would - # otherwise mask the parsed value (#7418). - response = AgentResponse.from_updates(updates, output_format_type=response_format) + # approval / user-input handling is preserved. The coalesced update text + # can include preamble from auto-approved turns, which would otherwise be + # parsed as one invalid string, so resolve the terminal inner response's + # structured value — reading ``value`` also completes its lazy parse — and + # forward it via ``from_updates``' public ``value`` argument (#7418). A + # parse failure is left unset so the outer response still surfaces the + # error lazily on ``value`` access, matching the error timing of a + # middleware-free streaming run. final = holder["final"] + value: Any = None if final is not None: - value = final.value - if final._value_parsed: # pyright: ignore[reportPrivateUsage] - response._value = value # pyright: ignore[reportPrivateUsage] - response._value_parsed = True # pyright: ignore[reportPrivateUsage] - return response + # ``ValidationError`` subclasses ``ValueError``; ``TypeError`` covers a + # malformed ``response_format`` in the terminal inner response. + with contextlib.suppress(ValueError, TypeError): + value = final.value + return AgentResponse.from_updates(updates, output_format_type=response_format, value=value) return ResponseStream(_stream(), finalizer=_finalize) diff --git a/python/packages/core/tests/core/test_harness_tool_approval.py b/python/packages/core/tests/core/test_harness_tool_approval.py index 1ff1655c1c0..c9a2a604bb2 100644 --- a/python/packages/core/tests/core/test_harness_tool_approval.py +++ b/python/packages/core/tests/core/test_harness_tool_approval.py @@ -3116,13 +3116,16 @@ class Answer(BaseModel): def echo(text: str) -> str: return text - default_options = {"response_format": Answer} if via == "default_options" else None - run_options = {"response_format": Answer} if via == "run_options" else None + # ``Any`` keeps the parametrized options out of ``Agent``'s generic client/options + # inference: the plain response-format dicts cannot be unified with the fixture's + # ``MockBaseChatClient[ChatOptions[None]]`` client type. + default_options: Any = {"response_format": Answer} if via == "default_options" else None + run_options: Any = {"response_format": Answer} if via == "run_options" else None agent = Agent( client=chat_client_base, tools=[echo], middleware=[ToolApprovalMiddleware()], - default_options=default_options, # type: ignore[arg-type, typeddict-item] # pyrefly: ignore[bad-argument-type] # ty: ignore[invalid-argument-type] + default_options=default_options, ) session = AgentSession(session_id=f"structured-stream-{via}") chat_client_base.streaming_responses = [ @@ -3286,3 +3289,45 @@ def echo(text: str) -> str: assert isinstance(response.value, Answer) assert response.value.answer == "42" + + +async def test_streaming_tool_approval_defers_structured_parse_error_to_value_access( + chat_client_base: MockBaseChatClient, +) -> None: + """A structured-output parse failure must surface on ``value`` access, not while streaming. + + The outer finalizer resolves the terminal inner response's value eagerly so a coalesced + preamble cannot mask it (#7418). When that resolution fails, the value is left unset so + the outer response still parses lazily on ``value`` access — streaming iteration and + finalization succeed, matching the error timing of a middleware-free streaming run. + """ + from pydantic import BaseModel + + class Answer(BaseModel): + answer: str + + bad_text = "not json" + + agent = Agent( + client=chat_client_base, + middleware=[ToolApprovalMiddleware()], + ) + session = AgentSession(session_id="structured-stream-parse-error") + chat_client_base.streaming_responses = [ + [ + ChatResponseUpdate( + role="assistant", + contents=[Content.from_text(bad_text)], + finish_reason="stop", + ) + ] + ] + + stream = agent.run("return an Answer", stream=True, session=session, options={"response_format": Answer}) + async for _ in stream: + pass # must not raise + response = await stream.get_final_response() + + assert response.text == bad_text + with pytest.raises(ValueError): + _ = response.value