diff --git a/python/packages/core/AGENTS.md b/python/packages/core/AGENTS.md
index 8e633a9334d..9327da2a423 100644
--- a/python/packages/core/AGENTS.md
+++ b/python/packages/core/AGENTS.md
@@ -200,6 +200,22 @@ agent_framework/
   every output-capable executor not selected by `output_from`.
 - **`WorkflowRunResult`** - Non-streaming workflow result with Workflow Output `get_outputs()`
   and Intermediate Output `get_intermediate_outputs()` accessors
+- **Request-info trust boundary** - Pending workflow state is authoritative for request data and Python types.
+  For a plain workflow, retain the emitted typed event or use
+  `await workflow.get_pending_request_info(request_id)`. For a workflow agent, pass the complete emitted
+  function-call `Content` to `await agent.resolve_request_info(content)`; the resolver validates correlation and
+  compatibility metadata, then returns the workflow-held event without trusting copied request data. Both lookups
+  are non-consuming; successful response submission through `run(...)` consumes the pending request. Serialized
+  request and response type names in the current wire envelope are compatibility metadata only, not authorization
+  data. Out-of-process consumers that do not own the live workflow can use
+  `WorkflowEvent.rehydrate_request_info(...)` or `WorkflowAgent.RequestInfoFunctionArgs.rehydrate(...)`, supplying
+  `allowed_types` for custom or parameterized annotations that are not available from an already-loaded module.
+  Default module resolution reads only top-level types physically present in an already-loaded module namespace;
+  nested, function-local, lazily exported, and parameterized annotations require `allowed_types`. These transport-only
+  rehydration methods do not establish that a request is still pending. `WorkflowEvent.from_dict` and
+  `WorkflowAgent.RequestInfoFunctionArgs.from_dict` remain only as deprecated compatibility paths; migrate them to the
+  corresponding `rehydrate...` method for transport decoding or to the authoritative workflow/agent resolver when the
+  owning workflow is available.
 - **Orchestrators**: `SequentialOrchestrator`, `ConcurrentOrchestrator`, `GroupChatOrchestrator`, `MagenticOrchestrator`, `HandoffOrchestrator`
 
 ## Built-in Providers
diff --git a/python/packages/core/agent_framework/_workflows/_agent.py b/python/packages/core/agent_framework/_workflows/_agent.py
index f0b51969c07..da7701e85f0 100644
--- a/python/packages/core/agent_framework/_workflows/_agent.py
+++ b/python/packages/core/agent_framework/_workflows/_agent.py
@@ -2,9 +2,11 @@
 
 from __future__ import annotations
 
+import json
 import logging
 import sys
 import uuid
+import warnings
 from collections.abc import AsyncIterable, Awaitable, Mapping, Sequence
 from dataclasses import dataclass
 from datetime import datetime, timezone
@@ -64,18 +66,69 @@ def to_dict(self) -> dict[str, Any]:
             return {"request_id": self.request_id, "request_event": self.request_event.to_dict()}
 
         @classmethod
-        def from_dict(cls, payload: dict[str, Any]) -> WorkflowAgent.RequestInfoFunctionArgs:
+        def rehydrate(
+            cls,
+            payload: Mapping[str, Any],
+            *,
+            allowed_types: Mapping[str, object] | None = None,
+        ) -> WorkflowAgent.RequestInfoFunctionArgs:
+            """Rehydrate request-info arguments from trusted transport data.
+
+            Args:
+                payload: Serialized request-info function arguments.
+
+            Keyword Args:
+                allowed_types: Optional mapping of serialized names to trusted custom types or typing annotations.
+
+            Returns:
+                The rehydrated request-info arguments.
+
+            Raises:
+                ValueError: If required request-info fields are missing or empty.
+            """
+            if not isinstance(payload, Mapping):
+                raise ValueError("Serialized request-info arguments payload must be a mapping.")
             if "request_id" not in payload or "request_event" not in payload:
                 raise ValueError(
                     "Invalid payload for RequestInfoFunctionArgs. 'request_id' and 'request_event' are required."
                 )
             if not payload["request_id"]:
                 raise ValueError("request_id cannot be empty.")
+            request_event = payload["request_event"]
+            if not isinstance(request_event, Mapping):
+                raise ValueError("Serialized request-info field 'request_event' must be a mapping.")
+            request_event_mapping = cast(Mapping[str, Any], request_event)
 
             return cls(
                 request_id=payload.get("request_id", ""),
-                request_event=WorkflowEvent.from_dict(payload.get("request_event", {})),
+                request_event=WorkflowEvent.rehydrate_request_info(
+                    request_event_mapping,
+                    allowed_types=allowed_types,
+                ),
+            )
+
+        @classmethod
+        def from_dict(
+            cls,
+            payload: dict[str, Any],
+            *,
+            allowed_types: Mapping[str, object] | None = None,
+        ) -> WorkflowAgent.RequestInfoFunctionArgs:
+            """Reconstruct request-info arguments with optional trusted custom types.
+
+            Deprecated:
+                Use :meth:`rehydrate` for transport decoding or
+                :meth:`WorkflowAgent.resolve_request_info` with a live workflow.
+            """
+            warnings.warn(
+                "`WorkflowAgent.RequestInfoFunctionArgs.from_dict` is deprecated and will be removed "
+                "in a future version; use `WorkflowAgent.RequestInfoFunctionArgs.rehydrate` for transport "
+                "decoding or `WorkflowAgent.resolve_request_info` "
+                "(`await agent.resolve_request_info(content)`) with a live workflow instead.",
+                DeprecationWarning,
+                stacklevel=2,
             )
+            return cls.rehydrate(payload, allowed_types=allowed_types)
 
     def __init__(
         self,
@@ -130,6 +183,83 @@ def __init__(
     def workflow(self) -> Workflow:
         return self._workflow
 
+    async def resolve_request_info(self, content: Content) -> WorkflowEvent[Any]:
+        """Resolve a request-info function call against the wrapped workflow's pending state.
+
+        Copied request data is ignored. Correlation and compatibility metadata are
+        validated against the authoritative event retained by the workflow.
+
+        Args:
+            content: Complete request-info function-call content emitted by this agent.
+
+        Returns:
+            The original pending request-info event retained by the workflow.
+
+        Raises:
+            ValueError: If the content is malformed, forged, stale, or replayed.
+            TypeError: If the pending request contains an unsupported wire type annotation.
+        """
+        if not isinstance(content, Content):
+            raise ValueError("Request-info content must be a Content instance.")
+        content_type = content.type
+        if type(content_type) is not str or content_type != "function_call":
+            raise ValueError("Request-info content type must be 'function_call'.")
+
+        function_name = content.name
+        if type(function_name) is not str or function_name != self.REQUEST_INFO_FUNCTION_NAME:
+            raise ValueError(f"Request-info function-call field 'name' must be {self.REQUEST_INFO_FUNCTION_NAME!r}.")
+
+        if isinstance(content.arguments, str):
+            try:
+                arguments = json.loads(content.arguments)
+            except json.JSONDecodeError:
+                raise ValueError("Request-info function-call field 'arguments' must be a JSON object.") from None
+        else:
+            arguments = content.parse_arguments()
+        if not isinstance(arguments, Mapping):
+            raise ValueError("Request-info function-call field 'arguments' must be a JSON object.")
+        arguments_mapping = cast(Mapping[str, Any], arguments)
+        arguments_dict = dict(arguments_mapping)
+
+        call_id = content.call_id
+        if type(call_id) is not str or not call_id:
+            raise ValueError("Request-info function-call field 'call_id' must be a non-empty string.")
+
+        request_id = arguments_dict.get("request_id")
+        if type(request_id) is not str or not request_id:
+            raise ValueError("Request-info function-call field 'arguments.request_id' must be a non-empty string.")
+
+        request_event_value = arguments_dict.get("request_event")
+        if not isinstance(request_event_value, Mapping):
+            raise ValueError("Request-info function-call field 'request_event' must be a JSON object.")
+        request_event_mapping = cast(Mapping[str, Any], request_event_value)
+        request_event = dict(request_event_mapping)
+
+        event_request_id = request_event.get("request_id")
+        if type(event_request_id) is not str or not event_request_id:
+            raise ValueError("Request-info function-call field 'request_event.request_id' must be a non-empty string.")
+        if call_id != request_id or request_id != event_request_id:
+            raise ValueError(
+                "Request-info correlation IDs in fields 'call_id', 'arguments.request_id', "
+                "and 'request_event.request_id' must match."
+            )
+
+        pending_event = await self._workflow.get_pending_request_info(request_id)
+        expected_arguments = self.RequestInfoFunctionArgs(
+            request_id=request_id,
+            request_event=pending_event,
+        ).to_dict()
+        expected_request_event = cast(dict[str, Any], expected_arguments["request_event"])
+        for field in ("type", "source_executor_id", "request_type", "response_type"):
+            expected_value = expected_request_event[field]
+            actual_value = request_event.get(field)
+            if type(actual_value) is not str or actual_value != expected_value:
+                raise ValueError(
+                    f"Request-info function-call field 'request_event.{field}' does not match the pending request."
+                )
+
+        return pending_event
+
     # region Run Methods
 
     @overload
diff --git a/python/packages/core/agent_framework/_workflows/_events.py b/python/packages/core/agent_framework/_workflows/_events.py
index 8788599e652..df29d8183f9 100644
--- a/python/packages/core/agent_framework/_workflows/_events.py
+++ b/python/packages/core/agent_framework/_workflows/_events.py
@@ -6,7 +6,7 @@
 import sys
 import traceback as _traceback
 import warnings
-from collections.abc import Generator
+from collections.abc import Generator, Mapping
 from contextlib import contextmanager
 from contextvars import ContextVar
 from dataclasses import dataclass
@@ -426,23 +426,71 @@ def to_dict(self) -> dict[str, Any]:
         }
 
     @classmethod
-    def from_dict(cls, data: dict[str, Any]) -> WorkflowEvent[Any]:
-        """Create a REQUEST_INFO event from a dictionary."""
+    def rehydrate_request_info(
+        cls,
+        data: Mapping[str, Any],
+        *,
+        allowed_types: Mapping[str, object] | None = None,
+    ) -> WorkflowEvent[Any]:
+        """Rehydrate a request-info event from trusted transport data.
+
+        Args:
+            data: Serialized request-info event fields.
+            allowed_types: Optional mapping of serialized names to trusted custom types or typing annotations.
+                Use this for nested, function-local, lazily exported, or parameterized annotations that cannot be
+                resolved as top-level types from an already-loaded module namespace.
+
+        Returns:
+            The rehydrated request-info event.
+
+        Raises:
+            ValueError: If the request-info event data is not a mapping.
+            KeyError: If a required request-info field is missing.
+            TypeError: If the request data type does not match its serialized metadata.
+            ModuleNotFoundError: If a response type's module is not already loaded.
+            AttributeError: If a loaded response-type module does not contain the serialized top-level type.
+        """
+        if not isinstance(data, Mapping):
+            raise ValueError("Serialized request-info event data must be a mapping.")
+
         for prop in ["data", "request_id", "source_executor_id", "request_type", "response_type"]:
             if prop not in data:
                 raise KeyError(f"Missing '{prop}' field in WorkflowEvent dictionary.")
-
         request_data = data["data"]
-        request_type = deserialize_type(data["request_type"])
-
-        if request_type is not type(request_data):
-            raise TypeError(
-                "Mismatch between deserialized request_data type and request_type field in WorkflowEvent dictionary."
-            )
+        request_type = cast(builtins.type[Any], type(request_data))
+        if serialize_type(request_type) != data["request_type"]:
+            raise TypeError("Mismatch between request_data type and request_type field in WorkflowEvent dictionary.")
 
         return cls.request_info(
             request_id=data["request_id"],
             source_executor_id=data["source_executor_id"],
             request_data=cast(Any, request_data),  # type: ignore
-            response_type=deserialize_type(data["response_type"]),
+            response_type=deserialize_type(data["response_type"], allowed_types=allowed_types),
+        )
+
+    @classmethod
+    def from_dict(
+        cls,
+        data: dict[str, Any],
+        *,
+        allowed_types: Mapping[str, object] | None = None,
+    ) -> WorkflowEvent[Any]:
+        """Create a REQUEST_INFO event from a dictionary.
+
+        Deprecated:
+            Use :meth:`rehydrate_request_info` for transport decoding or
+            ``Workflow.get_pending_request_info`` for authoritative lookup.
+
+        Args:
+            data: Serialized request-info event fields.
+            allowed_types: Optional mapping of serialized names to trusted custom types or typing annotations.
+        """
+        warnings.warn(
+            "`WorkflowEvent.from_dict` is deprecated and will be removed in a future version; "
+            "use `WorkflowEvent.rehydrate_request_info` for transport decoding or "
+            "`Workflow.get_pending_request_info` (`await workflow.get_pending_request_info(request_id)`) "
+            "for authoritative lookup instead.",
+            DeprecationWarning,
+            stacklevel=2,
         )
+        return cls.rehydrate_request_info(data, allowed_types=allowed_types)
diff --git a/python/packages/core/agent_framework/_workflows/_typing_utils.py b/python/packages/core/agent_framework/_workflows/_typing_utils.py
index 404b83f7092..42250d93e14 100644
--- a/python/packages/core/agent_framework/_workflows/_typing_utils.py
+++ b/python/packages/core/agent_framework/_workflows/_typing_utils.py
@@ -1,7 +1,12 @@
 # Copyright (c) Microsoft. All rights reserved.
 
+from __future__ import annotations
+
+import builtins
+import sys
 import typing
-from types import UnionType
+from collections.abc import Mapping
+from types import ModuleType, UnionType
 from typing import Any, TypeGuard, Union, cast, get_args, get_origin
 
 import typing_extensions
@@ -14,6 +19,120 @@
 _TYPEVAR_TYPES: tuple[type, ...] = (type(typing.TypeVar("_T")), type(typing_extensions.TypeVar("_T")))  # pyright: ignore[reportUnknownVariableType]
 
 
+def _validate_serialized_type_name(serialized_name: object) -> str:
+    if not isinstance(serialized_name, str):
+        raise TypeError("Serialized type names must be strings.")
+    parts = serialized_name.split(".")
+    if len(parts) < 2 or any(not part for part in parts):
+        raise ValueError(f"Malformed serialized type name '{serialized_name}'.")
+    return serialized_name
+
+
+def _is_runtime_type(value: object) -> TypeGuard[type[Any]]:
+    if not isinstance(value, type):
+        return False
+
+    # Python 3.10 reports GenericAlias instances as types even though they are not class objects.
+    try:
+        type.__getattribute__(value, "__module__")
+        type.__getattribute__(value, "__qualname__")
+    except TypeError:
+        return False
+
+    return True
+
+
+def _trusted_annotation_name(value: object) -> str:
+    if _is_runtime_type(value):
+        module_name = type.__getattribute__(value, "__module__")
+        qualified_name = type.__getattribute__(value, "__qualname__")
+    else:
+        module_name = getattr(value, "__module__", None)
+        qualified_name = getattr(value, "__qualname__", None)
+        if not isinstance(module_name, str) or not isinstance(qualified_name, str):
+            raise TypeError("Serialized type values must be actual types or supported typing annotations.")
+
+    if not isinstance(module_name, str) or not module_name or not isinstance(qualified_name, str) or not qualified_name:
+        raise ValueError("Types must have non-empty string __module__ and __qualname__ attributes.")
+    return _validate_serialized_type_name(f"{module_name}.{qualified_name}")
+
+
+def _serialized_type_name(value: object) -> str:
+    if get_origin(value) in (Union, UnionType):
+        raise TypeError("Union annotations cannot be serialized without losing type arguments.")
+    return _trusted_annotation_name(value)
+
+
+def _is_supported_type_annotation(value: object) -> bool:
+    if _is_runtime_type(value):
+        return True
+
+    if get_origin(value) in (Union, UnionType):
+        return True
+
+    try:
+        _trusted_annotation_name(value)
+    except (TypeError, ValueError):
+        return False
+
+    return get_origin(value) is not None
+
+
+def _trusted_annotation_matches_name(serialized_name: str, value: object) -> bool:
+    try:
+        if _trusted_annotation_name(value) == serialized_name:
+            return True
+    except TypeError:
+        pass
+
+    return (
+        serialized_name in ("typing.Optional", "typing.Union")
+        and get_origin(value) in (Union, UnionType)
+        and (serialized_name == "typing.Union" or type(None) in get_args(value))
+    )
+
+
+def _build_legacy_typing_aliases() -> dict[str, object]:
+    aliases: dict[str, object] = {}
+    for public_name, value in vars(typing).items():
+        if public_name.startswith("_") or not _is_runtime_type(get_origin(value)):
+            continue
+
+        try:
+            serialized_name = _trusted_annotation_name(value)
+        except (TypeError, ValueError):
+            continue
+
+        if serialized_name == f"typing.{public_name}":
+            aliases[serialized_name] = value
+
+    return aliases
+
+
+_BUILTIN_SERIALIZED_TYPES: dict[str, type[Any]] = {
+    _serialized_type_name(value): value for value in vars(builtins).values() if isinstance(value, type)
+}
+_LEGACY_TYPING_ALIASES = _build_legacy_typing_aliases()
+
+
+def _resolve_loaded_serialized_type(serialized_name: str) -> object:
+    module_name, _, type_name = serialized_name.rpartition(".")
+    loaded_module = sys.modules.get(module_name)
+    if not isinstance(loaded_module, ModuleType):
+        raise ModuleNotFoundError(f"No module named {module_name!r}", name=module_name)
+
+    namespace = ModuleType.__getattribute__(loaded_module, "__dict__")
+    if type_name not in namespace:
+        raise AttributeError(f"{module_name!r} has no attribute {type_name!r}")
+
+    resolved = namespace[type_name]
+    if not _is_runtime_type(resolved):
+        raise TypeError(f"{serialized_name!r} does not resolve to a type or supported typing annotation.")
+    if _serialized_type_name(resolved) != serialized_name:
+        raise ValueError(f"Serialized type name '{serialized_name}' does not match the resolved type.")
+    return resolved
+
+
 def is_typevar(x: Any) -> bool:
     """Check if x is an unresolved TypeVar instance (from typing or typing_extensions).
 
@@ -264,29 +383,82 @@ def try_coerce_to_type(data: Any, target_type: type | UnionType | Any) -> Any:
     return original_data
 
 
-def serialize_type(t: type) -> str:
+def serialize_type(t: object) -> str:
     """Serialize a type to a string.
 
+    Typing annotations retain their historical module and qualified-name wire
+    representation.
+
     For example,
 
     serialize_type(int) => "builtins.int"
+
+    Raises:
+        TypeError: If the annotation cannot be represented without losing type arguments.
     """
-    return f"{t.__module__}.{t.__qualname__}"
+    return _serialized_type_name(t)
 
 
-def deserialize_type(serialized_type_string: str) -> type:
+def deserialize_type(
+    serialized_type_string: str,
+    *,
+    allowed_types: Mapping[str, object] | None = None,
+) -> type:
     """Deserialize a serialized type string.
 
+    Resolution is limited to built-in types, supported aliases from the trusted
+    :mod:`typing` module, exact entries in ``allowed_types``, and types already
+    present in loaded module namespaces. It never imports or invokes attribute
+    hooks on a module selected by the serialized value.
+
+    Args:
+        serialized_type_string: Fully qualified serialized type name.
+        allowed_types: Optional per-call mapping of serialized names to trusted types or typing annotations. The
+            explicit key supplies the stable wire name, including historical names such as ``typing.Optional`` after
+            runtime canonicalization. Use explicit entries for nested, local, lazily exported, or parameterized types.
+
     For example,
 
     deserialize_type("builtins.int") => int
-    """
-    import importlib
-
-    module_name, _, type_name = serialized_type_string.rpartition(".")
-    module = importlib.import_module(module_name)
 
-    return cast(type, getattr(module, type_name))
+    Raises:
+        ModuleNotFoundError: If the serialized module is not already loaded.
+        AttributeError: If a loaded module does not physically contain the serialized top-level type in its namespace.
+        TypeError: If the resolved value is not a supported type annotation.
+        ValueError: If an explicit or loaded type does not match the serialized name.
+    """
+    serialized_name = _validate_serialized_type_name(serialized_type_string)
+
+    explicit_type: object | None = None
+    if allowed_types is not None:
+        if not isinstance(allowed_types, Mapping):
+            raise TypeError("allowed_types must be a mapping of serialized names to types or typing annotations.")
+        for allowed_name, allowed_type in allowed_types.items():
+            validated_name = _validate_serialized_type_name(allowed_name)
+            if not _is_supported_type_annotation(allowed_type):
+                raise TypeError(
+                    f"allowed_types entry '{validated_name}' must be an actual type or supported typing annotation."
+                )
+            if not _trusted_annotation_matches_name(validated_name, allowed_type):
+                raise ValueError(f"allowed_types entry '{validated_name}' does not match the supplied type.")
+            if validated_name == serialized_name:
+                explicit_type = allowed_type
+
+    builtin_type = _BUILTIN_SERIALIZED_TYPES.get(serialized_name)
+    legacy_typing_alias = _LEGACY_TYPING_ALIASES.get(serialized_name)
+
+    if explicit_type is not None:
+        resolved_type = explicit_type
+    elif builtin_type is not None:
+        resolved_type = builtin_type
+    elif legacy_typing_alias is not None:
+        resolved_type = legacy_typing_alias
+    else:
+        resolved_type = _resolve_loaded_serialized_type(serialized_name)
+
+    if not _trusted_annotation_matches_name(serialized_name, resolved_type):
+        raise ValueError(f"Serialized type name '{serialized_name}' does not match the resolved type.")
+    return cast(type, resolved_type)
 
 
 def is_type_compatible(source_type: type | UnionType | Any, target_type: type | UnionType | Any) -> bool:
diff --git a/python/packages/core/agent_framework/_workflows/_workflow.py b/python/packages/core/agent_framework/_workflows/_workflow.py
index 77060b93a91..d78cbb5e25e 100644
--- a/python/packages/core/agent_framework/_workflows/_workflow.py
+++ b/python/packages/core/agent_framework/_workflows/_workflow.py
@@ -383,6 +383,27 @@ def status(self) -> WorkflowRunState:
         """
         return self._status
 
+    async def get_pending_request_info(self, request_id: str) -> WorkflowEvent[Any]:
+        """Return the authoritative pending request-info event for ``request_id``.
+
+        Lookup is non-consuming. The event remains pending until a response is
+        successfully submitted through :meth:`run`.
+
+        Args:
+            request_id: Correlation ID of the pending request-info event.
+
+        Returns:
+            The original typed event retained by the workflow.
+
+        Raises:
+            ValueError: If ``request_id`` does not identify a pending request.
+        """
+        pending_requests = await self._runner.context.get_pending_request_info_events()
+        try:
+            return pending_requests[request_id]
+        except KeyError:
+            raise ValueError(f"No pending request-info event found for request ID {request_id!r}.") from None
+
     def to_dict(self) -> dict[str, Any]:
         """Serialize the workflow definition into a JSON-ready dictionary."""
         data: dict[str, Any] = {
diff --git a/python/packages/core/tests/workflow/test_request_info_and_response.py b/python/packages/core/tests/workflow/test_request_info_and_response.py
index 399a300bfe3..9dc4586a135 100644
--- a/python/packages/core/tests/workflow/test_request_info_and_response.py
+++ b/python/packages/core/tests/workflow/test_request_info_and_response.py
@@ -2,6 +2,8 @@
 
 from dataclasses import dataclass
 
+import pytest
+
 from agent_framework import (
     WorkflowBuilder,
     WorkflowContext,
@@ -199,6 +201,52 @@ async def test_approval_workflow(self):
         assert executor.approval_received is True
         assert executor.final_result == "Operation approved: Please approve the operation: test operation"
 
+    async def test_pending_request_info_lookup_returns_original_event_without_consuming_it(self):
+        executor = ApprovalRequiredExecutor(id="approval_executor")
+        workflow = WorkflowBuilder(start_executor=executor).build()
+
+        result = await workflow.run("test operation")
+        request_info_event = result.get_request_info_events()[0]
+
+        first_lookup = await workflow.get_pending_request_info(request_info_event.request_id)
+        second_lookup = await workflow.get_pending_request_info(request_info_event.request_id)
+
+        assert first_lookup is request_info_event
+        assert second_lookup is request_info_event
+        assert first_lookup.data is request_info_event.data
+        assert first_lookup.request_type is UserApprovalRequest
+        assert first_lookup.response_type is bool
+
+    async def test_pending_request_info_lookup_rejects_unknown_request_id(self):
+        executor = ApprovalRequiredExecutor(id="approval_executor")
+        workflow = WorkflowBuilder(start_executor=executor).build()
+
+        with pytest.raises(
+            ValueError,
+            match="No pending request-info event found for request ID 'missing-request'",
+        ):
+            await workflow.get_pending_request_info("missing-request")
+
+    async def test_successful_response_consumes_pending_request_info(self):
+        executor = ApprovalRequiredExecutor(id="approval_executor")
+        workflow = WorkflowBuilder(start_executor=executor).build()
+        result = await workflow.run("test operation")
+        request_info_event = result.get_request_info_events()[0]
+
+        with pytest.raises(ValueError, match="Response type mismatch"):
+            await workflow.run(responses={request_info_event.request_id: object()})
+
+        assert await workflow.get_pending_request_info(request_info_event.request_id) is request_info_event
+
+        await workflow.run(responses={request_info_event.request_id: True})
+
+        with pytest.raises(ValueError, match="No pending request-info event found") as completed_error:
+            await workflow.get_pending_request_info(request_info_event.request_id)
+        assert request_info_event.data.prompt not in str(completed_error.value)
+
+        with pytest.raises(ValueError, match="No pending request-info event found"):
+            await workflow.get_pending_request_info(request_info_event.request_id)
+
     async def test_calculation_workflow(self):
         """Test end-to-end workflow with calculation request."""
         executor = CalculationExecutor(id="calc_executor")
diff --git a/python/packages/core/tests/workflow/test_request_info_event_rehydrate.py b/python/packages/core/tests/workflow/test_request_info_event_rehydrate.py
index c05ab66984e..2ba6629f905 100644
--- a/python/packages/core/tests/workflow/test_request_info_event_rehydrate.py
+++ b/python/packages/core/tests/workflow/test_request_info_event_rehydrate.py
@@ -1,8 +1,15 @@
 # Copyright (c) Microsoft. All rights reserved.
 
 import json
+import sys
+import typing
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
+from pathlib import Path
+from types import ModuleType
+from typing import Any, cast
+
+import pytest
 
 from agent_framework import (
     FileCheckpointStorage,
@@ -47,6 +54,267 @@ class TimedApproval:
     issued_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
 
 
+def test_workflow_event_rehydrate_request_info_accepts_explicit_custom_type_mapping() -> None:
+    """Transport rehydration accepts caller-supplied trusted custom types without a warning."""
+
+    @dataclass
+    class ExplicitRequest:
+        prompt: str
+
+    class ExplicitResponse:
+        pass
+
+    request_type_name = f"{ExplicitRequest.__module__}.{ExplicitRequest.__qualname__}"
+    response_type_name = f"{ExplicitResponse.__module__}.{ExplicitResponse.__qualname__}"
+
+    event = WorkflowEvent.rehydrate_request_info(
+        {
+            "type": "request_info",
+            "data": ExplicitRequest(prompt="Approve?"),
+            "request_id": "request-123",
+            "source_executor_id": "review_gateway",
+            "request_type": request_type_name,
+            "response_type": response_type_name,
+        },
+        allowed_types={
+            request_type_name: ExplicitRequest,
+            response_type_name: ExplicitResponse,
+        },
+    )
+
+    assert type(event.data) is ExplicitRequest
+    assert event.request_type is ExplicitRequest
+    assert event.response_type is ExplicitResponse
+
+
+def test_workflow_event_rehydrate_request_info_rejects_non_mapping_data() -> None:
+    """Transport rehydration rejects malformed event containers at the interface."""
+    malformed_data = "data request_id source_executor_id request_type response_type"
+
+    with pytest.raises(ValueError, match="request-info event data must be a mapping"):
+        WorkflowEvent.rehydrate_request_info(cast(Any, malformed_data))
+
+
+@pytest.mark.parametrize(
+    ("serialized_name", "response_annotation"),
+    [
+        ("typing.Optional", typing.Optional[str]),
+        ("typing.Union", typing.Union[str, int]),
+    ],
+)
+def test_workflow_event_rehydrate_request_info_accepts_parameterized_annotation_mapping(
+    serialized_name: str,
+    response_annotation: object,
+) -> None:
+    """Transport rehydration accepts an exact trusted parameterized response annotation."""
+    event = WorkflowEvent.rehydrate_request_info(
+        {
+            "type": "request_info",
+            "data": "Approve?",
+            "request_id": "request-123",
+            "source_executor_id": "review_gateway",
+            "request_type": "builtins.str",
+            "response_type": serialized_name,
+        },
+        allowed_types={serialized_name: response_annotation},
+    )
+
+    assert event.response_type == response_annotation
+
+
+@pytest.mark.parametrize(
+    "response_annotation",
+    [
+        str | None,
+        typing.Optional[str],
+        typing.Union[str, int],
+    ],
+)
+def test_workflow_event_to_dict_rejects_lossy_union_response_type(response_annotation: object) -> None:
+    """Wire serialization rejects unions whose member types cannot be preserved."""
+    event = WorkflowEvent.request_info(
+        request_id="request-123",
+        source_executor_id="review_gateway",
+        request_data="Approve?",
+        response_type=cast(Any, response_annotation),
+    )
+
+    with pytest.raises(TypeError, match="Union annotations cannot be serialized without losing type arguments"):
+        event.to_dict()
+
+
+def test_workflow_event_rehydrate_request_info_resolves_loaded_module_type(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Transport rehydration resolves a module-level type without importing its module."""
+    module_name = "_loaded_request_info_types"
+    loaded_module = ModuleType(module_name)
+
+    class LoadedResponse:
+        pass
+
+    LoadedResponse.__module__ = module_name
+    LoadedResponse.__qualname__ = "LoadedResponse"
+    loaded_module.__dict__["LoadedResponse"] = LoadedResponse
+    monkeypatch.setitem(sys.modules, module_name, loaded_module)
+
+    event = WorkflowEvent.rehydrate_request_info({
+        "type": "request_info",
+        "data": "Approve?",
+        "request_id": "request-123",
+        "source_executor_id": "review_gateway",
+        "request_type": "builtins.str",
+        "response_type": f"{module_name}.LoadedResponse",
+    })
+
+    assert event.response_type is LoadedResponse
+
+
+def test_workflow_event_rehydrate_request_info_handles_colliding_request_and_response_names() -> None:
+    """Request validation remains concrete when the response annotation shares its wire name."""
+    response_annotation = list[str]
+
+    event = WorkflowEvent.rehydrate_request_info(
+        {
+            "type": "request_info",
+            "data": [],
+            "request_id": "request-123",
+            "source_executor_id": "review_gateway",
+            "request_type": "builtins.list",
+            "response_type": "builtins.list",
+        },
+        allowed_types={"builtins.list": response_annotation},
+    )
+
+    assert event.request_type is list
+    assert event.response_type == response_annotation
+
+
+def test_workflow_event_from_dict_accepts_explicit_custom_type_mapping() -> None:
+    """Legacy reconstruction accepts caller-supplied trusted custom types."""
+
+    @dataclass
+    class ExplicitRequest:
+        prompt: str
+
+    class ExplicitResponse:
+        pass
+
+    request_type_name = f"{ExplicitRequest.__module__}.{ExplicitRequest.__qualname__}"
+    response_type_name = f"{ExplicitResponse.__module__}.{ExplicitResponse.__qualname__}"
+    with pytest.warns(
+        DeprecationWarning,
+        match=r"WorkflowEvent\.from_dict.*will be removed in a future version"
+        r".*WorkflowEvent\.rehydrate_request_info",
+    ) as recorded_warnings:
+        event = WorkflowEvent.from_dict(
+            {
+                "type": "request_info",
+                "data": ExplicitRequest(prompt="Approve?"),
+                "request_id": "request-123",
+                "source_executor_id": "review_gateway",
+                "request_type": request_type_name,
+                "response_type": response_type_name,
+            },
+            allowed_types={
+                request_type_name: ExplicitRequest,
+                response_type_name: ExplicitResponse,
+            },
+        )
+
+    assert len(recorded_warnings) == 1
+    assert recorded_warnings[0].filename == __file__
+    assert type(event.data) is ExplicitRequest
+    assert event.request_type is ExplicitRequest
+    assert event.response_type is ExplicitResponse
+
+
+def test_workflow_event_from_dict_requires_exact_request_data_type() -> None:
+    """Request metadata cannot authorize subclass request data."""
+
+    class BaseRequest:
+        pass
+
+    class DerivedRequest(BaseRequest):
+        pass
+
+    request_type_name = f"{BaseRequest.__module__}.{BaseRequest.__qualname__}"
+    with (
+        pytest.warns(DeprecationWarning),
+        pytest.raises(TypeError, match="Mismatch between request_data type"),
+    ):
+        WorkflowEvent.from_dict(
+            {
+                "type": "request_info",
+                "data": DerivedRequest(),
+                "request_id": "request-123",
+                "source_executor_id": "review_gateway",
+                "request_type": request_type_name,
+                "response_type": "builtins.bool",
+            },
+            allowed_types={request_type_name: BaseRequest},
+        )
+
+
+def _write_observable_type_module(tmp_path: Path, module_name: str) -> Path:
+    marker_path = tmp_path / f"{module_name}.imported"
+    module_path = tmp_path / f"{module_name}.py"
+    module_path.write_text(f"from pathlib import Path\nPath({str(marker_path)!r}).touch()\nclass Attack:\n    pass\n")
+    return marker_path
+
+
+def test_workflow_event_from_dict_does_not_import_request_type(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A malicious request-type name cannot execute module code."""
+    module_name = "_malicious_request_info_request_type"
+    marker_path = _write_observable_type_module(tmp_path, module_name)
+    monkeypatch.syspath_prepend(str(tmp_path))
+
+    with pytest.warns(DeprecationWarning), pytest.raises(TypeError, match="Mismatch between request_data type"):
+        WorkflowEvent.from_dict({
+            "type": "request_info",
+            "data": "Approve?",
+            "request_id": "request-123",
+            "source_executor_id": "review_gateway",
+            "request_type": f"{module_name}.Attack",
+            "response_type": "builtins.bool",
+        })
+
+    assert module_name not in sys.modules
+    assert not marker_path.exists()
+
+
+def test_workflow_event_from_dict_does_not_import_response_type(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A malicious response-type name cannot execute module code."""
+    module_name = "_malicious_request_info_response_type"
+    marker_path = _write_observable_type_module(tmp_path, module_name)
+    monkeypatch.syspath_prepend(str(tmp_path))
+
+    with (
+        pytest.warns(DeprecationWarning),
+        pytest.raises(
+            ModuleNotFoundError,
+            match=f"No module named '{module_name}'",
+        ),
+    ):
+        WorkflowEvent.from_dict({
+            "type": "request_info",
+            "data": "Approve?",
+            "request_id": "request-123",
+            "source_executor_id": "review_gateway",
+            "request_type": "builtins.str",
+            "response_type": f"{module_name}.Attack",
+        })
+
+    assert module_name not in sys.modules
+    assert not marker_path.exists()
+
+
 async def test_rehydrate_request_info_event() -> None:
     """Rehydration should succeed for valid request info events."""
     request_info_event = WorkflowEvent.request_info(
@@ -195,8 +463,12 @@ async def test_checkpoint_with_pending_request_info_events():
         assert completed, "Workflow should reach idle with pending requests state after restoration"
         assert restored_request_event is not None, "Restored request info event should be emitted"
 
-        # Verify the restored event matches the original
+        # Verify the restored workflow retains the same typed event it emitted.
+        restored_lookup = await restored_workflow.get_pending_request_info(request_info_event.request_id)
+        assert restored_lookup is restored_request_event
         assert restored_request_event.source_executor_id == request_info_event.source_executor_id
+        assert restored_request_event.request_type is UserApprovalRequest
+        assert restored_request_event.response_type is bool
         assert isinstance(restored_request_event.data, UserApprovalRequest)
         assert restored_request_event.data.prompt == request_info_event.data.prompt
         assert restored_request_event.data.context == request_info_event.data.context
@@ -218,6 +490,8 @@ async def test_checkpoint_with_pending_request_info_events():
         assert new_executor.approval_received is True
         expected_result = "Operation approved: Please approve the operation: checkpoint test operation"
         assert new_executor.final_result == expected_result
+        with pytest.raises(ValueError, match="No pending request-info event found"):
+            await restored_workflow.get_pending_request_info(request_info_event.request_id)
 
 
 async def test_checkpoint_restore_with_responses_does_not_reemit_handled_requests():
diff --git a/python/packages/core/tests/workflow/test_typing_utils.py b/python/packages/core/tests/workflow/test_typing_utils.py
index cc3e7bf4ea3..873e9bc400b 100644
--- a/python/packages/core/tests/workflow/test_typing_utils.py
+++ b/python/packages/core/tests/workflow/test_typing_utils.py
@@ -1,11 +1,16 @@
 # Copyright (c) Microsoft. All rights reserved.
 
+import importlib
+import sys
+import typing
 from dataclasses import dataclass
+from pathlib import Path
+from types import ModuleType
 from typing import Any, Generic, Optional, TypeVar, Union
 
 import pytest
 
-from agent_framework import WorkflowEvent
+from agent_framework import Message, WorkflowEvent
 from agent_framework._workflows._typing_utils import (
     deserialize_type,
     is_instance_of,
@@ -284,6 +289,32 @@ class TestClass:
     assert serialize_type(TestClass) == expected
 
 
+def test_serialize_type_parameterized_generic_preserves_wire_name() -> None:
+    """Parameterized generics preserve their historical wire names."""
+    serialized_name = serialize_type(list[Message])
+
+    assert serialized_name == "builtins.list"
+    assert deserialize_type(serialized_name) is list
+    legacy_list_alias = vars(typing)["List"]
+    legacy_list = legacy_list_alias[Message]
+    legacy_serialized_name = serialize_type(legacy_list)
+    assert legacy_serialized_name == "typing.List"
+    assert deserialize_type(legacy_serialized_name) is legacy_list_alias
+
+
+@pytest.mark.parametrize(
+    "alias_name",
+    ["List", "Dict", "Tuple", "Set", "FrozenSet", "Sequence", "Mapping", "Callable", "Type"],
+)
+def test_deserialize_type_supports_legacy_typing_aliases(alias_name: str) -> None:
+    """Trusted standard typing aliases retain their historical wire compatibility."""
+    legacy_alias = vars(typing)[alias_name]
+    serialized_name = f"typing.{alias_name}"
+
+    assert serialize_type(legacy_alias) == serialized_name
+    assert deserialize_type(serialized_name) is legacy_alias
+
+
 def test_deserialize_type() -> None:
     """Test deserialization of type strings back to types."""
     # Test built-in types
@@ -324,19 +355,202 @@ def test_serialize_deserialize_roundtrip() -> None:
     assert instance.type == "request_info"
 
 
+def test_deserialize_type_accepts_explicit_custom_type_mapping() -> None:
+    """Callers can resolve a trusted custom type without first serializing it."""
+
+    class ExplicitlyAllowedType:
+        pass
+
+    serialized_name = f"{ExplicitlyAllowedType.__module__}.{ExplicitlyAllowedType.__qualname__}"
+
+    assert (
+        deserialize_type(serialized_name, allowed_types={serialized_name: ExplicitlyAllowedType})
+        is ExplicitlyAllowedType
+    )
+
+
+def test_deserialize_type_accepts_historical_optional_name_for_union_annotation() -> None:
+    """A trusted Optional annotation can rehydrate payloads emitted before runtime canonicalization."""
+    optional_string = str | None
+
+    assert deserialize_type("typing.Optional", allowed_types={"typing.Optional": optional_string}) == optional_string
+
+
+def test_serialize_type_preserves_unusual_runtime_type_name() -> None:
+    """Compatibility names remain the exact module and qualified name strings."""
+
+    class UnusualType:
+        pass
+
+    UnusualType.__qualname__ = "Request-Type"
+    expected_name = f"{UnusualType.__module__}.{UnusualType.__qualname__}"
+
+    assert serialize_type(UnusualType) == expected_name
+    assert deserialize_type(expected_name, allowed_types={expected_name: UnusualType}) is UnusualType
+
+
+def test_deserialize_type_rejects_non_type_mapping_value() -> None:
+    """Explicit compatibility mappings cannot resolve arbitrary objects."""
+    invalid_mapping: Any = {"trusted.Request": object()}
+
+    with pytest.raises(TypeError, match="must be an actual type"):
+        deserialize_type("trusted.Request", allowed_types=invalid_mapping)
+
+
+def test_deserialize_type_rejects_alias_mapping() -> None:
+    """A type cannot be authorized under a different serialized name."""
+
+    class TrustedRequest:
+        pass
+
+    with pytest.raises(ValueError, match="does not match the supplied type"):
+        deserialize_type("trusted.Alias", allowed_types={"trusted.Alias": TrustedRequest})
+
+
+def test_deserialize_type_rejects_subclass_for_base_name() -> None:
+    """Subclass compatibility is insufficient for serialized type identity."""
+
+    class BaseRequest:
+        pass
+
+    class DerivedRequest(BaseRequest):
+        pass
+
+    base_name = f"{BaseRequest.__module__}.{BaseRequest.__qualname__}"
+    with pytest.raises(ValueError, match="does not match the supplied type"):
+        deserialize_type(base_name, allowed_types={base_name: DerivedRequest})
+
+
+def test_deserialize_type_explicit_mapping_resolves_same_named_local_type() -> None:
+    """An exact per-call mapping selects a trusted local type with a reused compatibility name."""
+
+    class RegisteredRequest:
+        pass
+
+    class ConflictingRequest:
+        pass
+
+    serialized_name = serialize_type(RegisteredRequest)
+    ConflictingRequest.__module__ = RegisteredRequest.__module__
+    ConflictingRequest.__qualname__ = RegisteredRequest.__qualname__
+
+    assert deserialize_type(serialized_name, allowed_types={serialized_name: ConflictingRequest}) is ConflictingRequest
+
+
+def test_serialize_type_allows_repeated_factory_types_with_same_name() -> None:
+    """Independent factories can serialize distinct types that share one compatibility name."""
+
+    def make_request_type() -> type:
+        class Request:
+            pass
+
+        return Request
+
+    first_type = make_request_type()
+    second_type = make_request_type()
+
+    serialized_name = serialize_type(first_type)
+    assert serialize_type(second_type) == serialized_name
+
+    with pytest.raises(ModuleNotFoundError, match="<locals>"):
+        deserialize_type(serialized_name)
+
+    assert deserialize_type(serialized_name, allowed_types={serialized_name: second_type}) is second_type
+
+
+def test_serialize_type_allows_reloaded_type_with_same_name(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Module reload can replace a type without permanently poisoning serialization."""
+    module_name = "_request_info_reload_type"
+    module_path = tmp_path / f"{module_name}.py"
+    module_path.write_text("class ReloadedRequest:\n    pass\n", encoding="utf-8")
+    monkeypatch.syspath_prepend(str(tmp_path))
+    importlib.invalidate_caches()
+
+    module = importlib.import_module(module_name)
+    try:
+        first_type = module.ReloadedRequest
+        serialized_name = serialize_type(first_type)
+
+        reloaded_module = importlib.reload(module)
+        second_type = reloaded_module.ReloadedRequest
+
+        assert second_type is not first_type
+        assert serialize_type(second_type) == serialized_name
+        assert deserialize_type(serialized_name) is second_type
+    finally:
+        sys.modules.pop(module_name, None)
+
+
+@pytest.mark.parametrize("serialized_name", ["", "int", ".int", "builtins..int", "builtins.int."])
+def test_deserialize_type_rejects_malformed_names(serialized_name: str) -> None:
+    """Malformed serialized names fail before compatibility lookup."""
+    with pytest.raises(ValueError, match="Malformed serialized type name"):
+        deserialize_type(serialized_name)
+
+
+def test_deserialize_type_does_not_access_payload_selected_module_attributes(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Unknown names do not invoke module attribute hooks."""
+    module_name = "_request_info_module_with_attribute_hook"
+    accessed_attributes: list[str] = []
+
+    class ObservableModule(ModuleType):
+        def __getattribute__(self, name: str) -> Any:
+            if name == "__dict__":
+                accessed_attributes.append(name)
+            return ModuleType.__getattribute__(self, name)
+
+        def __getattr__(self, name: str) -> Any:
+            accessed_attributes.append(name)
+            raise AttributeError(name)
+
+    selected_module = ObservableModule(module_name)
+    monkeypatch.setitem(sys.modules, module_name, selected_module)
+
+    with pytest.raises(AttributeError, match="has no attribute 'Attack'"):
+        deserialize_type(f"{module_name}.Attack")
+
+    assert accessed_attributes == []
+
+
 def test_deserialize_type_error_handling() -> None:
     """Test error handling in deserialize_type function."""
-    import pytest
-
-    # Test with non-existent module
-    with pytest.raises(ModuleNotFoundError):
+    with pytest.raises(ModuleNotFoundError, match="No module named 'nonexistent.module'"):
         deserialize_type("nonexistent.module.Type")
 
-    # Test with non-existent type in existing module
-    with pytest.raises(AttributeError):
+    with pytest.raises(AttributeError, match="has no attribute 'NonExistentType'"):
         deserialize_type("builtins.NonExistentType")
 
 
+def test_deserialize_type_requires_exact_loaded_module_boundary(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A loaded parent package does not turn an unloaded submodule into an attribute lookup."""
+    importlib.import_module("xml")
+
+    monkeypatch.delitem(sys.modules, "xml.not_loaded", raising=False)
+
+    with pytest.raises(ModuleNotFoundError, match="No module named 'xml.not_loaded'"):
+        deserialize_type("xml.not_loaded.Type")
+
+
+def test_deserialize_type_does_not_import_unknown_module(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Unknown serialized types must fail without importing payload-selected modules."""
+    imported_modules: list[str] = []
+
+    def track_import(module_name: str) -> None:
+        imported_modules.append(module_name)
+
+    monkeypatch.setattr(importlib, "import_module", track_import)
+
+    with pytest.raises(ModuleNotFoundError, match="No module named 'untrusted_request_info_payload'"):
+        deserialize_type("untrusted_request_info_payload.Attack")
+
+    assert imported_modules == []
+
+
 def test_type_compatibility_basic() -> None:
     """Test basic type compatibility scenarios."""
     # Exact type match
diff --git a/python/packages/core/tests/workflow/test_workflow_agent.py b/python/packages/core/tests/workflow/test_workflow_agent.py
index 0cf69aa9414..9f50211816b 100644
--- a/python/packages/core/tests/workflow/test_workflow_agent.py
+++ b/python/packages/core/tests/workflow/test_workflow_agent.py
@@ -1,9 +1,12 @@
 # Copyright (c) Microsoft. All rights reserved.
 
+import json
 import uuid
 from collections.abc import Awaitable, Sequence
+from copy import deepcopy
 from dataclasses import dataclass
-from typing import Any, Literal, overload
+from types import MappingProxyType
+from typing import Any, Literal, cast, overload
 
 import pytest
 from typing_extensions import Never
@@ -37,9 +40,8 @@
 class HandoffRequest:
     """Module-level dataclass used by request_info tests.
 
-    Defined at module scope (not nested inside a test method) so
-    ``serialize_type``/``deserialize_type`` can round-trip the request_type via
-    the importable qualified name ``tests.workflow.test_workflow_agent.HandoffRequest``.
+    Its stable qualified name is recorded by ``serialize_type`` for same-process
+    compatibility round trips.
     """
 
     target_agent: str
@@ -159,6 +161,16 @@ async def handle_message(
         await ctx.send_message([response_message])
 
 
+async def _create_pending_request_info_call() -> tuple[WorkflowAgent, Content]:
+    workflow = WorkflowBuilder(start_executor=RequestingExecutor(id="requester")).build()
+    agent = WorkflowAgent(workflow=workflow, name="Request Test Agent")
+    response = await agent.run("Start request")
+    function_call = next(
+        content for message in response.messages for content in message.contents if content.type == "function_call"
+    )
+    return agent, function_call
+
+
 class TestWorkflowAgent:
     """Test cases for WorkflowAgent end-to-end functionality."""
 
@@ -274,17 +286,12 @@ async def test_end_to_end_request_info_handling(self):
         assert request_event.get("type") == "request_info"
         assert deserialize_type(request_event.get("response_type")) is str
 
-        deserialized_args = WorkflowAgent.RequestInfoFunctionArgs.from_dict(request_function_call.arguments)  # ty: ignore[invalid-argument-type]
-        assert deserialized_args.request_id == request_function_call.call_id
-        assert isinstance(deserialized_args.request_event, WorkflowEvent)
-        assert deserialized_args.request_event.type == "request_info"
-        assert deserialized_args.request_event.data == "Mock request data"
-        assert deserialized_args.request_event.response_type is str
-
-        # Verify the request is tracked in pending_requests
-        pending_requests = await workflow._runner_context.get_pending_request_info_events()
-        assert len(pending_requests) == 1
-        assert request_function_call.call_id in pending_requests
+        resolved_event = await agent.resolve_request_info(request_function_call)
+        assert resolved_event is await workflow.get_pending_request_info(request_function_call.call_id)
+        assert await agent.resolve_request_info(request_function_call) is resolved_event
+        assert resolved_event.type == "request_info"
+        assert resolved_event.data == "Mock request data"
+        assert resolved_event.response_type is str
 
         # Now provide a function result response with updated arguments to test continuation
         function_result = Content.from_function_result(
@@ -301,8 +308,183 @@ async def test_end_to_end_request_info_handling(self):
         assert isinstance(continuation_result, AgentResponse)
 
         # Verify cleanup - pending requests should be cleared after function response handling
-        pending_requests = await workflow._runner_context.get_pending_request_info_events()
-        assert len(pending_requests) == 0
+        with pytest.raises(ValueError, match="No pending request-info event found"):
+            await workflow.get_pending_request_info(request_function_call.call_id)
+        for _ in range(2):
+            with pytest.raises(ValueError, match="No pending request-info event found") as exc_info:
+                await agent.resolve_request_info(request_function_call)
+            assert "Mock request data" not in str(exc_info.value)
+
+    @pytest.mark.parametrize(
+        ("tampered_field", "expected_error"),
+        [
+            ("type", "content type"),
+            ("name", "field 'name'"),
+            ("arguments", "field 'arguments'"),
+            ("request_event", "field 'request_event'"),
+        ],
+    )
+    async def test_resolve_request_info_rejects_invalid_function_call_content(
+        self,
+        tampered_field: str,
+        expected_error: str,
+    ) -> None:
+        agent, function_call = await _create_pending_request_info_call()
+        tampered_content = deepcopy(function_call)
+
+        if tampered_field == "type":
+            tampered_content.type = "text"
+        elif tampered_field == "name":
+            tampered_content.name = "other_function"
+        elif tampered_field == "arguments":
+            tampered_content.arguments = None
+        else:
+            arguments = tampered_content.arguments
+            assert isinstance(arguments, dict)
+            cast(dict[str, Any], arguments)["request_event"] = "copied event"
+
+        with pytest.raises(ValueError, match=expected_error):
+            await agent.resolve_request_info(tampered_content)
+
+    async def test_resolve_request_info_rejects_non_content_value(self) -> None:
+        """The resolver distinguishes an invalid object from invalid Content metadata."""
+        agent, _ = await _create_pending_request_info_call()
+
+        with pytest.raises(ValueError, match="must be a Content instance"):
+            await agent.resolve_request_info(cast(Any, {}))
+
+    async def test_resolve_request_info_accepts_json_string_arguments(self) -> None:
+        """The resolver accepts a request-info call restored from JSON transport."""
+        agent, function_call = await _create_pending_request_info_call()
+        function_call.arguments = json.dumps(function_call.arguments)
+
+        resolved_event = await agent.resolve_request_info(function_call)
+
+        assert resolved_event.data == "Mock request data"
+
+    async def test_resolve_request_info_accepts_mapping_arguments(self) -> None:
+        """The resolver accepts read-only mappings restored by a transport adapter."""
+        agent, function_call = await _create_pending_request_info_call()
+        arguments = cast(dict[str, Any], function_call.arguments)
+        request_event = cast(dict[str, Any], arguments["request_event"])
+        function_call.arguments = MappingProxyType({
+            **arguments,
+            "request_event": MappingProxyType(request_event),
+        })
+
+        resolved_event = await agent.resolve_request_info(function_call)
+
+        assert resolved_event.data == "Mock request data"
+
+    @pytest.mark.parametrize("arguments", ["not-json", '["not", "an", "object"]'])
+    async def test_resolve_request_info_rejects_non_object_json_arguments(self, arguments: str) -> None:
+        """Malformed and non-object JSON fail at the transport argument boundary."""
+        agent, function_call = await _create_pending_request_info_call()
+        function_call.arguments = arguments
+
+        with pytest.raises(ValueError, match="field 'arguments' must be a JSON object"):
+            await agent.resolve_request_info(function_call)
+
+    @pytest.mark.parametrize(
+        ("tampering", "expected_error"),
+        [
+            ("missing_call_id", "field 'call_id'.*non-empty string"),
+            ("missing_outer_request_id", "field 'arguments.request_id'.*non-empty string"),
+            ("empty_nested_request_id", "field 'request_event.request_id'.*non-empty string"),
+            ("mismatched_call_id", "correlation IDs.*must match"),
+            ("mismatched_outer_request_id", "correlation IDs.*must match"),
+            ("mismatched_nested_request_id", "correlation IDs.*must match"),
+        ],
+    )
+    async def test_resolve_request_info_rejects_invalid_correlation_ids(
+        self,
+        tampering: str,
+        expected_error: str,
+    ) -> None:
+        agent, function_call = await _create_pending_request_info_call()
+        tampered_content = deepcopy(function_call)
+        arguments = tampered_content.arguments
+        assert isinstance(arguments, dict)
+        arguments_dict = cast(dict[str, Any], arguments)
+        request_event = arguments_dict["request_event"]
+        assert isinstance(request_event, dict)
+
+        if tampering == "missing_call_id":
+            tampered_content.call_id = None
+        elif tampering == "missing_outer_request_id":
+            del arguments_dict["request_id"]
+        elif tampering == "empty_nested_request_id":
+            request_event["request_id"] = ""
+        elif tampering == "mismatched_call_id":
+            tampered_content.call_id = "forged-call-id"
+        elif tampering == "mismatched_outer_request_id":
+            arguments_dict["request_id"] = "forged-outer-id"
+        else:
+            request_event["request_id"] = "forged-nested-id"
+
+        with pytest.raises(ValueError, match=expected_error) as exc_info:
+            await agent.resolve_request_info(tampered_content)
+        assert "Mock request data" not in str(exc_info.value)
+
+    @pytest.mark.parametrize(
+        "metadata_field",
+        [
+            "type",
+            "source_executor_id",
+            "request_type",
+            "response_type",
+        ],
+    )
+    async def test_resolve_request_info_rejects_mismatched_event_metadata(self, metadata_field: str) -> None:
+        agent, function_call = await _create_pending_request_info_call()
+        tampered_content = deepcopy(function_call)
+        arguments = tampered_content.arguments
+        assert isinstance(arguments, dict)
+        request_event = cast(dict[str, Any], arguments)["request_event"]
+        assert isinstance(request_event, dict)
+        request_event[metadata_field] = "forged.metadata"
+
+        with pytest.raises(ValueError, match=rf"request_event\.{metadata_field}") as exc_info:
+            await agent.resolve_request_info(tampered_content)
+        assert "Mock request data" not in str(exc_info.value)
+
+    async def test_resolve_request_info_ignores_copied_request_data(self) -> None:
+        class UntrustedCopiedRequestData:
+            def __eq__(self, other: object) -> bool:
+                raise AssertionError("copied request data must not be compared")
+
+        agent, function_call = await _create_pending_request_info_call()
+        tampered_content = deepcopy(function_call)
+        arguments = tampered_content.arguments
+        assert isinstance(arguments, dict)
+        request_event = cast(dict[str, Any], arguments)["request_event"]
+        assert isinstance(request_event, dict)
+        forged_data = UntrustedCopiedRequestData()
+        request_event["data"] = forged_data
+
+        resolved_event = await agent.resolve_request_info(tampered_content)
+
+        assert resolved_event.data == "Mock request data"
+        assert resolved_event.data is not forged_data
+
+    async def test_resolve_request_info_rejects_unknown_request_through_workflow_lookup(self) -> None:
+        agent, function_call = await _create_pending_request_info_call()
+        tampered_content = deepcopy(function_call)
+        tampered_content.call_id = "unknown-request"
+        arguments = tampered_content.arguments
+        assert isinstance(arguments, dict)
+        arguments_dict = cast(dict[str, Any], arguments)
+        arguments_dict["request_id"] = "unknown-request"
+        request_event = arguments_dict["request_event"]
+        assert isinstance(request_event, dict)
+        request_event["request_id"] = "unknown-request"
+
+        with pytest.raises(
+            ValueError,
+            match="No pending request-info event found for request ID 'unknown-request'",
+        ) as exc_info:
+            await agent.resolve_request_info(tampered_content)
+        assert "Mock request data" not in str(exc_info.value)
 
     def test_request_info_dataclass_arguments_are_serialized_when_content_is_created(self) -> None:
         """Test WorkflowAgent prepares request_info arguments before observability captures messages."""
@@ -328,12 +510,106 @@ def test_request_info_dataclass_arguments_are_serialized_when_content_is_created
         assert deserialize_type(request_event.get("response_type")) is str
         assert request_event.get("data") == HandoffRequest(target_agent="helper", reason="overflow")
 
-        deserialized_args = WorkflowAgent.RequestInfoFunctionArgs.from_dict(request_function_call.arguments)  # ty: ignore[invalid-argument-type]
-        assert deserialized_args.request_id == "request_123"
-        assert isinstance(deserialized_args.request_event, WorkflowEvent)
-        assert deserialized_args.request_event.type == "request_info"
-        assert deserialized_args.request_event.data == HandoffRequest(target_agent="helper", reason="overflow")
-        assert deserialized_args.request_event.response_type is str
+    def test_request_info_generic_response_type_serializes_for_workflow_agent(self) -> None:
+        """WorkflowAgent preserves request-info calls with parameterized generic response types."""
+        executor = SimpleExecutor(id="executor1", response_text="Response")
+        workflow = WorkflowBuilder(start_executor=executor).build()
+        agent = WorkflowAgent(workflow=workflow, name="Request Test Agent")
+        event = WorkflowEvent.request_info(
+            request_id="request_123",
+            source_executor_id="executor1",
+            request_data=HandoffRequest(target_agent="helper", reason="overflow"),
+            response_type=list[Message],
+        )
+
+        request_function_call = agent._process_request_info_event(event)  # pyright: ignore[reportPrivateUsage]
+
+        assert isinstance(request_function_call.arguments, dict)
+        request_event = request_function_call.arguments["request_event"]
+        assert isinstance(request_event, dict)
+        assert request_event["response_type"] == "builtins.list"
+
+    def test_request_info_function_args_rehydrate_accepts_custom_types(self) -> None:
+        """Transport rehydration reconstructs the complete request-info argument envelope."""
+
+        @dataclass
+        class ExplicitRequest:
+            prompt: str
+
+        serialized_name = f"{ExplicitRequest.__module__}.{ExplicitRequest.__qualname__}"
+        args = WorkflowAgent.RequestInfoFunctionArgs.rehydrate(
+            {
+                "request_id": "request-123",
+                "request_event": {
+                    "type": "request_info",
+                    "data": ExplicitRequest(prompt="Approve?"),
+                    "request_id": "request-123",
+                    "source_executor_id": "review_gateway",
+                    "request_type": serialized_name,
+                    "response_type": "builtins.bool",
+                },
+            },
+            allowed_types={serialized_name: ExplicitRequest},
+        )
+
+        assert args.request_id == "request-123"
+        assert type(args.request_event.data) is ExplicitRequest
+        assert args.request_event.request_type is ExplicitRequest
+        assert args.request_event.response_type is bool
+
+    @pytest.mark.parametrize(
+        ("payload", "expected_error"),
+        [
+            ("request_id request_event", "request-info arguments payload must be a mapping"),
+            (
+                {"request_id": "request-123", "request_event": "data request_id source_executor_id request_type"},
+                "request_event'.*must be a mapping",
+            ),
+        ],
+    )
+    def test_request_info_function_args_rehydrate_rejects_non_mapping_payloads(
+        self,
+        payload: object,
+        expected_error: str,
+    ) -> None:
+        """Transport rehydration rejects malformed envelope containers at the interface."""
+        with pytest.raises(ValueError, match=expected_error):
+            WorkflowAgent.RequestInfoFunctionArgs.rehydrate(cast(Any, payload))
+
+    def test_legacy_request_info_function_args_parser_warns_once_and_accepts_custom_types(self) -> None:
+        """Legacy argument parsing warns once and forwards caller-supplied trusted custom types."""
+
+        @dataclass
+        class ExplicitRequest:
+            prompt: str
+
+        serialized_name = f"{ExplicitRequest.__module__}.{ExplicitRequest.__qualname__}"
+        with pytest.warns(
+            DeprecationWarning,
+            match=r"RequestInfoFunctionArgs\.from_dict.*will be removed in a future version"
+            r".*RequestInfoFunctionArgs\.rehydrate",
+        ) as recorded_warnings:
+            args = WorkflowAgent.RequestInfoFunctionArgs.from_dict(
+                {
+                    "request_id": "request-123",
+                    "request_event": {
+                        "type": "request_info",
+                        "data": ExplicitRequest(prompt="Approve?"),
+                        "request_id": "request-123",
+                        "source_executor_id": "review_gateway",
+                        "request_type": serialized_name,
+                        "response_type": "builtins.bool",
+                    },
+                },
+                allowed_types={serialized_name: ExplicitRequest},
+            )
+
+        assert len(recorded_warnings) == 1
+        assert recorded_warnings[0].filename == __file__
+        assert args.request_id == "request-123"
+        assert type(args.request_event.data) is ExplicitRequest
+        assert args.request_event.request_type is ExplicitRequest
+        assert args.request_event.response_type is bool
 
     def test_process_request_info_event_passes_through_function_approval_request(self) -> None:
         """If the event data is already a function approval request, it is forwarded unchanged.
@@ -506,8 +782,8 @@ async def handle_response(
         )
         assert forwarded is approval_request, "Approval request must surface unchanged"
 
-        pending = await workflow._runner_context.get_pending_request_info_events()
-        assert approval_id in pending
+        pending_event = await workflow.get_pending_request_info(approval_id)
+        assert pending_event.data is approval_request
 
         # Respond with approved=True.
         approval_response = approval_request.to_function_approval_response(approved=True)  # type: ignore[attr-defined]
@@ -517,8 +793,8 @@ async def handle_response(
         final_text = " ".join(m.text or "" for m in final.messages)
         assert "delete_file approved=True" in final_text
 
-        pending = await workflow._runner_context.get_pending_request_info_events()
-        assert approval_id not in pending
+        with pytest.raises(ValueError, match="No pending request-info event found"):
+            await workflow.get_pending_request_info(approval_id)
 
     async def test_function_approval_request_flows_end_to_end_denied(self) -> None:
         """End-to-end denied path: ``approved=False`` is delivered to the executor's
@@ -586,8 +862,8 @@ async def handle_response(
         final_text = " ".join(m.text or "" for m in final.messages)
         assert "send_email approved=False" in final_text
 
-        pending = await workflow._runner_context.get_pending_request_info_events()
-        assert approval_id not in pending
+        with pytest.raises(ValueError, match="No pending request-info event found"):
+            await workflow.get_pending_request_info(approval_id)
 
     async def test_request_info_non_approval_flows_end_to_end(self) -> None:
         """End-to-end: when request data is not a function approval content, the
@@ -652,15 +928,11 @@ async def handle_response(
         assert request_payload.get("type") == "request_info"
         assert request_payload.get("data") == HandoffRequest(target_agent="helper", reason="overflow")
 
-        deserialized_args = WorkflowAgent.RequestInfoFunctionArgs.from_dict(function_call.arguments)  # ty: ignore[invalid-argument-type]
-        assert deserialized_args.request_id == request_id
-        assert isinstance(deserialized_args.request_event, WorkflowEvent)
-        assert deserialized_args.request_event.type == "request_info"
-        assert deserialized_args.request_event.data == HandoffRequest(target_agent="helper", reason="overflow")
-        assert deserialized_args.request_event.response_type is str
-
-        pending = await workflow._runner_context.get_pending_request_info_events()
-        assert request_id in pending
+        resolved_event = await agent.resolve_request_info(function_call)
+        assert resolved_event is await workflow.get_pending_request_info(request_id)
+        assert resolved_event.type == "request_info"
+        assert resolved_event.data == HandoffRequest(target_agent="helper", reason="overflow")
+        assert resolved_event.response_type is str
 
         # Respond with a function_result keyed by the call_id.
         function_result = Content.from_function_result(call_id=request_id, result="ok-do-it")
@@ -675,8 +947,8 @@ async def handle_response(
         assert captured["original"].target_agent == "helper"
         assert captured["response"] == "ok-do-it"
 
-        pending = await workflow._runner_context.get_pending_request_info_events()
-        assert request_id not in pending
+        with pytest.raises(ValueError, match="No pending request-info event found"):
+            await agent.resolve_request_info(function_call)
 
     def test_workflow_as_agent_method(self) -> None:
         """Test that Workflow.as_agent() creates a properly configured WorkflowAgent."""
@@ -2180,8 +2452,8 @@ async def start(messages: list[Message], ctx: WorkflowContext[AgentExecutorReque
         assert function_call.arguments == {"path": "/tmp/secret.txt"}
 
         # The agent must be paused awaiting the approval response.
-        pending = await workflow._runner_context.get_pending_request_info_events()
-        assert approval_id in pending
+        pending_event = await workflow.get_pending_request_info(approval_id)
+        assert pending_event.data is approval
 
     async def test_tool_approval_request_forwarded_unchanged_streaming(self) -> None:
         """Streaming variant: the approval request is forwarded as-is in updates."""
@@ -2262,8 +2534,8 @@ async def start(messages: list[Message], ctx: WorkflowContext[AgentExecutorReque
         assert approvals_seen[0].approved is True  # type: ignore[attr-defined]
 
         # The pending approval should now be cleared.
-        pending = await workflow._runner_context.get_pending_request_info_events()
-        assert approval_id not in pending
+        with pytest.raises(ValueError, match="No pending request-info event found"):
+            await workflow.get_pending_request_info(approval_id)
 
         # The final assistant message reflects the resumption.
         final_text = " ".join(m.text or "" for m in final_result.messages)
@@ -2317,8 +2589,8 @@ async def start(messages: list[Message], ctx: WorkflowContext[AgentExecutorReque
         assert approvals_seen[0].approved is False  # type: ignore[attr-defined]
 
         # Pending approval cleared regardless of approve/reject.
-        pending = await workflow._runner_context.get_pending_request_info_events()
-        assert approval_id not in pending
+        with pytest.raises(ValueError, match="No pending request-info event found"):
+            await workflow.get_pending_request_info(approval_id)
 
         # The final assistant message reflects the rejection.
         final_text = " ".join(m.text or "" for m in final_result.messages)
@@ -2348,6 +2620,6 @@ async def start(messages: list[Message], ctx: WorkflowContext[AgentExecutorReque
 
         await agent.run("go")
 
-        pending = await workflow._runner_context.get_pending_request_info_events()
+        pending_event = await workflow.get_pending_request_info(approval_id)
         # The agent's approval id is used as the workflow's pending request id.
-        assert list(pending.keys()) == [approval_id]
+        assert pending_event.request_id == approval_id
diff --git a/python/packages/orchestrations/tests/test_handoff.py b/python/packages/orchestrations/tests/test_handoff.py
index 21964ebddb4..9620b725595 100644
--- a/python/packages/orchestrations/tests/test_handoff.py
+++ b/python/packages/orchestrations/tests/test_handoff.py
@@ -259,6 +259,42 @@ async def test_handoff():
     assert request.source_executor_id == escalation.name
 
 
+async def test_handoff_as_agent_serializes_generic_request_info_response_type() -> None:
+    """Handoff request-info remains usable when the workflow is wrapped as an agent."""
+    triage = MockHandoffAgent(name="triage", handoff_to="specialist")
+    specialist = MockHandoffAgent(name="specialist")
+    workflow = (
+        HandoffBuilder(
+            participants=_as_handoff_agents(triage, specialist),
+            termination_condition=lambda conversation: (
+                sum(1 for message in conversation if message.role == "user") >= 2
+            ),
+        )
+        .with_start_agent(_as_handoff_agent(triage))
+        .build()
+    )
+    agent = workflow.as_agent(name="handoff-agent")
+
+    response = await agent.run("Need technical support")
+
+    request_calls = [
+        content
+        for message in response.messages
+        for content in message.contents
+        if content.type == "function_call" and content.name == agent.REQUEST_INFO_FUNCTION_NAME
+    ]
+    assert len(request_calls) == 1
+    request_call = request_calls[0]
+    assert isinstance(request_call.arguments, dict)
+    request_event = request_call.arguments["request_event"]
+    assert isinstance(request_event, dict)
+    assert request_event["response_type"] == "builtins.list"
+
+    resolved_event = await agent.resolve_request_info(request_call)
+
+    assert resolved_event.response_type == list[Message]
+
+
 def _latest_request_info_event(events: list[WorkflowEvent]) -> WorkflowEvent[Any]:
     request_events = [event for event in events if event.type == "request_info"]
     assert request_events
diff --git a/python/samples/03-workflows/README.md b/python/samples/03-workflows/README.md
index b3cb399dd60..80b506f77a6 100644
--- a/python/samples/03-workflows/README.md
+++ b/python/samples/03-workflows/README.md
@@ -53,7 +53,7 @@ Write workflows as plain Python async functions — no graph concepts, no execut
 | Azure AI Agents (Shared Thread)        | [agents/azure_ai_agents_with_shared_session.py](./agents/azure_ai_agents_with_shared_session.py)                 | Share a common message session between multiple Azure AI agents in a workflow                        |
 | Custom Agent Executors                 | [agents/custom_agent_executors.py](./agents/custom_agent_executors.py)                                         | Create executors to handle agent run methods                                                         |
 | Workflow as Agent (Reflection Pattern) | [agents/workflow_as_agent_reflection_pattern.py](./agents/workflow_as_agent_reflection_pattern.py)             | Wrap a workflow so it can behave like an agent (reflection pattern)                                  |
-| Workflow as Agent + HITL               | [agents/workflow_as_agent_human_in_the_loop.py](./agents/workflow_as_agent_human_in_the_loop.py)               | Extend workflow-as-agent with human-in-the-loop capability                                           |
+| Workflow as Agent + HITL               | [agents/workflow_as_agent_human_in_the_loop.py](./agents/workflow_as_agent_human_in_the_loop.py)               | Resolve authoritative pending request data for workflow-agent human-in-the-loop calls                 |
 | Workflow as Agent with Session         | [agents/workflow_as_agent_with_session.py](./agents/workflow_as_agent_with_session.py)                           | Use AgentSession to maintain conversation history across workflow-as-agent invocations                |
 | Workflow as Agent kwargs               | [agents/workflow_as_agent_kwargs.py](./agents/workflow_as_agent_kwargs.py)                                     | Pass custom context (data, user tokens) via kwargs through workflow.as_agent() to @tool tools |
 
diff --git a/python/samples/03-workflows/agents/handoff_workflow_as_agent.py b/python/samples/03-workflows/agents/handoff_workflow_as_agent.py
index 625344c6c7e..fc9ae1bc849 100644
--- a/python/samples/03-workflows/agents/handoff_workflow_as_agent.py
+++ b/python/samples/03-workflows/agents/handoff_workflow_as_agent.py
@@ -116,29 +116,21 @@ def create_agents(client: FoundryChatClient) -> tuple[Agent, Agent, Agent, Agent
     return triage_agent, refund_agent, order_agent, return_agent
 
 
-def handle_response_and_requests(response: AgentResponse) -> dict[str, HandoffAgentUserRequest]:
-    """Process agent response messages and extract any user requests.
-
-    This function inspects the agent response and:
-    - Displays agent messages to the console
-    - Collects HandoffAgentUserRequest instances for response handling
-
-    Args:
-        response: The AgentResponse from the agent run call.
-
-    Returns:
-        A dictionary mapping request IDs to HandoffAgentUserRequest instances.
-    """
+async def handle_response_and_requests(
+    agent: WorkflowAgent,
+    response: AgentResponse,
+) -> dict[str, HandoffAgentUserRequest]:
+    """Display an agent response and resolve its pending user requests."""
     pending_requests: dict[str, HandoffAgentUserRequest] = {}
     for message in response.messages:
         if message.text:
             print(f"- {message.author_name or message.role}: {message.text}")
         for content in message.contents:
             if content.type == "function_call" and content.name == WorkflowAgent.REQUEST_INFO_FUNCTION_NAME:
-                request_function_args = WorkflowAgent.RequestInfoFunctionArgs.from_dict(content.arguments)  # type: ignore
-                request_id = request_function_args.request_id
-                request_event = request_function_args.request_event
-                pending_requests[request_id] = request_event.data
+                request_event = await agent.resolve_request_info(content)
+                if not isinstance(request_event.data, HandoffAgentUserRequest):
+                    raise ValueError("Handoff request payload must be a HandoffAgentUserRequest.")
+                pending_requests[request_event.request_id] = request_event.data
 
     return pending_requests
 
@@ -205,7 +197,7 @@ async def main() -> None:
     initial_message = "Hello, I need assistance with my recent purchase."
     print(f"- User: {initial_message}")
     response = await agent.run(initial_message)
-    pending_requests = handle_response_and_requests(response)
+    pending_requests = await handle_response_and_requests(agent, response)
 
     # Process the request/response cycle
     # The workflow will continue requesting input until:
@@ -228,7 +220,7 @@ async def main() -> None:
             Content("function_result", call_id=req_id, result=response) for req_id, response in responses.items()
         ]
         response = await agent.run(Message("tool", function_results))
-        pending_requests = handle_response_and_requests(response)
+        pending_requests = await handle_response_and_requests(agent, response)
 
 
 if __name__ == "__main__":
diff --git a/python/samples/03-workflows/agents/workflow_as_agent_human_in_the_loop.py b/python/samples/03-workflows/agents/workflow_as_agent_human_in_the_loop.py
index 54ce8973d87..beb55d5df20 100644
--- a/python/samples/03-workflows/agents/workflow_as_agent_human_in_the_loop.py
+++ b/python/samples/03-workflows/agents/workflow_as_agent_human_in_the_loop.py
@@ -133,14 +133,14 @@ async def main() -> None:
     human_review_function_call: Content | None = None
     for message in response.messages:
         for content in message.contents:
-            if content.name == WorkflowAgent.REQUEST_INFO_FUNCTION_NAME:
+            if content.type == "function_call" and content.name == WorkflowAgent.REQUEST_INFO_FUNCTION_NAME:
                 human_review_function_call = content
 
     # Handle the human review if required.
     if human_review_function_call:
-        # Parse the human review request arguments.
-        human_request_args = WorkflowAgent.RequestInfoFunctionArgs.from_dict(human_review_function_call.arguments)  # type: ignore
-        request_payload = human_request_args.request_event.data
+        # Resolve the complete function call against the workflow's authoritative pending state.
+        human_request_event = await agent.resolve_request_info(human_review_function_call)
+        request_payload = human_request_event.data
         if not isinstance(request_payload, HumanReviewRequest):
             raise ValueError("Human review request payload must be a HumanReviewRequest.")
         if not request_payload.agent_request:
@@ -150,7 +150,7 @@ async def main() -> None:
         # Create the function call result object to send back to the agent.
         human_review_function_result = Content(
             "function_result",
-            call_id=human_review_function_call.call_id,  # type: ignore
+            call_id=human_request_event.request_id,
             result=human_response,
         )
         # Send the human review result back to the agent.
diff --git a/python/samples/03-workflows/orchestrations/README.md b/python/samples/03-workflows/orchestrations/README.md
index eb7d8d477f1..98ca7753499 100644
--- a/python/samples/03-workflows/orchestrations/README.md
+++ b/python/samples/03-workflows/orchestrations/README.md
@@ -62,7 +62,7 @@ from agent_framework.orchestrations import (
 | Handoff (Autonomous)                     | [handoff_autonomous.py](./handoff_autonomous.py)                                                 | Autonomous mode: specialists iterate independently until invoking a handoff tool using `.with_autonomous_mode()` |
 | Handoff with Code Interpreter            | [handoff_with_code_interpreter_file.py](./handoff_with_code_interpreter_file.py)                 | Retrieve file IDs from code interpreter output in handoff workflow                                               |
 | Handoff with Tool Approval + Checkpoint  | [handoff_with_tool_approval_checkpoint_resume.py](./handoff_with_tool_approval_checkpoint_resume.py) | Capture tool-approval decisions in checkpoints and resume from persisted state                                  |
-| Handoff Orchestration as Agent           | [handoff_workflow_as_agent.py](../agents/handoff_workflow_as_agent.py)                                   | Build a HandoffBuilder workflow and expose it as an agent, including HITL request/response flow                |
+| Handoff Orchestration as Agent           | [handoff_workflow_as_agent.py](../agents/handoff_workflow_as_agent.py)                                   | Resolve authoritative pending HITL requests from a HandoffBuilder workflow agent                              |
 
 ### magentic