Python: validate functional workflow HITL response types

python/packages/core/agent_framework/_workflows/_functional.py

@@ -65,7 +65,7 @@
     WorkflowRunState,
     _framework_event,
 )
-from ._workflow import WorkflowRunResult
+from ._workflow import WorkflowRunResult, _coerce_request_info_response
 
 logger = logging.getLogger(__name__)
 
@@ -244,6 +244,9 @@ async def request_info(
         found, value = self._get_response(rid)
         if found:
             self._pending_requests.pop(rid, None)
+            # Functional workflows intentionally allow None responses; _set_responses logs a warning for them.
+            if value is not None:
+                value = _coerce_request_info_response(value, response_type, rid)
             return value
 
         # No response — emit event and interrupt

python/packages/core/agent_framework/_workflows/_workflow.py

@@ -58,6 +58,18 @@
 _MISSING: Any = object()
 
 
+def _coerce_request_info_response(value: Any, response_type: type, request_id: str) -> Any:
+    """Convert and validate a response supplied for a pending request."""
+    if response_type is Content and isinstance(value, str):
+        value = Content.from_text(text=value)
+    value = try_coerce_to_type(value, response_type)
+    if not is_instance_of(value, response_type):
+        raise ValueError(
+            f"Response type mismatch for request ID {request_id}: expected {response_type}, got {type(value)}"
+        )
+    return value
+
+
 def _coalesce_renamed_kwarg(old_name: str, old_value: Any, new_name: str, new_value: Any) -> Any:
     """Resolve a renamed keyword argument while keeping the deprecated name working.
 
@@ -1081,15 +1093,7 @@ async def _send_responses_internal(self, responses: Mapping[str, Any]) -> None:
             if request_id not in pending_requests:
                 raise ValueError(f"Response provided for unknown request ID: {request_id}")
             pending_request = pending_requests[request_id]
-            if pending_request.response_type is Content and isinstance(response, str):
-                response = Content.from_text(text=response)
-            # Try to coerce raw values (e.g., dicts from JSON) to the expected type
-            response = try_coerce_to_type(response, pending_request.response_type)
-            if not is_instance_of(response, pending_request.response_type):
-                raise ValueError(
-                    f"Response type mismatch for request ID {request_id}: "
-                    f"expected {pending_request.response_type}, got {type(response)}"
-                )
+            response = _coerce_request_info_response(response, pending_request.response_type, request_id)
             coerced_responses[request_id] = response
 
         # Cancelling siblings on error, like every other concurrent write into runner state. Each

python/packages/core/tests/workflow/test_functional_workflow.py

@@ -18,6 +18,7 @@
 from agent_framework import (
     AgentResponseUpdate,
     CheckpointStorage,
+    Content,
     ExperimentalFeature,
     FunctionalWorkflow,
     FunctionalWorkflowAgent,
@@ -332,6 +333,44 @@ async def review_wf(doc: str, ctx: RunContext) -> str:
         assert outputs == ["Final: Looks great!"]
         assert result2.get_final_state() == WorkflowRunState.IDLE
 
+    async def test_request_info_resume_rejects_response_type_mismatch(self):
+        @built_workflow
+        async def typed_wf(data: str, ctx: RunContext) -> str:
+            answer = await ctx.request_info("number", response_type=int, request_id="typed")
+            return f"{answer}:{type(answer).__name__}"
+
+        await typed_wf.run("input")
+
+        with pytest.raises(ValueError, match="Response type mismatch for request ID typed"):
+            await typed_wf.run(responses={"typed": "not-an-int"})
+
+    async def test_request_info_resume_coerces_json_like_response(self):
+        @dataclass
+        class Decision:
+            approved: bool
+
+        @built_workflow
+        async def typed_wf(data: str, ctx: RunContext) -> str:
+            decision = await ctx.request_info("decision", response_type=Decision, request_id="decision")
+            return f"{decision.approved}:{type(decision).__name__}"
+
+        await typed_wf.run("input")
+        result = await typed_wf.run(responses={"decision": {"approved": True}})
+
+        assert result.get_outputs() == ["True:Decision"]
+
+    async def test_request_info_resume_converts_text_to_content(self):
+        @built_workflow
+        async def content_wf(data: str, ctx: RunContext) -> str:
+            answer = await ctx.request_info("message", response_type=Content, request_id="content")
+            assert isinstance(answer, Content)
+            return f"{answer.type}:{answer.text}"
+
+        await content_wf.run("input")
+        result = await content_wf.run(responses={"content": "hello"})
+
+        assert result.get_outputs() == ["text:hello"]
+
     async def test_fresh_message_while_pending_requests_warns(self, caplog: pytest.LogCaptureFixture) -> None:
         """A fresh message while request_info events are pending is allowed but logs a warning."""
 

python/packages/core/tests/workflow/test_workflow.py

@@ -111,6 +111,24 @@ async def mock_handler_b(
             await ctx.send_message(NumberMessage(data=data))
 
 
+def test_coerce_request_info_response_converts_content() -> None:
+    """Request responses should use the shared Content conversion path."""
+    from agent_framework._workflows._workflow import _coerce_request_info_response
+
+    response = _coerce_request_info_response("hello", Content, "content")
+
+    assert isinstance(response, Content)
+    assert response.text == "hello"
+
+
+def test_coerce_request_info_response_rejects_mismatched_type() -> None:
+    """Request responses that cannot be validated should report their request ID."""
+    from agent_framework._workflows._workflow import _coerce_request_info_response
+
+    with pytest.raises(ValueError, match="Response type mismatch for request ID typed"):
+        _coerce_request_info_response("not-an-int", int, "typed")
+
+
 async def test_fresh_message_while_pending_advances_state_without_abandoning_requests(
     caplog: pytest.LogCaptureFixture,
 ) -> None: