[BREAKING] Python: Replace RequestInfoExecutor with request_info API and @response_handler

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

@@ -75,12 +75,7 @@
     StandardMagenticManager,
 )
 from ._orchestration_state import OrchestrationState
-from ._request_info_executor import (
-    PendingRequestDetails,
-    RequestInfoExecutor,
-    RequestInfoMessage,
-    RequestResponse,
-)
+from ._request_info_mixin import response_handler
 from ._runner import Runner
 from ._runner_context import (
     InProcRunnerContext,
@@ -91,7 +86,6 @@
 from ._shared_state import SharedState
 from ._validation import (
     EdgeDuplicationError,
-    ExecutorDuplicationError,
     GraphConnectivityError,
     TypeCompatibilityError,
     ValidationTypeEnum,
@@ -102,7 +96,7 @@
 from ._workflow import Workflow, WorkflowRunResult
 from ._workflow_builder import WorkflowBuilder
 from ._workflow_context import WorkflowContext
-from ._workflow_executor import WorkflowExecutor
+from ._workflow_executor import SubWorkflowRequestMessage, SubWorkflowResponseMessage, WorkflowExecutor
 
 __all__ = [
     "DEFAULT_MANAGER_INSTRUCTIONS",
@@ -121,7 +115,6 @@
     "EdgeDuplicationError",
     "Executor",
     "ExecutorCompletedEvent",
-    "ExecutorDuplicationError",
     "ExecutorEvent",
     "ExecutorFailedEvent",
     "ExecutorInvokedEvent",
@@ -150,17 +143,15 @@
     "ManagerDirectiveModel",
     "Message",
     "OrchestrationState",
-    "PendingRequestDetails",
     "RequestInfoEvent",
-    "RequestInfoExecutor",
-    "RequestInfoMessage",
-    "RequestResponse",
     "Runner",
     "RunnerContext",
     "SequentialBuilder",
     "SharedState",
     "SingleEdgeGroup",
     "StandardMagenticManager",
+    "SubWorkflowRequestMessage",
+    "SubWorkflowResponseMessage",
     "SwitchCaseEdgeGroup",
     "SwitchCaseEdgeGroupCase",
     "SwitchCaseEdgeGroupDefault",
@@ -189,5 +180,6 @@
     "executor",
     "get_checkpoint_summary",
     "handler",
+    "response_handler",
     "validate_workflow_graph",
 ]

python/packages/core/agent_framework/_workflows/__init__.pyi

@@ -72,12 +72,7 @@ from ._magentic import (
     StandardMagenticManager,
 )
 from ._orchestration_state import OrchestrationState
-from ._request_info_executor import (
-    PendingRequestDetails,
-    RequestInfoExecutor,
-    RequestInfoMessage,
-    RequestResponse,
-)
+from ._request_info_mixin import response_handler
 from ._runner import Runner
 from ._runner_context import (
     InProcRunnerContext,
@@ -88,7 +83,6 @@ from ._sequential import SequentialBuilder
 from ._shared_state import SharedState
 from ._validation import (
     EdgeDuplicationError,
-    ExecutorDuplicationError,
     GraphConnectivityError,
     TypeCompatibilityError,
     ValidationTypeEnum,
@@ -99,7 +93,7 @@ from ._viz import WorkflowViz
 from ._workflow import Workflow, WorkflowRunResult
 from ._workflow_builder import WorkflowBuilder
 from ._workflow_context import WorkflowContext
-from ._workflow_executor import WorkflowExecutor
+from ._workflow_executor import SubWorkflowRequestMessage, SubWorkflowResponseMessage, WorkflowExecutor
 
 __all__ = [
     "DEFAULT_MANAGER_INSTRUCTIONS",
@@ -118,7 +112,6 @@ __all__ = [
     "EdgeDuplicationError",
     "Executor",
     "ExecutorCompletedEvent",
-    "ExecutorDuplicationError",
     "ExecutorEvent",
     "ExecutorFailedEvent",
     "ExecutorInvokedEvent",
@@ -146,17 +139,15 @@ __all__ = [
     "MagenticPlanReviewRequest",
     "Message",
     "OrchestrationState",
-    "PendingRequestDetails",
     "RequestInfoEvent",
-    "RequestInfoExecutor",
-    "RequestInfoMessage",
-    "RequestResponse",
     "Runner",
     "RunnerContext",
     "SequentialBuilder",
     "SharedState",
     "SingleEdgeGroup",
     "StandardMagenticManager",
+    "SubWorkflowRequestMessage",
+    "SubWorkflowResponseMessage",
     "SwitchCaseEdgeGroup",
     "SwitchCaseEdgeGroupCase",
     "SwitchCaseEdgeGroupDefault",
@@ -185,5 +176,6 @@ __all__ = [
     "executor",
     "get_checkpoint_summary",
     "handler",
+    "response_handler",
     "validate_workflow_graph",
 ]

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

@@ -44,6 +44,7 @@ class WorkflowCheckpoint:
     # Core workflow state
     messages: dict[str, list[dict[str, Any]]] = field(default_factory=dict)  # type: ignore[misc]
     shared_state: dict[str, Any] = field(default_factory=dict)  # type: ignore[misc]
+    pending_request_info_events: dict[str, dict[str, Any]] = field(default_factory=dict)  # type: ignore[misc]
 
     # Runtime state
     iteration_count: int = 0

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

@@ -15,7 +15,6 @@
 _MAX_ENCODE_DEPTH = 100
 _CYCLE_SENTINEL = "<cycle>"
 
-
 logger = logging.getLogger(__name__)
 
 
@@ -75,7 +74,7 @@ def _enc(v: Any, stack: set[int], depth: int) -> Any:
                 # type(v) already narrows sufficiently; cast was redundant
                 dc_cls: type[Any] = type(v)
                 field_values: dict[str, Any] = {}
-                for f in fields(v):  # type: ignore[arg-type]
+                for f in fields(v):
                     field_values[f.name] = _enc(getattr(v, f.name), stack, depth + 1)
                 return {
                     DATACLASS_MARKER: f"{dc_cls.__module__}:{dc_cls.__name__}",
@@ -189,6 +188,35 @@ def decode_checkpoint_value(value: Any) -> Any:
     return value
 
 
+def _supports_model_protocol(obj: object) -> bool:
+    """Detect objects that expose dictionary serialization hooks."""
+    try:
+        obj_type: type[Any] = type(obj)
+    except Exception:
+        return False
+
+    has_to_dict = hasattr(obj, "to_dict") and callable(getattr(obj, "to_dict", None))  # type: ignore[arg-type]
+    has_from_dict = hasattr(obj_type, "from_dict") and callable(getattr(obj_type, "from_dict", None))
+
+    has_to_json = hasattr(obj, "to_json") and callable(getattr(obj, "to_json", None))  # type: ignore[arg-type]
+    has_from_json = hasattr(obj_type, "from_json") and callable(getattr(obj_type, "from_json", None))
+
+    return (has_to_dict and has_from_dict) or (has_to_json and has_from_json)
+
+
+def _import_qualified_name(qualname: str) -> type[Any] | None:
+    if ":" not in qualname:
+        return None
+    module_name, class_name = qualname.split(":", 1)
+    module = sys.modules.get(module_name)
+    if module is None:
+        module = importlib.import_module(module_name)
+    attr: Any = module
+    for part in class_name.split("."):
+        attr = getattr(attr, part)
+    return attr if isinstance(attr, type) else None
+
+
 def _instantiate_checkpoint_dataclass(cls: type[Any], payload: Any) -> Any | None:
     if not isinstance(cls, type):
         logger.debug(f"Checkpoint decoder received non-type dataclass reference: {cls!r}")
@@ -220,32 +248,3 @@ def _instantiate_checkpoint_dataclass(cls: type[Any], payload: Any) -> Any | Non
     except Exception as exc:
         logger.warning(f"Checkpoint decoder encountered unexpected error calling {cls.__name__}({payload!r}): {exc}")
     return None
-
-
-def _supports_model_protocol(obj: object) -> bool:
-    """Detect objects that expose dictionary serialization hooks."""
-    try:
-        obj_type: type[Any] = type(obj)
-    except Exception:
-        return False
-
-    has_to_dict = hasattr(obj, "to_dict") and callable(getattr(obj, "to_dict", None))  # type: ignore[arg-type]
-    has_from_dict = hasattr(obj_type, "from_dict") and callable(getattr(obj_type, "from_dict", None))
-
-    has_to_json = hasattr(obj, "to_json") and callable(getattr(obj, "to_json", None))  # type: ignore[arg-type]
-    has_from_json = hasattr(obj_type, "from_json") and callable(getattr(obj_type, "from_json", None))
-
-    return (has_to_dict and has_from_dict) or (has_to_json and has_from_json)
-
-
-def _import_qualified_name(qualname: str) -> type[Any] | None:
-    if ":" not in qualname:
-        return None
-    module_name, class_name = qualname.split(":", 1)
-    module = sys.modules.get(module_name)
-    if module is None:
-        module = importlib.import_module(module_name)
-    attr: Any = module
-    for part in class_name.split("."):
-        attr = getattr(attr, part)
-    return attr if isinstance(attr, type) else None