Python: Fix hosted MCP tool approval flow for all session/streaming combinations (#4054)

* fix openai hosted mcp samples * addressed copilot comments * Update python/samples/02-agents/providers/azure_openai/azure_responses_client_with_hosted_mcp.py Co-authored-by: Dmytro Struk <13853051+dmytrostruk@users.noreply.github.com> --------- Co-authored-by: Dmytro Struk <13853051+dmytrostruk@users.noreply.github.com>
2026-06-16 21:04:09 +08:00 · 2026-02-18 15:28:11 -08:00
parent 988ef6a50e
commit 21769e2cd1
6 changed files with 242 additions and 12 deletions
@@ -266,6 +266,45 @@ async def test_chat_client_agent_update_session_id_streaming_does_not_use_respon
    assert session.service_session_id is None


+async def test_chat_client_agent_streaming_session_id_set_without_get_final_response(
+    chat_client_base: SupportsChatGetResponse,
+) -> None:
+    """Test that session.service_session_id is set during streaming iteration.
+
+    This verifies the eager propagation of conversation_id via transform hook,
+    which is needed for multi-turn flows (e.g. hosted MCP approval) where the
+    user iterates the stream and then makes a follow-up call without calling
+    get_final_response().
+    """
+    chat_client_base.streaming_responses = [
+        [
+            ChatResponseUpdate(
+                contents=[Content.from_text("part 1")],
+                role="assistant",
+                response_id="resp_123",
+                conversation_id="resp_123",
+            ),
+            ChatResponseUpdate(
+                contents=[Content.from_text(" part 2")],
+                role="assistant",
+                response_id="resp_123",
+                conversation_id="resp_123",
+                finish_reason="stop",
+            ),
+        ]
+    ]
+
+    agent = Agent(client=chat_client_base)
+    session = agent.create_session()
+    assert session.service_session_id is None
+
+    # Only iterate — do NOT call get_final_response()
+    async for _ in agent.run("Hello", session=session, stream=True):
+        pass
+
+    assert session.service_session_id == "resp_123"
+
+
 async def test_chat_client_agent_update_session_messages(client: SupportsChatGetResponse) -> None:
    from agent_framework._sessions import InMemoryHistoryProvider

@@ -1255,6 +1255,152 @@ async def test_hosted_tool_approval_response(chat_client_base: SupportsChatGetRe
    assert response is not None


+async def test_hosted_mcp_approval_response_passthrough(chat_client_base: SupportsChatGetResponse):
+    """Test that hosted MCP approval responses pass through without local execution.
+
+    When an MCP approval response has server_label in function_call.additional_properties,
+    the function invocation layer must not intercept it. The approval request/response
+    should be forwarded to the API as-is so the service can execute the hosted tool.
+    """
+
+    @tool(name="local_function")
+    def local_func(arg1: str) -> str:
+        return f"Local {arg1}"
+
+    # Simulate an MCP approval request from the service (has server_label)
+    mcp_function_call = Content.from_function_call(
+        call_id="mcpr_abc123",
+        name="microsoft_docs_search",
+        arguments='{"query": "azure storage"}',
+        additional_properties={"server_label": "Microsoft_Learn_MCP"},
+    )
+    mcp_approval_request = Content.from_function_approval_request(
+        id="mcpr_abc123",
+        function_call=mcp_function_call,
+    )
+    mcp_approval_response = mcp_approval_request.to_function_approval_response(approved=True)
+
+    # The second call (after approval) should return a final response
+    chat_client_base.run_responses = [
+        ChatResponse(messages=Message(role="assistant", text="Here are the docs results.")),
+    ]
+
+    # Build message list mimicking handle_approvals_without_session:
+    # [original query, assistant with approval_request, user with approval_response]
+    messages = [
+        Message(role="user", text="Search docs for azure storage"),
+        Message(role="assistant", contents=[mcp_approval_request]),
+        Message(role="user", contents=[mcp_approval_response]),
+    ]
+
+    response = await chat_client_base.get_response(
+        messages,
+        tool_choice="auto",
+        tools=[local_func],
+    )
+
+    # The response should succeed without errors
+    assert response is not None
+    assert response.messages[0].text == "Here are the docs results."
+
+    # The approval contents should NOT have been mutated by the function invocation layer.
+    # The assistant message should still have the original approval_request content.
+    assistant_msg = messages[1]
+    assert assistant_msg.contents[0].type == "function_approval_request"
+    # The user message should still have the original approval_response content.
+    user_msg = messages[2]
+    assert user_msg.contents[0].type == "function_approval_response"
+
+
+def test_is_hosted_tool_approval_with_server_label():
+    """Test that _is_hosted_tool_approval returns True for MCP approvals with server_label."""
+    from agent_framework._tools import _is_hosted_tool_approval
+
+    mcp_fc = Content.from_function_call(
+        call_id="mcpr_abc",
+        name="docs_search",
+        arguments="{}",
+        additional_properties={"server_label": "Microsoft_Learn_MCP"},
+    )
+    mcp_request = Content.from_function_approval_request(id="mcpr_abc", function_call=mcp_fc)
+    mcp_response = mcp_request.to_function_approval_response(approved=True)
+
+    assert _is_hosted_tool_approval(mcp_request) is True
+    assert _is_hosted_tool_approval(mcp_response) is True
+
+
+def test_is_hosted_tool_approval_without_server_label():
+    """Test that _is_hosted_tool_approval returns False for regular tool approvals."""
+    from agent_framework._tools import _is_hosted_tool_approval
+
+    regular_fc = Content.from_function_call(call_id="call_1", name="my_func", arguments="{}")
+    regular_request = Content.from_function_approval_request(id="call_1", function_call=regular_fc)
+    regular_response = regular_request.to_function_approval_response(approved=True)
+
+    assert _is_hosted_tool_approval(regular_request) is False
+    assert _is_hosted_tool_approval(regular_response) is False
+    # Also test with None/non-content objects
+    assert _is_hosted_tool_approval(None) is False
+    assert _is_hosted_tool_approval("not a content") is False
+
+
+async def test_mixed_local_and_hosted_approval_flow(chat_client_base: SupportsChatGetResponse):
+    """Test that mixed local + hosted MCP approvals are handled correctly.
+
+    When a response contains both a local tool approval and a hosted MCP approval,
+    the local approval should be processed normally while the hosted MCP approval
+    should pass through untouched to the API.
+    """
+
+    @tool(name="local_function", approval_mode="always_require")
+    def local_func(arg1: str) -> str:
+        return f"Local {arg1}"
+
+    # Simulate the LLM returning both a local function call and an MCP approval request
+    local_fc = Content.from_function_call(call_id="call_local", name="local_function", arguments='{"arg1": "test"}')
+    mcp_fc = Content.from_function_call(
+        call_id="mcpr_hosted",
+        name="microsoft_docs_search",
+        arguments='{"query": "azure"}',
+        additional_properties={"server_label": "Microsoft_Learn_MCP"},
+    )
+    mcp_approval_request = Content.from_function_approval_request(id="mcpr_hosted", function_call=mcp_fc)
+
+    # First response: LLM returns a local function call that needs approval
+    chat_client_base.run_responses = [
+        ChatResponse(messages=Message(role="assistant", contents=[local_fc])),
+        # After local approval + hosted approval, the final response
+        ChatResponse(messages=Message(role="assistant", text="Done with both tools.")),
+    ]
+
+    # User approves the local function call
+    local_approval_response = Content.from_function_approval_response(
+        approved=True, id="call_local", function_call=local_fc
+    )
+    # User also has an MCP approval response (hosted)
+    mcp_approval_response = mcp_approval_request.to_function_approval_response(approved=True)
+
+    messages = [
+        Message(role="user", text="Search docs and run local"),
+        Message(role="assistant", contents=[local_fc, mcp_approval_request]),
+        Message(role="user", contents=[local_approval_response]),
+        Message(role="user", contents=[mcp_approval_response]),
+    ]
+
+    response = await chat_client_base.get_response(
+        messages,
+        tool_choice="auto",
+        tools=[local_func],
+    )
+
+    assert response is not None
+    # The hosted MCP approval contents should NOT have been mutated
+    assistant_msg = messages[1]
+    assert assistant_msg.contents[1].type == "function_approval_request"
+    mcp_user_msg = messages[3]
+    assert mcp_user_msg.contents[0].type == "function_approval_response"
+
+
 async def test_unapproved_tool_execution_raises_exception(chat_client_base: SupportsChatGetResponse):
    """Test that attempting to execute an unapproved tool raises ToolException."""