diff --git a/python/packages/core/agent_framework/_workflows/_functional.py b/python/packages/core/agent_framework/_workflows/_functional.py index e83fa9e0e5c..1e1d4697d10 100644 --- a/python/packages/core/agent_framework/_workflows/_functional.py +++ b/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 diff --git a/python/packages/core/agent_framework/_workflows/_workflow.py b/python/packages/core/agent_framework/_workflows/_workflow.py index 84a43101c11..18fb5c9a4db 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow.py +++ b/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 diff --git a/python/packages/core/tests/workflow/test_functional_workflow.py b/python/packages/core/tests/workflow/test_functional_workflow.py index 5ed7ae0e8a3..395920bf22c 100644 --- a/python/packages/core/tests/workflow/test_functional_workflow.py +++ b/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.""" diff --git a/python/packages/core/tests/workflow/test_workflow.py b/python/packages/core/tests/workflow/test_workflow.py index 014694b9b80..762eff1d36e 100644 --- a/python/packages/core/tests/workflow/test_workflow.py +++ b/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: