Python: Add Handoff orchestration pattern support

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

@@ -52,6 +52,7 @@
     handler,
 )
 from ._function_executor import FunctionExecutor, executor
+from ._handoff import HandoffBuilder, HandoffUserInputRequest
 from ._magentic import (
     MagenticAgentDeltaEvent,
     MagenticAgentExecutor,
@@ -126,6 +127,8 @@
     "FileCheckpointStorage",
     "FunctionExecutor",
     "GraphConnectivityError",
+    "HandoffBuilder",
+    "HandoffUserInputRequest",
     "InMemoryCheckpointStorage",
     "InProcRunnerContext",
     "MagenticAgentDeltaEvent",

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

@@ -50,6 +50,7 @@ from ._executor import (
     handler,
 )
 from ._function_executor import FunctionExecutor, executor
+from ._handoff import HandoffBuilder, HandoffUserInputRequest
 from ._magentic import (
     MagenticAgentDeltaEvent,
     MagenticAgentExecutor,
@@ -124,6 +125,8 @@ __all__ = [
     "FileCheckpointStorage",
     "FunctionExecutor",
     "GraphConnectivityError",
+    "HandoffBuilder",
+    "HandoffUserInputRequest",
     "InMemoryCheckpointStorage",
     "InProcRunnerContext",
     "MagenticAgentDeltaEvent",

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

@@ -0,0 +1,77 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from collections.abc import Iterable
+from typing import Any, cast
+
+from agent_framework import ChatMessage, Role
+
+from ._runner_context import _decode_checkpoint_value, _encode_checkpoint_value  # type: ignore
+
+"""Utilities for serializing and deserializing chat conversations for persistence.
+
+These helpers convert rich `ChatMessage` instances to checkpoint-friendly payloads
+using the same encoding primitives as the workflow runner. This preserves
+`additional_properties` and other metadata without relying on unsafe mechanisms
+such as pickling.
+"""
+
+
+def encode_chat_messages(messages: Iterable[ChatMessage]) -> list[dict[str, Any]]:
+    """Serialize chat messages into checkpoint-safe payloads."""
+    encoded: list[dict[str, Any]] = []
+    for message in messages:
+        encoded.append({
+            "role": _encode_checkpoint_value(message.role),
+            "contents": [_encode_checkpoint_value(content) for content in message.contents],
+            "author_name": message.author_name,
+            "message_id": message.message_id,
+            "additional_properties": {
+                key: _encode_checkpoint_value(value) for key, value in message.additional_properties.items()
+            },
+        })
+    return encoded
+
+
+def decode_chat_messages(payload: Iterable[dict[str, Any]]) -> list[ChatMessage]:
+    """Restore chat messages from checkpoint-safe payloads."""
+    restored: list[ChatMessage] = []
+    for item in payload:
+        if not isinstance(item, dict):
+            continue
+
+        role_value = _decode_checkpoint_value(item.get("role"))
+        if isinstance(role_value, Role):
+            role = role_value
+        elif isinstance(role_value, dict):
+            role_dict = cast(dict[str, Any], role_value)
+            role = Role.from_dict(role_dict)
+        elif isinstance(role_value, str):
+            role = Role(value=role_value)
+        else:
+            role = Role.ASSISTANT
+
+        contents_field = item.get("contents", [])
+        contents: list[Any] = []
+        if isinstance(contents_field, list):
+            contents_iter: list[Any] = contents_field  # type: ignore[assignment]
+            for entry in contents_iter:
+                decoded_entry: Any = _decode_checkpoint_value(entry)
+                contents.append(decoded_entry)
+
+        additional_field = item.get("additional_properties", {})
+        additional: dict[str, Any] = {}
+        if isinstance(additional_field, dict):
+            additional_dict = cast(dict[str, Any], additional_field)
+            for key, value in additional_dict.items():
+                additional[key] = _decode_checkpoint_value(value)
+
+        restored.append(
+            ChatMessage(
+                role=role,
+                contents=contents,
+                author_name=item.get("author_name"),
+                message_id=item.get("message_id"),
+                additional_properties=additional,
+            )
+        )
+    return restored