Python: Preserve structured output on streaming harness tool-approval path

python/packages/core/agent_framework/_harness/_tool_approval.py

@@ -79,6 +79,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()
 
@@ -438,6 +456,13 @@ def _process_stream(
         call_next: Callable[[], Awaitable[None]],
         state: ToolApprovalState,
     ) -> ResponseStream[AgentResponseUpdate, AgentResponse]:
+        # Last inner AgentResponse. Returning it from the outer finalizer matches
+        # the non-streaming path and preserves an already-parsed structured value
+        # even when earlier auto-approved turns yielded assistant text that
+        # AgentResponse.from_updates would coalesce into the JSON message.
+        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.")
@@ -484,7 +509,7 @@ async def _stream() -> AsyncIterable[AgentResponseUpdate]:
                             additional_properties=update.additional_properties,
                             raw_representation=update.raw_representation,
                         )
-                await context.result.get_final_response()
+                holder["final"] = await context.result.get_final_response()
                 if not approval_requests:
                     return
 
@@ -499,7 +524,12 @@ async def _stream() -> AsyncIterable[AgentResponseUpdate]:
                 context.messages = []
                 context.result = None
 
-        return ResponseStream(_stream(), finalizer=AgentResponse.from_updates)
+        def _finalize(updates: Sequence[AgentResponseUpdate]) -> AgentResponse:
+            if holder["final"] is not None:
+                return holder["final"]
+            return AgentResponse.from_updates(updates, output_format_type=response_format)
+
+        return ResponseStream(_stream(), finalizer=_finalize)
 
     def _prepare_inbound_messages(
         self,

python/packages/core/tests/core/test_harness_tool_approval.py

@@ -1426,3 +1426,197 @@ 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, 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 = [
+        [
+            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_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:
+    """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"