Python: [Breaking] removed pydantic from types and workflows (#917)

* removed pydantic from types

* fix test

* fix test

* fix tests

* fix assistants client

* Remove Pydantic usage from workflow code.

* updated pydantic removal

* updated lock and test fixes

* fix mypy

* updated build system

* updated chat client parsing

* fix broken test

---------

Co-authored-by: Evan Mattson <evan.mattson@microsoft.com>

python/packages/main/agent_framework/_agents.py

@@ -25,8 +25,8 @@ from ._types import (
     ChatOptions,
     ChatResponse,
     ChatResponseUpdate,
-    ChatToolMode,
     Role,
+    ToolMode,
 )
 from .exceptions import AgentExecutionException
 from .observability import use_agent_observability
@@ -345,11 +345,11 @@ class ChatAgent(BaseAgent):
         stop: str | Sequence[str] | None = None,
         store: bool | None = None,
         temperature: float | None = None,
-        tool_choice: ChatToolMode | Literal["auto", "required", "none"] | dict[str, Any] | None = "auto",
+        tool_choice: ToolMode | Literal["auto", "required", "none"] | dict[str, Any] | None = "auto",
         tools: ToolProtocol
         | Callable[..., Any]
         | MutableMapping[str, Any]
-        | list[ToolProtocol | Callable[..., Any] | MutableMapping[str, Any]]
+        | Sequence[ToolProtocol | Callable[..., Any] | MutableMapping[str, Any]]
         | None = None,
         top_p: float | None = None,
         user: str | None = None,
@@ -412,12 +412,12 @@ class ChatAgent(BaseAgent):
         # We ignore the MCP Servers here and store them separately,
         # we add their functions to the tools list at runtime
         normalized_tools: list[ToolProtocol | Callable[..., Any] | MutableMapping[str, Any]] = (  # type:ignore[reportUnknownVariableType]
-            [] if tools is None else tools if isinstance(tools, list) else [tools]
+            [] if tools is None else tools if isinstance(tools, list) else [tools]  # type: ignore[list-item]
         )
         self._local_mcp_tools = [tool for tool in normalized_tools if isinstance(tool, MCPTool)]
         agent_tools = [tool for tool in normalized_tools if not isinstance(tool, MCPTool)]
         self.chat_options = ChatOptions(
-            ai_model_id=model,
+            model_id=model,
             frequency_penalty=frequency_penalty,
             instructions=instructions,
             logit_bias=logit_bias,
@@ -430,7 +430,7 @@ class ChatAgent(BaseAgent):
             store=store,
             temperature=temperature,
             tool_choice=tool_choice,
-            tools=agent_tools,  # type: ignore[reportArgumentType]
+            tools=agent_tools,
             top_p=top_p,
             user=user,
             additional_properties=request_kwargs or {},  # type: ignore
@@ -489,7 +489,7 @@ class ChatAgent(BaseAgent):
         stop: str | Sequence[str] | None = None,
         store: bool | None = None,
         temperature: float | None = None,
-        tool_choice: ChatToolMode | Literal["auto", "required", "none"] | dict[str, Any] | None = None,
+        tool_choice: ToolMode | Literal["auto", "required", "none"] | dict[str, Any] | None = None,
         tools: ToolProtocol
         | Callable[..., Any]
         | MutableMapping[str, Any]
@@ -558,7 +558,7 @@ class ChatAgent(BaseAgent):
             messages=thread_messages,
             chat_options=run_chat_options
             & ChatOptions(
-                ai_model_id=model,
+                model_id=model,
                 conversation_id=thread.service_thread_id,
                 frequency_penalty=frequency_penalty,
                 logit_bias=logit_bias,
@@ -571,7 +571,7 @@ class ChatAgent(BaseAgent):
                 store=store,
                 temperature=temperature,
                 tool_choice=tool_choice,
-                tools=final_tools,  # type: ignore[reportArgumentType]
+                tools=final_tools,
                 top_p=top_p,
                 user=user,
                 additional_properties=additional_properties or {},
@@ -615,7 +615,7 @@ class ChatAgent(BaseAgent):
         stop: str | Sequence[str] | None = None,
         store: bool | None = None,
         temperature: float | None = None,
-        tool_choice: ChatToolMode | Literal["auto", "required", "none"] | dict[str, Any] | None = None,
+        tool_choice: ToolMode | Literal["auto", "required", "none"] | dict[str, Any] | None = None,
         tools: ToolProtocol
         | Callable[..., Any]
         | MutableMapping[str, Any]
@@ -692,7 +692,7 @@ class ChatAgent(BaseAgent):
                 logit_bias=logit_bias,
                 max_tokens=max_tokens,
                 metadata=metadata,
-                ai_model_id=model,
+                model_id=model,
                 presence_penalty=presence_penalty,
                 response_format=response_format,
                 seed=seed,

python/packages/main/agent_framework/_clients.py

@@ -3,7 +3,7 @@
 import asyncio
 from abc import ABC, abstractmethod
 from collections.abc import AsyncIterable, Callable, MutableMapping, MutableSequence, Sequence
-from typing import TYPE_CHECKING, Any, Generic, Literal, Protocol, TypeVar, runtime_checkable
+from typing import TYPE_CHECKING, Any, Literal, Protocol, TypeVar, runtime_checkable
 
 from pydantic import BaseModel, Field
 
@@ -25,8 +25,7 @@ from ._types import (
     ChatOptions,
     ChatResponse,
     ChatResponseUpdate,
-    ChatToolMode,
-    GeneratedEmbeddings,
+    ToolMode,
 )
 
 if TYPE_CHECKING:
@@ -42,7 +41,6 @@ logger = get_logger()
 __all__ = [
     "BaseChatClient",
     "ChatClientProtocol",
-    "EmbeddingGenerator",
 ]
 
 
@@ -73,7 +71,7 @@ class ChatClientProtocol(Protocol):
         stop: str | Sequence[str] | None = None,
         store: bool | None = None,
         temperature: float | None = None,
-        tool_choice: ChatToolMode | Literal["auto", "required", "none"] | dict[str, Any] | None = "auto",
+        tool_choice: ToolMode | Literal["auto", "required", "none"] | dict[str, Any] | None = "auto",
         tools: ToolProtocol
         | Callable[..., Any]
         | MutableMapping[str, Any]
@@ -130,7 +128,7 @@ class ChatClientProtocol(Protocol):
         stop: str | Sequence[str] | None = None,
         store: bool | None = None,
         temperature: float | None = None,
-        tool_choice: ChatToolMode | Literal["auto", "required", "none"] | dict[str, Any] | None = "auto",
+        tool_choice: ToolMode | Literal["auto", "required", "none"] | dict[str, Any] | None = "auto",
         tools: ToolProtocol
         | Callable[..., Any]
         | MutableMapping[str, Any]
@@ -310,7 +308,7 @@ class BaseChatClient(AFBaseModel, ABC):
         stop: str | Sequence[str] | None = None,
         store: bool | None = None,
         temperature: float | None = None,
-        tool_choice: ChatToolMode | Literal["auto", "required", "none"] | dict[str, Any] | None = "auto",
+        tool_choice: ToolMode | Literal["auto", "required", "none"] | dict[str, Any] | None = "auto",
         tools: ToolProtocol
         | Callable[..., Any]
         | MutableMapping[str, Any]
@@ -354,7 +352,7 @@ class BaseChatClient(AFBaseModel, ABC):
                 raise TypeError("chat_options must be an instance of ChatOptions")
         else:
             chat_options = ChatOptions(
-                ai_model_id=model,
+                model_id=model,
                 frequency_penalty=frequency_penalty,
                 logit_bias=logit_bias,
                 max_tokens=max_tokens,
@@ -392,7 +390,7 @@ class BaseChatClient(AFBaseModel, ABC):
         stop: str | Sequence[str] | None = None,
         store: bool | None = None,
         temperature: float | None = None,
-        tool_choice: ChatToolMode | Literal["auto", "required", "none"] | dict[str, Any] | None = "auto",
+        tool_choice: ToolMode | Literal["auto", "required", "none"] | dict[str, Any] | None = "auto",
         tools: ToolProtocol
         | Callable[..., Any]
         | MutableMapping[str, Any]
@@ -435,7 +433,7 @@ class BaseChatClient(AFBaseModel, ABC):
                 raise TypeError("chat_options must be an instance of ChatOptions")
         else:
             chat_options = ChatOptions(
-                ai_model_id=model,
+                model_id=model,
                 frequency_penalty=frequency_penalty,
                 logit_bias=logit_bias,
                 max_tokens=max_tokens,
@@ -448,7 +446,7 @@ class BaseChatClient(AFBaseModel, ABC):
                 temperature=temperature,
                 top_p=top_p,
                 tool_choice=tool_choice,
-                tools=self._normalize_tools(tools),  # type: ignore
+                tools=self._normalize_tools(tools),
                 user=user,
                 additional_properties=additional_properties or {},
             )
@@ -467,15 +465,15 @@ class BaseChatClient(AFBaseModel, ABC):
         This function should be overridden by subclasses to customize tool handling.
         Because it currently parses only AIFunctions.
         """
-        chat_tool_mode: ChatToolMode | None = chat_options.tool_choice  # type: ignore
-        if chat_tool_mode is None or chat_tool_mode == ChatToolMode.NONE:
+        chat_tool_mode = chat_options.tool_choice
+        if chat_tool_mode is None or chat_tool_mode == ToolMode.NONE or chat_tool_mode == "none":
             chat_options.tools = None
-            chat_options.tool_choice = ChatToolMode.NONE.mode
+            chat_options.tool_choice = ToolMode.NONE.mode
             return
         if not chat_options.tools:
-            chat_options.tool_choice = ChatToolMode.NONE.mode
+            chat_options.tool_choice = ToolMode.NONE.mode
         else:
-            chat_options.tool_choice = chat_tool_mode.mode
+            chat_options.tool_choice = chat_tool_mode.mode if isinstance(chat_tool_mode, ToolMode) else chat_tool_mode
 
     def service_url(self) -> str:
         """Get the URL of the service.
@@ -528,28 +526,3 @@ class BaseChatClient(AFBaseModel, ABC):
             middleware=middleware,
             **kwargs,
         )
-
-
-# region Embedding Client
-
-
-@runtime_checkable
-class EmbeddingGenerator(Protocol, Generic[TInput, TEmbedding]):
-    """A protocol for an embedding generator that can create embeddings from input data."""
-
-    async def generate(
-        self,
-        input_data: Sequence[TInput],
-        **kwargs: Any,
-    ) -> GeneratedEmbeddings[TEmbedding]:
-        """Generates an embedding for the given input data.
-
-        Args:
-            input_data: The input data to generate an embedding for.
-            **kwargs: Additional options for the request.
-
-        Returns:
-            The generated embedding, this acts like a list, but has additional metadata and usage details.
-
-        """
-        ...

python/packages/main/agent_framework/_mcp.py

@@ -347,7 +347,7 @@ class MCPTool:
         return types.CreateMessageResult(
             role="assistant",
             content=mcp_content,
-            model=response.ai_model_id or "unknown",
+            model=response.model_id or "unknown",
         )
 
     async def logging_callback(self, params: types.LoggingMessageNotificationParams) -> None:

python/packages/main/agent_framework/_tools.py

@@ -77,6 +77,14 @@ ArgsT = TypeVar("ArgsT", bound=BaseModel)
 ReturnT = TypeVar("ReturnT")
 
 
+class _NoOpHistogram:
+    def record(self, *args: Any, **kwargs: Any) -> None:  # pragma: no cover - trivial
+        return None
+
+
+_NOOP_HISTOGRAM = _NoOpHistogram()
+
+
 def _parse_inputs(
     inputs: "Contents | dict[str, Any] | str | list[Contents | dict[str, Any] | str] | None",
 ) -> list["Contents"]:
@@ -367,12 +375,24 @@ class HostedFileSearchTool(BaseTool):
 
 def _default_histogram() -> Histogram:
     """Get the default histogram for function invocation duration."""
-    return get_meter().create_histogram(
-        name=OtelAttr.MEASUREMENT_FUNCTION_INVOCATION_DURATION,
-        unit=OtelAttr.DURATION_UNIT,
-        description="Measures the duration of a function's execution",
-        explicit_bucket_boundaries_advisory=OPERATION_DURATION_BUCKET_BOUNDARIES,
-    )
+    from .observability import OBSERVABILITY_SETTINGS  # local import to avoid circulars
+
+    if not OBSERVABILITY_SETTINGS.ENABLED:  # type: ignore[name-defined]
+        return _NOOP_HISTOGRAM  # type: ignore[return-value]
+    meter = get_meter()
+    try:
+        return meter.create_histogram(
+            name=OtelAttr.MEASUREMENT_FUNCTION_INVOCATION_DURATION,
+            unit=OtelAttr.DURATION_UNIT,
+            description="Measures the duration of a function's execution",
+            explicit_bucket_boundaries_advisory=OPERATION_DURATION_BUCKET_BOUNDARIES,
+        )
+    except TypeError:
+        return meter.create_histogram(
+            name=OtelAttr.MEASUREMENT_FUNCTION_INVOCATION_DURATION,
+            unit=OtelAttr.DURATION_UNIT,
+            description="Measures the duration of a function's execution",
+        )
 
 
 class AIFunction(BaseTool, Generic[ArgsT, ReturnT]):

python/packages/main/agent_framework/_workflow/__init__.py

@@ -1,7 +1,5 @@
 # Copyright (c) Microsoft. All rights reserved.
 
-import contextlib
-
 from ._agent import WorkflowAgent
 from ._checkpoint import (
     CheckpointStorage,
@@ -182,8 +180,3 @@ __all__ = [
     "handler",
     "validate_workflow_graph",
 ]
-
-# Rebuild models to resolve forward references after all imports are complete
-with contextlib.suppress(AttributeError, TypeError, ValueError):
-    # Rebuild WorkflowExecutor to resolve Workflow forward reference
-    WorkflowExecutor.model_rebuild()

python/packages/main/agent_framework/_workflow/_agent.py

@@ -1,13 +1,13 @@
 # Copyright (c) Microsoft. All rights reserved.
 
+import json
 import logging
 import uuid
 from collections.abc import AsyncIterable, Sequence
+from dataclasses import dataclass
 from datetime import datetime
 from typing import TYPE_CHECKING, Any, ClassVar, TypedDict, cast
 
-from pydantic import BaseModel
-
 from agent_framework import (
     AgentRunResponse,
     AgentRunResponseUpdate,
@@ -40,10 +40,28 @@ class WorkflowAgent(BaseAgent):
     # Class variable for the request info function name
     REQUEST_INFO_FUNCTION_NAME: ClassVar[str] = "request_info"
 
-    class RequestInfoFunctionArgs(BaseModel):
+    @dataclass
+    class RequestInfoFunctionArgs:
         request_id: str
         data: Any
 
+        def to_dict(self) -> dict[str, Any]:
+            return {"request_id": self.request_id, "data": self.data}
+
+        def to_json(self) -> str:
+            return json.dumps(self.to_dict())
+
+        @classmethod
+        def from_dict(cls, payload: dict[str, Any]) -> "WorkflowAgent.RequestInfoFunctionArgs":
+            return cls(request_id=payload.get("request_id", ""), data=payload.get("data"))
+
+        @classmethod
+        def from_json(cls, raw: str) -> "WorkflowAgent.RequestInfoFunctionArgs":
+            data = json.loads(raw)
+            if not isinstance(data, dict):
+                raise ValueError("RequestInfoFunctionArgs JSON payload must decode to a mapping")
+            return cls.from_dict(data)
+
     def __init__(
         self,
         workflow: "Workflow",
@@ -64,6 +82,7 @@ class WorkflowAgent(BaseAgent):
         """
         if id is None:
             id = f"WorkflowAgent_{uuid.uuid4().hex[:8]}"
+        # Initialize with standard BaseAgent parameters first
         # Validate the workflow's start executor can handle agent-facing message inputs
         try:
             start_executor = workflow.get_start_executor()
@@ -74,9 +93,16 @@ class WorkflowAgent(BaseAgent):
             raise ValueError("Workflow's start executor cannot handle list[ChatMessage]")
 
         super().__init__(id=id, name=name, description=description, **kwargs)
+        self._workflow: "Workflow" = workflow
+        self._pending_requests: dict[str, RequestInfoEvent] = {}
 
-        self.workflow: "Workflow" = workflow
-        self.pending_requests: dict[str, RequestInfoEvent] = {}
+    @property
+    def workflow(self) -> "Workflow":
+        return self._workflow
+
+    @property
+    def pending_requests(self) -> dict[str, RequestInfoEvent]:
+        return self._pending_requests
 
     async def run(
         self,
@@ -240,7 +266,7 @@ class WorkflowAgent(BaseAgent):
                 function_call = FunctionCallContent(
                     call_id=request_id,
                     name=self.REQUEST_INFO_FUNCTION_NAME,
-                    arguments=self.RequestInfoFunctionArgs(request_id=request_id, data=event.data).model_dump(),
+                    arguments=self.RequestInfoFunctionArgs(request_id=request_id, data=event.data).to_dict(),
                 )
                 return AgentRunResponseUpdate(
                     contents=[function_call],

python/packages/main/agent_framework/_workflow/_checkpoint.py

@@ -5,6 +5,7 @@ import json
 import logging
 import os
 import uuid
+from collections.abc import Mapping
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
 from pathlib import Path
@@ -40,7 +41,7 @@ class WorkflowCheckpoint:
         return asdict(self)
 
     @classmethod
-    def from_dict(cls, data: dict[str, Any]) -> "WorkflowCheckpoint":
+    def from_dict(cls, data: Mapping[str, Any]) -> "WorkflowCheckpoint":
         return cls(**data)
 
 

python/packages/main/agent_framework/_workflow/_edge_runner.py

@@ -4,7 +4,8 @@ import asyncio
 import logging
 from abc import ABC, abstractmethod
 from collections import defaultdict
-from typing import Any
+from collections.abc import Callable
+from typing import Any, cast
 
 from ..observability import EdgeGroupDeliveryStatus, OtelAttr, create_edge_group_processing_span
 from ._edge import Edge, EdgeGroup, FanInEdgeGroup, FanOutEdgeGroup, SingleEdgeGroup, SwitchCaseEdgeGroup
@@ -145,9 +146,11 @@ class FanOutEdgeRunner(EdgeRunner):
     def __init__(self, edge_group: FanOutEdgeGroup, executors: dict[str, Executor]) -> None:
         super().__init__(edge_group, executors)
         self._edges = edge_group.edges
-        self._target_ids = edge_group.target_ids
+        self._target_ids = edge_group.target_executor_ids
         self._target_map = {edge.target_id: edge for edge in self._edges}
-        self._selection_func = edge_group.selection_func
+        self._selection_func = cast(
+            Callable[[Any, list[str]], list[str]] | None, getattr(edge_group, "selection_func", None)
+        )
 
     async def send_message(self, message: Message, shared_state: SharedState, ctx: RunnerContext) -> bool:
         """Send a message through all edges in the fan-out edge group."""

python/packages/main/agent_framework/_workflow/_executor.py

@@ -4,6 +4,7 @@ import contextlib
 import functools
 import importlib
 import inspect
+import json
 import logging
 import uuid
 from collections.abc import Awaitable, Callable, Iterable, Mapping, Sequence
@@ -11,10 +12,7 @@ from dataclasses import asdict, dataclass, field, fields, is_dataclass
 from textwrap import shorten
 from typing import Any, ClassVar, Generic, TypeVar, cast
 
-from pydantic import Field
-
 from .._agents import AgentProtocol
-from .._pydantic import AFBaseModel
 from .._threads import AgentThread
 from .._types import AgentRunResponse, AgentRunResponseUpdate, ChatMessage
 from ..observability import create_processing_span
@@ -29,6 +27,7 @@ from ._events import (
     WorkflowErrorDetails,
     _framework_event_origin,  # type: ignore[reportPrivateUsage]
 )
+from ._model_utils import DictConvertible
 from ._runner_context import Message, RunnerContext, _decode_checkpoint_value  # type: ignore
 from ._shared_state import SharedState
 from ._typing_utils import is_instance_of
@@ -63,7 +62,7 @@ class WorkflowCheckpointSummary:
     pending_requests: list[PendingRequestDetails]
 
 
-class Executor(AFBaseModel):
+class Executor(DictConvertible):
     """Base class for all workflow executors that process messages and perform computations.
 
     ## Overview
@@ -192,38 +191,44 @@ class Executor(AFBaseModel):
 
     # Provide a default so static analyzers (e.g., pyright) don't require passing `id`.
     # Runtime still sets a concrete value in __init__.
-    id: str = Field(
-        ...,
-        min_length=1,
-        description="Unique identifier for the executor",
-    )
-    type_: str = Field(default="", alias="type", description="The type of executor, corresponding to the class name")
-
-    def __init__(self, id: str, **kwargs: Any) -> None:
+    def __init__(
+        self,
+        id: str,
+        *,
+        type: str | None = None,
+        type_: str | None = None,
+        defer_discovery: bool = False,
+        **_: Any,
+    ) -> None:
         """Initialize the executor with a unique identifier.
 
         Args:
             id: A unique identifier for the executor.
-            kwargs: Additional keyword arguments. Unused in this implementation.
+            type: The executor type name. If not provided, uses class name.
+            type_: Alternative parameter name for executor type.
+            defer_discovery: If True, defer handler method discovery until later.
+            **_: Additional keyword arguments. Unused in this implementation.
         """
         if not id:
             raise ValueError("Executor ID must be a non-empty string.")
 
-        kwargs.update({"id": id})
-        if "type" not in kwargs and "type_" not in kwargs:
-            kwargs["type_"] = self.__class__.__name__
+        resolved_type = type or type_ or self.__class__.__name__
+        self.id = id
+        self.type = resolved_type
+        self.type_ = resolved_type
 
-        super().__init__(**kwargs)
+        from builtins import type as builtin_type
 
-        self._handlers: dict[type, Callable[[Any, WorkflowContext[Any, Any]], Awaitable[None]]] = {}
+        self._handlers: dict[builtin_type[Any], Callable[[Any, WorkflowContext[Any, Any]], Awaitable[None]]] = {}
         self._handler_specs: list[dict[str, Any]] = []
-        self._discover_handlers()
+        if not defer_discovery:
+            self._discover_handlers()
 
-        if not self._handlers:
-            raise ValueError(
-                f"Executor {self.__class__.__name__} has no handlers defined. "
-                "Please define at least one handler using the @handler decorator."
-            )
+            if not self._handlers:
+                raise ValueError(
+                    f"Executor {self.__class__.__name__} has no handlers defined. "
+                    "Please define at least one handler using the @handler decorator."
+                )
 
     async def execute(
         self,
@@ -455,6 +460,10 @@ class Executor(AFBaseModel):
 
         return list(output_types)
 
+    def to_dict(self) -> dict[str, Any]:
+        """Serialize executor definition for workflow topology export."""
+        return {"id": self.id, "type": self.type}
+
 
 # endregion: Executor
 
@@ -729,15 +738,42 @@ class RequestInfoExecutor(Executor):
                 "value": safe_value,
             }
 
-        model_dump_fn = getattr(request_data, "model_dump", None)
-        if callable(model_dump_fn):
+        to_dict_fn = getattr(request_data, "to_dict", None)
+        if callable(to_dict_fn):
             try:
-                dumped = model_dump_fn(mode="json")
+                dumped = to_dict_fn()
             except TypeError:
-                dumped = model_dump_fn()
+                dumped = to_dict_fn()
             safe_value = self._make_json_safe(dumped)
             return {
-                "kind": "pydantic",
+                "kind": "dict",
+                "type": f"{data_cls.__module__}:{data_cls.__qualname__}",
+                "value": safe_value,
+            }
+
+        to_json_fn = getattr(request_data, "to_json", None)
+        if callable(to_json_fn):
+            try:
+                dumped = to_json_fn()
+            except TypeError:
+                dumped = to_json_fn()
+            converted = dumped
+            if isinstance(dumped, (str, bytes, bytearray)):
+                decoded: str | bytes | bytearray
+                if isinstance(dumped, (bytes, bytearray)):
+                    try:
+                        decoded = dumped.decode()
+                    except Exception:
+                        decoded = dumped
+                else:
+                    decoded = dumped
+                try:
+                    converted = json.loads(decoded)
+                except Exception:
+                    converted = decoded
+            safe_value = self._make_json_safe(converted)
+            return {
+                "kind": "dict" if isinstance(converted, dict) else "json",
                 "type": f"{data_cls.__module__}:{data_cls.__qualname__}",
                 "value": safe_value,
             }
@@ -801,7 +837,7 @@ class RequestInfoExecutor(Executor):
     def _decode_request_data(self, metadata: dict[str, Any]) -> RequestInfoMessage:
         kind = metadata.get("kind")
         type_name = metadata.get("type", "")
-        value = metadata.get("value", {})
+        value: Any = metadata.get("value", {})
         if type_name:
             try:
                 imported = self._import_qualname(type_name)
@@ -823,18 +859,30 @@ class RequestInfoExecutor(Executor):
 
         if kind == "dataclass" and isinstance(value, dict):
             with contextlib.suppress(TypeError):
-                return target_cls(**value)
+                return target_cls(**value)  # type: ignore[arg-type]
 
-        if kind == "pydantic" and isinstance(value, dict):
-            model_validate = getattr(target_cls, "model_validate", None)
-            if callable(model_validate):
-                return cast(RequestInfoMessage, model_validate(value))
+        # Backwards-compat handling for checkpoints that used to store pydantic as "dict"
+        if kind in {"dict", "pydantic", "json"} and isinstance(value, dict):
+            from_dict = getattr(target_cls, "from_dict", None)
+            if callable(from_dict):
+                with contextlib.suppress(Exception):
+                    return cast(RequestInfoMessage, from_dict(value))
+
+        if kind == "json" and isinstance(value, str):
+            from_json = getattr(target_cls, "from_json", None)
+            if callable(from_json):
+                with contextlib.suppress(Exception):
+                    return cast(RequestInfoMessage, from_json(value))
+            with contextlib.suppress(Exception):
+                parsed = json.loads(value)
+                if isinstance(parsed, dict):
+                    return self._decode_request_data({"kind": "dict", "type": type_name, "value": parsed})
 
         if isinstance(value, dict):
             with contextlib.suppress(TypeError):
-                return target_cls(**value)
+                return target_cls(**value)  # type: ignore[arg-type]
             instance = object.__new__(target_cls)
-            instance.__dict__.update(value)
+            instance.__dict__.update(value)  # type: ignore[arg-type]
             return instance
 
         with contextlib.suppress(Exception):
@@ -877,12 +925,37 @@ class RequestInfoExecutor(Executor):
                 return cast(dict[str, Any], data)
             return None
 
-        model_dump = getattr(request, "model_dump", None)
-        if callable(model_dump):
+        to_dict = getattr(request, "to_dict", None)
+        if callable(to_dict):
             try:
-                dump = self._make_json_safe(model_dump(mode="json"))
+                dump = self._make_json_safe(to_dict())
             except TypeError:
-                dump = self._make_json_safe(model_dump())
+                dump = self._make_json_safe(to_dict())
+            if isinstance(dump, dict):
+                return cast(dict[str, Any], dump)
+            return None
+
+        to_json = getattr(request, "to_json", None)
+        if callable(to_json):
+            try:
+                raw = to_json()
+            except TypeError:
+                raw = to_json()
+            converted = raw
+            if isinstance(raw, (str, bytes, bytearray)):
+                decoded: str | bytes | bytearray
+                if isinstance(raw, (bytes, bytearray)):
+                    try:
+                        decoded = raw.decode()
+                    except Exception:
+                        decoded = raw
+                else:
+                    decoded = raw
+                try:
+                    converted = json.loads(decoded)
+                except Exception:
+                    converted = decoded
+            dump = self._make_json_safe(converted)
             if isinstance(dump, dict):
                 return cast(dict[str, Any], dump)
             return None
@@ -900,11 +973,11 @@ class RequestInfoExecutor(Executor):
             return value
         if isinstance(value, Mapping):
             safe_dict: dict[str, Any] = {}
-            for key, val in value.items():
-                safe_dict[str(key)] = self._make_json_safe(val)
+            for key, val in value.items():  # type: ignore[attr-defined]
+                safe_dict[str(key)] = self._make_json_safe(val)  # type: ignore[arg-type]
             return safe_dict
         if isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
-            return [self._make_json_safe(item) for item in value]
+            return [self._make_json_safe(item) for item in value]  # type: ignore[misc]
         return repr(value)
 
     async def has_pending_request(self, request_id: str, ctx: WorkflowContext[Any]) -> bool:
@@ -958,7 +1031,7 @@ class RequestInfoExecutor(Executor):
             shared_pending = None
 
         if isinstance(shared_pending, dict):
-            for key, value in shared_pending.items():
+            for key, value in shared_pending.items():  # type: ignore[attr-defined]
                 if isinstance(key, str) and isinstance(value, dict):
                     combined[key] = cast(dict[str, Any], value)
 
@@ -971,7 +1044,7 @@ class RequestInfoExecutor(Executor):
         if isinstance(state, dict):
             state_pending = state.get("pending_requests")
             if isinstance(state_pending, dict):
-                for key, value in state_pending.items():
+                for key, value in state_pending.items():  # type: ignore[attr-defined]
                     if isinstance(key, str) and isinstance(value, dict) and key not in combined:
                         combined[key] = cast(dict[str, Any], value)
 
@@ -1035,17 +1108,15 @@ class RequestInfoExecutor(Executor):
         details: dict[str, Any],
     ) -> RequestInfoMessage | None:
         try:
-            model_validate = getattr(request_cls, "model_validate", None)
-            if callable(model_validate):
-                return cast(RequestInfoMessage, model_validate(details))
+            from_dict = getattr(request_cls, "from_dict", None)
+            if callable(from_dict):
+                return cast(RequestInfoMessage, from_dict(details))
         except (TypeError, ValueError) as exc:
-            logger.debug(
-                f"RequestInfoExecutor {self.id} validation failed for {request_cls.__name__} via model_validate: {exc}"
-            )
+            logger.debug(f"RequestInfoExecutor {self.id} failed to hydrate {request_cls.__name__} via from_dict: {exc}")
         except Exception as exc:
             logger.warning(
                 f"RequestInfoExecutor {self.id} encountered unexpected error during "
-                f"{request_cls.__name__}.model_validate: {exc}"
+                f"{request_cls.__name__}.from_dict: {exc}"
             )
 
         if is_dataclass(request_cls):
@@ -1100,16 +1171,16 @@ class RequestInfoExecutor(Executor):
 
         shared_map = checkpoint.shared_state.get(RequestInfoExecutor._PENDING_SHARED_STATE_KEY)
         if isinstance(shared_map, Mapping):
-            for request_id, snapshot in shared_map.items():
-                RequestInfoExecutor._merge_snapshot(pending, str(request_id), snapshot)
+            for request_id, snapshot in shared_map.items():  # type: ignore[attr-defined]
+                RequestInfoExecutor._merge_snapshot(pending, str(request_id), snapshot)  # type: ignore[arg-type]
 
         for state in checkpoint.executor_states.values():
             if not isinstance(state, Mapping):
                 continue
             inner = state.get("pending_requests")
             if isinstance(inner, Mapping):
-                for request_id, snapshot in inner.items():
-                    RequestInfoExecutor._merge_snapshot(pending, str(request_id), snapshot)
+                for request_id, snapshot in inner.items():  # type: ignore[attr-defined]
+                    RequestInfoExecutor._merge_snapshot(pending, str(request_id), snapshot)  # type: ignore[arg-type]
 
         for source_id, message_list in checkpoint.messages.items():
             if executor_filter is not None and source_id not in executor_filter:
@@ -1176,19 +1247,19 @@ class RequestInfoExecutor(Executor):
 
         RequestInfoExecutor._apply_update(
             details,
-            prompt=snapshot.get("prompt"),
-            draft=snapshot.get("draft"),
-            iteration=snapshot.get("iteration"),
-            source_executor_id=snapshot.get("source_executor_id"),
+            prompt=snapshot.get("prompt"),  # type: ignore[attr-defined]
+            draft=snapshot.get("draft"),  # type: ignore[attr-defined]
+            iteration=snapshot.get("iteration"),  # type: ignore[attr-defined]
+            source_executor_id=snapshot.get("source_executor_id"),  # type: ignore[attr-defined]
         )
 
-        extra = snapshot.get("details")
+        extra = snapshot.get("details")  # type: ignore[attr-defined]
         if isinstance(extra, Mapping):
             RequestInfoExecutor._apply_update(
                 details,
-                prompt=extra.get("prompt"),
-                draft=extra.get("draft"),
-                iteration=extra.get("iteration"),
+                prompt=extra.get("prompt"),  # type: ignore[attr-defined]
+                draft=extra.get("draft"),  # type: ignore[attr-defined]
+                iteration=extra.get("iteration"),  # type: ignore[attr-defined]
             )
 
     @staticmethod
@@ -1198,17 +1269,17 @@ class RequestInfoExecutor(Executor):
         raw_message: Mapping[str, Any],
     ) -> None:
         if isinstance(payload, RequestResponse):
-            request_id = payload.request_id or RequestInfoExecutor._get_field(payload.original_request, "request_id")
+            request_id = payload.request_id or RequestInfoExecutor._get_field(payload.original_request, "request_id")  # type: ignore[arg-type]
             if not request_id:
                 return
             details = pending.setdefault(request_id, PendingRequestDetails(request_id=request_id))
             RequestInfoExecutor._apply_update(
                 details,
-                prompt=RequestInfoExecutor._get_field(payload.original_request, "prompt"),
-                draft=RequestInfoExecutor._get_field(payload.original_request, "draft"),
-                iteration=RequestInfoExecutor._get_field(payload.original_request, "iteration"),
+                prompt=RequestInfoExecutor._get_field(payload.original_request, "prompt"),  # type: ignore[arg-type]
+                draft=RequestInfoExecutor._get_field(payload.original_request, "draft"),  # type: ignore[arg-type]
+                iteration=RequestInfoExecutor._get_field(payload.original_request, "iteration"),  # type: ignore[arg-type]
                 source_executor_id=raw_message.get("source_id"),
-                original_request=payload.original_request,
+                original_request=payload.original_request,  # type: ignore[arg-type]
             )
         elif isinstance(payload, RequestInfoMessage):
             request_id = getattr(payload, "request_id", None)
@@ -1252,7 +1323,7 @@ class RequestInfoExecutor(Executor):
         if obj is None:
             return None
         if isinstance(obj, Mapping):
-            return obj.get(key)
+            return obj.get(key)  # type: ignore[attr-defined,return-value]
         return getattr(obj, key, None)
 
     @staticmethod

python/packages/main/agent_framework/_workflow/_function_executor.py

@@ -59,17 +59,12 @@ class FunctionExecutor(Executor):
 
         # Initialize parent WITHOUT calling _discover_handlers yet
         # We'll manually set up the attributes first
-        executor_id = id or getattr(func, "__name__", "FunctionExecutor")
-        kwargs = {"id": executor_id, "type": "FunctionExecutor"}
+        executor_id = str(id or getattr(func, "__name__", "FunctionExecutor"))
+        kwargs = {"type": "FunctionExecutor"}
 
-        # Set up the base class attributes manually to avoid _discover_handlers
-        from pydantic import BaseModel
-
-        BaseModel.__init__(self, **kwargs)
-
-        self._handlers: dict[type, Callable[[Any, WorkflowContext[Any]], Any]] = {}
-        self._request_interceptors: dict[type | str, list[dict[str, Any]]] = {}
-        self._handler_specs: list[dict[str, Any]] = []
+        super().__init__(id=executor_id, defer_discovery=True, **kwargs)
+        self._handlers = {}
+        self._handler_specs = []
 
         # Store the original function and whether it has context
         self._original_func = func
@@ -111,6 +106,11 @@ class FunctionExecutor(Executor):
         # Now we can safely call _discover_handlers (it won't find any class-level handlers)
         self._discover_handlers()
 
+        if not self._handlers:
+            raise ValueError(
+                f"FunctionExecutor {self.__class__.__name__} failed to register handler for {func.__name__}"
+            )
+
 
 @overload
 def executor(func: Callable[..., Any]) -> FunctionExecutor: ...

python/packages/main/agent_framework/_workflow/_magentic.py

@@ -8,13 +8,11 @@ import re
 import sys
 from abc import ABC, abstractmethod
 from collections.abc import AsyncIterable, Awaitable, Callable
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from enum import Enum
-from typing import Annotated, Any, Literal, Protocol, TypeVar, Union, cast
+from typing import Any, Literal, Protocol, TypeVar, Union, cast
 from uuid import uuid4
 
-from pydantic import BaseModel, ConfigDict, Field
-
 from agent_framework import (
     AgentProtocol,
     AgentRunResponse,
@@ -26,11 +24,11 @@ from agent_framework import (
     Role,
 )
 from agent_framework._agents import BaseAgent
-from agent_framework._pydantic import AFBaseModel
 
-from ._checkpoint import CheckpointStorage
+from ._checkpoint import CheckpointStorage, WorkflowCheckpoint
 from ._events import WorkflowEvent
 from ._executor import Executor, RequestInfoMessage, RequestResponse, handler
+from ._model_utils import DictConvertible, encode_value
 from ._workflow import Workflow, WorkflowBuilder, WorkflowRunResult
 from ._workflow_context import WorkflowContext
 
@@ -51,6 +49,43 @@ ORCH_MSG_KIND_TASK_LEDGER = "task_ledger"
 ORCH_MSG_KIND_INSTRUCTION = "instruction"
 ORCH_MSG_KIND_NOTICE = "notice"
 
+
+def _message_to_payload(message: ChatMessage) -> Any:
+    if hasattr(message, "to_dict") and callable(getattr(message, "to_dict", None)):
+        with contextlib.suppress(Exception):
+            return message.to_dict()  # type: ignore[attr-defined]
+    if hasattr(message, "to_json") and callable(getattr(message, "to_json", None)):
+        with contextlib.suppress(Exception):
+            json_payload = message.to_json()  # type: ignore[attr-defined]
+            if isinstance(json_payload, str):
+                with contextlib.suppress(Exception):
+                    return json.loads(json_payload)
+            return json_payload
+    if hasattr(message, "__dict__"):
+        return encode_value(message.__dict__)
+    return message
+
+
+def _message_from_payload(payload: Any) -> ChatMessage:
+    if isinstance(payload, ChatMessage):
+        return payload
+    if hasattr(ChatMessage, "from_dict") and isinstance(payload, dict):
+        with contextlib.suppress(Exception):
+            return ChatMessage.from_dict(payload)  # type: ignore[attr-defined,no-any-return]
+    if hasattr(ChatMessage, "from_json") and isinstance(payload, str):
+        with contextlib.suppress(Exception):
+            return ChatMessage.from_json(payload)  # type: ignore[attr-defined,no-any-return]
+    if isinstance(payload, dict):
+        with contextlib.suppress(Exception):
+            return ChatMessage(**payload)  # type: ignore[arg-type]
+    if isinstance(payload, str):
+        with contextlib.suppress(Exception):
+            decoded = json.loads(payload)
+            if isinstance(decoded, dict):
+                return _message_from_payload(decoded)
+    raise TypeError("Unable to reconstruct ChatMessage from payload")
+
+
 # region Unified callback API (developer-facing)
 
 
@@ -275,11 +310,18 @@ def _new_chat_history() -> list[ChatMessage]:
     return []
 
 
+def _new_participant_descriptions() -> dict[str, str]:
+    """Typed default factory for participant descriptions dict to satisfy type checkers."""
+    return {}
+
+
 @dataclass
 class MagenticStartMessage:
     """A message to start a magentic workflow."""
 
-    task: ChatMessage
+    def __init__(self, task: ChatMessage) -> None:
+        """Create the start message."""
+        self.task = task
 
     @classmethod
     def from_string(cls, task_text: str) -> "MagenticStartMessage":
@@ -293,6 +335,16 @@ class MagenticStartMessage:
         """
         return cls(task=ChatMessage(role=Role.USER, text=task_text))
 
+    def to_dict(self) -> dict[str, Any]:
+        """Create a dict representation of the message."""
+        return {"task": self.task.to_dict()}
+
+    @classmethod
+    def from_dict(cls, value: dict[str, Any]) -> "MagenticStartMessage":
+        """Create from a dict."""
+        task = ChatMessage.from_dict(value["task"])
+        return cls(task=task)
+
 
 @dataclass
 class MagenticRequestMessage:
@@ -303,7 +355,6 @@ class MagenticRequestMessage:
     task_context: str = ""
 
 
-@dataclass
 class MagenticResponseMessage:
     """A response message type.
 
@@ -311,9 +362,27 @@ class MagenticResponseMessage:
     or target a specific agent by name.
     """
 
-    body: ChatMessage
-    target_agent: str | None = None  # deliver only to this agent if set
-    broadcast: bool = False  # deliver to all agents if True
+    def __init__(
+        self,
+        body: ChatMessage,
+        target_agent: str | None = None,  # deliver only to this agent if set
+        broadcast: bool = False,  # deliver to all agents if True
+    ) -> None:
+        self.body = body
+        self.target_agent = target_agent
+        self.broadcast = broadcast
+
+    def to_dict(self) -> dict[str, Any]:
+        """Create a dict representation of the message."""
+        return {"body": self.body.to_dict(), "target_agent": self.target_agent, "broadcast": self.broadcast}
+
+    @classmethod
+    def from_dict(cls, value: dict[str, Any]) -> "MagenticResponseMessage":
+        """Create from a dict."""
+        body = ChatMessage.from_dict(value["body"])
+        target_agent = value.get("target_agent")
+        broadcast = value.get("broadcast", False)
+        return cls(body=body, target_agent=target_agent, broadcast=broadcast)
 
 
 @dataclass
@@ -342,21 +411,44 @@ class MagenticPlanReviewReply:
     comments: str | None = None  # guidance for replan if no edited text provided
 
 
-class MagenticTaskLedger(AFBaseModel):
+@dataclass
+class MagenticTaskLedger(DictConvertible):
     """Task ledger for the Standard Magentic manager."""
 
-    facts: Annotated[ChatMessage, Field(description="The facts about the task.")]
-    plan: Annotated[ChatMessage, Field(description="The plan for the task.")]
+    facts: ChatMessage
+    plan: ChatMessage
+
+    def to_dict(self) -> dict[str, Any]:
+        return {"facts": _message_to_payload(self.facts), "plan": _message_to_payload(self.plan)}
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> "MagenticTaskLedger":
+        return cls(
+            facts=_message_from_payload(data.get("facts")),
+            plan=_message_from_payload(data.get("plan")),
+        )
 
 
-class MagenticProgressLedgerItem(AFBaseModel):
+@dataclass
+class MagenticProgressLedgerItem(DictConvertible):
     """A progress ledger item."""
 
     reason: str
     answer: str | bool
 
+    def to_dict(self) -> dict[str, Any]:
+        return {"reason": self.reason, "answer": self.answer}
 
-class MagenticProgressLedger(AFBaseModel):
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> "MagenticProgressLedgerItem":
+        answer_value = data.get("answer")
+        if not isinstance(answer_value, (str, bool)):
+            answer_value = ""  # Default to empty string if not str or bool
+        return cls(reason=data.get("reason", ""), answer=answer_value)
+
+
+@dataclass
+class MagenticProgressLedger(DictConvertible):
     """A progress ledger for tracking workflow progress."""
 
     is_request_satisfied: MagenticProgressLedgerItem
@@ -365,20 +457,61 @@ class MagenticProgressLedger(AFBaseModel):
     next_speaker: MagenticProgressLedgerItem
     instruction_or_question: MagenticProgressLedgerItem
 
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "is_request_satisfied": self.is_request_satisfied.to_dict(),
+            "is_in_loop": self.is_in_loop.to_dict(),
+            "is_progress_being_made": self.is_progress_being_made.to_dict(),
+            "next_speaker": self.next_speaker.to_dict(),
+            "instruction_or_question": self.instruction_or_question.to_dict(),
+        }
 
-class MagenticContext(AFBaseModel):
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> "MagenticProgressLedger":
+        return cls(
+            is_request_satisfied=MagenticProgressLedgerItem.from_dict(data.get("is_request_satisfied", {})),
+            is_in_loop=MagenticProgressLedgerItem.from_dict(data.get("is_in_loop", {})),
+            is_progress_being_made=MagenticProgressLedgerItem.from_dict(data.get("is_progress_being_made", {})),
+            next_speaker=MagenticProgressLedgerItem.from_dict(data.get("next_speaker", {})),
+            instruction_or_question=MagenticProgressLedgerItem.from_dict(data.get("instruction_or_question", {})),
+        )
+
+
+@dataclass
+class MagenticContext(DictConvertible):
     """Context for the Magentic manager."""
 
-    task: Annotated[ChatMessage, Field(description="The task to be completed.")]
-    chat_history: Annotated[list[ChatMessage], Field(description="The chat history to track conversation.")] = Field(
-        default_factory=_new_chat_history
-    )
-    participant_descriptions: Annotated[
-        dict[str, str], Field(description="The descriptions of the participants in the workflow.")
-    ]
-    round_count: Annotated[int, Field(description="The number of rounds completed.")] = 0
-    stall_count: Annotated[int, Field(description="The number of stalls detected.")] = 0
-    reset_count: Annotated[int, Field(description="The number of resets detected.")] = 0
+    task: ChatMessage
+    chat_history: list[ChatMessage] = field(default_factory=_new_chat_history)
+    participant_descriptions: dict[str, str] = field(default_factory=_new_participant_descriptions)
+    round_count: int = 0
+    stall_count: int = 0
+    reset_count: int = 0
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "task": _message_to_payload(self.task),
+            "chat_history": [_message_to_payload(msg) for msg in self.chat_history],
+            "participant_descriptions": dict(self.participant_descriptions),
+            "round_count": self.round_count,
+            "stall_count": self.stall_count,
+            "reset_count": self.reset_count,
+        }
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> "MagenticContext":
+        chat_history_payload = data.get("chat_history", [])
+        history: list[ChatMessage] = []
+        for item in chat_history_payload:
+            history.append(_message_from_payload(item))
+        return cls(
+            task=_message_from_payload(data.get("task")),
+            chat_history=history,
+            participant_descriptions=dict(data.get("participant_descriptions", {})),
+            round_count=data.get("round_count", 0),
+            stall_count=data.get("stall_count", 0),
+            reset_count=data.get("reset_count", 0),
+        )
 
     def reset(self) -> None:
         """Reset the context.
@@ -454,12 +587,15 @@ def _extract_json(text: str) -> dict[str, Any]:
     raise ValueError("Unable to parse JSON from model output.")
 
 
-TModel = TypeVar("TModel", bound=AFBaseModel)
+T = TypeVar("T")
 
 
-def _pd_validate(model: type[TModel], data: dict[str, Any]) -> TModel:
-    """Validate against a Pydantic model and return a typed instance."""
-    return model.model_validate(data)  # type: ignore[attr-defined]
+def _coerce_model(model_cls: type[T], data: dict[str, Any]) -> T:
+    # Use type: ignore to suppress mypy errors for dynamic attribute access
+    # We check with hasattr() first, so this is safe
+    if hasattr(model_cls, "from_dict") and callable(model_cls.from_dict):  # type: ignore[attr-defined]
+        return model_cls.from_dict(data)  # type: ignore[attr-defined,return-value,no-any-return]
+    return model_cls(**data)  # type: ignore[arg-type,call-arg]
 
 
 # endregion Utilities
@@ -467,15 +603,21 @@ def _pd_validate(model: type[TModel], data: dict[str, Any]) -> TModel:
 # region Magentic Manager
 
 
-class MagenticManagerBase(AFBaseModel, ABC):
+class MagenticManagerBase(ABC):
     """Base class for the Magentic One manager."""
 
-    max_stall_count: Annotated[int, Field(description="Max number of stalls before a reset.", ge=0)] = 3
-    max_reset_count: Annotated[int | None, Field(description="Max number of resets allowed.", ge=0)] = None
-    max_round_count: Annotated[int | None, Field(description="Max number of agent responses allowed.", gt=0)] = None
-
-    # Base prompt surface for type safety; concrete managers may override with a str field
-    task_ledger_full_prompt: str = ORCHESTRATOR_TASK_LEDGER_FULL_PROMPT
+    def __init__(
+        self,
+        *,
+        max_stall_count: int = 3,
+        max_reset_count: int | None = None,
+        max_round_count: int | None = None,
+    ) -> None:
+        self.max_stall_count = max_stall_count
+        self.max_reset_count = max_reset_count
+        self.max_round_count = max_round_count
+        # Base prompt surface for type safety; concrete managers may override with a str field.
+        self.task_ledger_full_prompt: str = ORCHESTRATOR_TASK_LEDGER_FULL_PROMPT
 
     @abstractmethod
     async def plan(self, magentic_context: MagenticContext) -> ChatMessage:
@@ -517,28 +659,13 @@ class StandardMagenticManager(MagenticManagerBase):
     - Final answer synthesis
     """
 
-    model_config = ConfigDict(arbitrary_types_allowed=True)
-
-    chat_client: ChatClientProtocol
-    task_ledger: MagenticTaskLedger | None = None
-    instructions: str | None = None
-
-    # Prompts may be overridden if needed
-    task_ledger_facts_prompt: str = ORCHESTRATOR_TASK_LEDGER_FACTS_PROMPT
-    task_ledger_plan_prompt: str = ORCHESTRATOR_TASK_LEDGER_PLAN_PROMPT
-    task_ledger_full_prompt: str = ORCHESTRATOR_TASK_LEDGER_FULL_PROMPT
-    task_ledger_facts_update_prompt: str = ORCHESTRATOR_TASK_LEDGER_FACTS_UPDATE_PROMPT
-    task_ledger_plan_update_prompt: str = ORCHESTRATOR_TASK_LEDGER_PLAN_UPDATE_PROMPT
-    progress_ledger_prompt: str = ORCHESTRATOR_PROGRESS_LEDGER_PROMPT
-    final_answer_prompt: str = ORCHESTRATOR_FINAL_ANSWER_PROMPT
-
-    progress_ledger_retry_count: int = Field(default=3)
+    task_ledger: MagenticTaskLedger | None
 
     def snapshot_state(self) -> dict[str, Any]:
         state = super().snapshot_state()
         if self.task_ledger is not None:
             state = dict(state)
-            state["task_ledger"] = self.task_ledger.model_dump(mode="json")
+            state["task_ledger"] = self.task_ledger.to_dict()
         return state
 
     def restore_state(self, state: dict[str, Any]) -> None:
@@ -546,7 +673,7 @@ class StandardMagenticManager(MagenticManagerBase):
         ledger = state.get("task_ledger")
         if ledger is not None:
             try:
-                self.task_ledger = MagenticTaskLedger.model_validate(ledger)
+                self.task_ledger = MagenticTaskLedger.from_dict(ledger)
             except Exception:  # pragma: no cover - defensive
                 logger.warning("Failed to restore manager task ledger from checkpoint state")
 
@@ -586,42 +713,36 @@ class StandardMagenticManager(MagenticManagerBase):
             max_round_count: Maximum number of rounds allowed.
             progress_ledger_retry_count: Maximum number of retries for the progress ledger.
         """
-        args: dict[str, Any] = {
-            "chat_client": chat_client,
-            "instructions": instructions,
-            "max_stall_count": max_stall_count,
-            "max_reset_count": max_reset_count,
-            "max_round_count": max_round_count,
-        }
+        super().__init__(
+            max_stall_count=max_stall_count,
+            max_reset_count=max_reset_count,
+            max_round_count=max_round_count,
+        )
 
-        # Optional prompt overrides
-        if task_ledger_facts_prompt is not None:
-            args["task_ledger_facts_prompt"] = task_ledger_facts_prompt
-        if task_ledger_plan_prompt is not None:
-            args["task_ledger_plan_prompt"] = task_ledger_plan_prompt
-        if task_ledger_full_prompt is not None:
-            args["task_ledger_full_prompt"] = task_ledger_full_prompt
-        if task_ledger_facts_update_prompt is not None:
-            args["task_ledger_facts_update_prompt"] = task_ledger_facts_update_prompt
-        if task_ledger_plan_update_prompt is not None:
-            args["task_ledger_plan_update_prompt"] = task_ledger_plan_update_prompt
-        if progress_ledger_prompt is not None:
-            args["progress_ledger_prompt"] = progress_ledger_prompt
-        if final_answer_prompt is not None:
-            args["final_answer_prompt"] = final_answer_prompt
-        if progress_ledger_retry_count is not None:
-            args["progress_ledger_retry_count"] = progress_ledger_retry_count
+        self.chat_client: ChatClientProtocol = chat_client
+        self.instructions: str | None = instructions
+        self.task_ledger: MagenticTaskLedger | None = task_ledger
 
-        super().__init__(**args)
+        # Prompts may be overridden if needed
+        self.task_ledger_facts_prompt: str = task_ledger_facts_prompt or ORCHESTRATOR_TASK_LEDGER_FACTS_PROMPT
+        self.task_ledger_plan_prompt: str = task_ledger_plan_prompt or ORCHESTRATOR_TASK_LEDGER_PLAN_PROMPT
+        self.task_ledger_full_prompt = task_ledger_full_prompt or ORCHESTRATOR_TASK_LEDGER_FULL_PROMPT
+        self.task_ledger_facts_update_prompt: str = (
+            task_ledger_facts_update_prompt or ORCHESTRATOR_TASK_LEDGER_FACTS_UPDATE_PROMPT
+        )
+        self.task_ledger_plan_update_prompt: str = (
+            task_ledger_plan_update_prompt or ORCHESTRATOR_TASK_LEDGER_PLAN_UPDATE_PROMPT
+        )
+        self.progress_ledger_prompt: str = progress_ledger_prompt or ORCHESTRATOR_PROGRESS_LEDGER_PROMPT
+        self.final_answer_prompt: str = final_answer_prompt or ORCHESTRATOR_FINAL_ANSWER_PROMPT
 
-        if task_ledger is not None:
-            self.task_ledger = task_ledger
+        self.progress_ledger_retry_count: int = (
+            progress_ledger_retry_count if progress_ledger_retry_count is not None else 3
+        )
 
     async def _complete(
         self,
         messages: list[ChatMessage],
-        *,
-        response_format: type[BaseModel] | None = None,
     ) -> ChatMessage:
         """Call the underlying ChatClientProtocol directly and return the last assistant message.
 
@@ -636,7 +757,7 @@ class StandardMagenticManager(MagenticManagerBase):
         request_messages.extend(messages)
 
         # Invoke the chat client non-streaming API
-        response = await self.chat_client.get_response(request_messages, response_format=response_format)
+        response = await self.chat_client.get_response(request_messages)
         try:
             out_messages: list[ChatMessage] | None = list(response.messages)  # type: ignore[assignment]
         except Exception:
@@ -753,13 +874,10 @@ class StandardMagenticManager(MagenticManagerBase):
         attempts = 0
         last_error: Exception | None = None
         while attempts < self.progress_ledger_retry_count:
-            raw = await self._complete(
-                [*magentic_context.chat_history, user_message],
-                response_format=MagenticProgressLedger,
-            )
+            raw = await self._complete([*magentic_context.chat_history, user_message])
             try:
                 ledger_dict = _extract_json(raw.text)
-                return _pd_validate(MagenticProgressLedger, ledger_dict)
+                return _coerce_model(MagenticProgressLedger, ledger_dict)
             except Exception as ex:
                 last_error = ex
                 attempts += 1
@@ -871,9 +989,9 @@ class MagenticOrchestratorExecutor(Executor):
             "terminated": self._terminated,
         }
         if self._context is not None:
-            state["magentic_context"] = self._context.model_dump(mode="json")
+            state["magentic_context"] = self._context.to_dict()
         if self._task_ledger is not None:
-            state["task_ledger"] = self._task_ledger.model_dump(mode="json")
+            state["task_ledger"] = _message_to_payload(self._task_ledger)
         manager_state: dict[str, Any] | None = None
         with contextlib.suppress(Exception):
             manager_state = self._manager.snapshot_state()
@@ -885,14 +1003,17 @@ class MagenticOrchestratorExecutor(Executor):
         ctx_payload = state.get("magentic_context")
         if ctx_payload is not None:
             try:
-                self._context = MagenticContext.model_validate(ctx_payload)
+                if isinstance(ctx_payload, dict):
+                    self._context = MagenticContext.from_dict(ctx_payload)  # type: ignore[arg-type]
+                else:
+                    self._context = None
             except Exception as exc:  # pragma: no cover - defensive
                 logger.warning("Failed to restore magentic context: %s", exc)
                 self._context = None
         ledger_payload = state.get("task_ledger")
         if ledger_payload is not None:
             try:
-                self._task_ledger = ChatMessage.model_validate(ledger_payload)
+                self._task_ledger = _message_from_payload(ledger_payload)
             except Exception as exc:  # pragma: no cover
                 logger.warning("Failed to restore task ledger message: %s", exc)
                 self._task_ledger = None
@@ -989,7 +1110,7 @@ class MagenticOrchestratorExecutor(Executor):
                 await self._message_callback(self.id, message.task, ORCH_MSG_KIND_USER_TASK)
 
         # Initial planning using the manager with real model calls
-        self._task_ledger = await self._manager.plan(self._context.model_copy(deep=True))
+        self._task_ledger = await self._manager.plan(self._context.clone(deep=True))
 
         # If a human must sign off, ask now and return. The response handler will resume.
         if self._require_plan_signoff:
@@ -1057,7 +1178,7 @@ class MagenticOrchestratorExecutor(Executor):
             return
 
         human = response.data
-        if human is None:
+        if human is None:  # type: ignore[unreachable]
             # Defensive fallback: treat as revise with empty comments
             human = MagenticPlanReviewReply(decision=MagenticPlanReviewDecision.REVISE, comments="")
 
@@ -1089,7 +1210,7 @@ class MagenticOrchestratorExecutor(Executor):
                     ChatMessage(role=Role.USER, text=f"Human plan feedback: {human.comments}")
                 )
                 # Ask the manager to replan based on comments; proceed immediately
-                self._task_ledger = await self._manager.replan(self._context.model_copy(deep=True))
+                self._task_ledger = await self._manager.replan(self._context.clone(deep=True))
 
             # Record the signed-off plan (no broadcast)
             if self._task_ledger:
@@ -1159,7 +1280,7 @@ class MagenticOrchestratorExecutor(Executor):
             )
 
         # Ask the manager to replan; this only adjusts the plan stage, not a full reset
-        self._task_ledger = await self._manager.replan(self._context.model_copy(deep=True))
+        self._task_ledger = await self._manager.replan(self._context.clone(deep=True))
         await self._send_plan_review_request(context)
 
     async def _run_outer_loop(
@@ -1215,7 +1336,7 @@ class MagenticOrchestratorExecutor(Executor):
 
         # Create progress ledger using the manager
         try:
-            current_progress_ledger = await self._manager.create_progress_ledger(ctx.model_copy(deep=True))
+            current_progress_ledger = await self._manager.create_progress_ledger(ctx.clone(deep=True))
         except Exception as ex:
             logger.warning("Magentic Orchestrator: Progress ledger creation failed, triggering reset: %s", ex)
             await self._reset_and_replan(context)
@@ -1298,7 +1419,7 @@ class MagenticOrchestratorExecutor(Executor):
         self._context.reset()
 
         # Replan
-        self._task_ledger = await self._manager.replan(self._context.model_copy(deep=True))
+        self._task_ledger = await self._manager.replan(self._context.clone(deep=True))
 
         # Internally reset all registered agent executors (no handler/messages involved)
         for agent in self._agent_executors.values():
@@ -1317,7 +1438,7 @@ class MagenticOrchestratorExecutor(Executor):
             return
 
         logger.info("Magentic Orchestrator: Preparing final answer")
-        final_answer = await self._manager.prepare_final_answer(self._context.model_copy(deep=True))
+        final_answer = await self._manager.prepare_final_answer(self._context.clone(deep=True))
 
         # Emit a completed event for the workflow
         await context.yield_output(final_answer)
@@ -1412,7 +1533,7 @@ class MagenticAgentExecutor(Executor):
 
     def snapshot_state(self) -> dict[str, Any]:
         return {
-            "chat_history": [msg.model_dump(mode="json") for msg in self._chat_history],
+            "chat_history": [_message_to_payload(msg) for msg in self._chat_history],
         }
 
     def restore_state(self, state: dict[str, Any]) -> None:
@@ -1423,7 +1544,7 @@ class MagenticAgentExecutor(Executor):
         restored: list[ChatMessage] = []
         for item in history_payload:
             try:
-                restored.append(ChatMessage.model_validate(item))
+                restored.append(_message_from_payload(item))
             except Exception as exc:  # pragma: no cover
                 logger.debug("Agent %s: Skipping invalid chat history item during restore: %s", self._agent_id, exc)
         self._chat_history = restored
@@ -1991,7 +2112,7 @@ class MagenticWorkflow:
         if not expected:
             return
 
-        checkpoint = None
+        checkpoint: WorkflowCheckpoint | None = None
         if checkpoint_storage is not None:
             try:
                 checkpoint = await checkpoint_storage.load_checkpoint(checkpoint_id)
@@ -2004,16 +2125,21 @@ class MagenticWorkflow:
             load_checkpoint = getattr(runner_context, "load_checkpoint", None)
             try:
                 if callable(has_checkpointing) and has_checkpointing() and callable(load_checkpoint):
-                    checkpoint = await load_checkpoint(checkpoint_id)  # type: ignore[func-returns-value]
+                    loaded_checkpoint = await load_checkpoint(checkpoint_id)  # type: ignore[misc]
+                    if loaded_checkpoint is not None:
+                        checkpoint = cast(WorkflowCheckpoint, loaded_checkpoint)
             except Exception:  # pragma: no cover - best effort
                 checkpoint = None
 
-        if checkpoint is None or not isinstance(getattr(checkpoint, "executor_states", None), dict):
+        if checkpoint is None:
             return
 
-        orchestrator_state = checkpoint.executor_states.get(getattr(orchestrator, "id", ""))
+        # At this point, checkpoint is guaranteed to be WorkflowCheckpoint
+        executor_states = checkpoint.executor_states
+        orchestrator_id = getattr(orchestrator, "id", "")
+        orchestrator_state = executor_states.get(orchestrator_id)
         if orchestrator_state is None:
-            orchestrator_state = checkpoint.executor_states.get("magentic_orchestrator")
+            orchestrator_state = executor_states.get("magentic_orchestrator")
 
         if not isinstance(orchestrator_state, dict):
             return
@@ -2022,11 +2148,13 @@ class MagenticWorkflow:
         if not isinstance(context_payload, dict):
             return
 
-        restored_participants = context_payload.get("participant_descriptions")
+        context_dict = cast(dict[str, Any], context_payload)
+        restored_participants = context_dict.get("participant_descriptions")
         if not isinstance(restored_participants, dict):
             return
 
-        restored_names = set(restored_participants.keys())
+        participants_dict = cast(dict[str, str], restored_participants)
+        restored_names: set[str] = set(participants_dict.keys())
         expected_names = set(expected.keys())
 
         if restored_names == expected_names:

python/packages/main/agent_framework/_workflow/_model_utils.py

@@ -0,0 +1,51 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import copy
+import sys
+from typing import Any, TypeVar
+
+if sys.version_info >= (3, 11):
+    from typing import Self  # pragma: no cover
+else:
+    from typing_extensions import Self  # pragma: no cover
+
+TModel = TypeVar("TModel", bound="DictConvertible")
+
+
+class DictConvertible:
+    """Mixin providing conversion helpers for plain Python models."""
+
+    def to_dict(self) -> dict[str, Any]:
+        raise NotImplementedError
+
+    @classmethod
+    def from_dict(cls: type[TModel], data: dict[str, Any]) -> TModel:
+        return cls(**data)  # type: ignore[arg-type]
+
+    def clone(self, *, deep: bool = True) -> Self:
+        return copy.deepcopy(self) if deep else copy.copy(self)  # type: ignore[return-value]
+
+    def to_json(self) -> str:
+        import json
+
+        return json.dumps(self.to_dict())
+
+    @classmethod
+    def from_json(cls: type[TModel], raw: str) -> TModel:
+        import json
+
+        data = json.loads(raw)
+        if not isinstance(data, dict):
+            raise ValueError("JSON payload must decode to a mapping")
+        return cls.from_dict(data)
+
+
+def encode_value(value: Any) -> Any:
+    """Recursively encode values for JSON-friendly serialization."""
+    if isinstance(value, DictConvertible):
+        return value.to_dict()
+    if isinstance(value, dict):
+        return {k: encode_value(v) for k, v in value.items()}  # type: ignore[misc]
+    if isinstance(value, (list, tuple, set)):
+        return [encode_value(v) for v in value]  # type: ignore[misc]
+    return value

python/packages/main/agent_framework/_workflow/_runner.py

@@ -6,9 +6,6 @@ from collections import defaultdict
 from collections.abc import AsyncGenerator, Sequence
 from typing import TYPE_CHECKING, Any
 
-if TYPE_CHECKING:
-    from ._executor import RequestInfoExecutor
-
 from ._checkpoint import CheckpointStorage, WorkflowCheckpoint
 from ._edge import EdgeGroup
 from ._edge_runner import EdgeRunner, create_edge_runner
@@ -16,7 +13,7 @@ from ._events import WorkflowEvent, WorkflowOutputEvent, _framework_event_origin
 from ._executor import Executor
 from ._runner_context import (
     _DATACLASS_MARKER,  # type: ignore
-    _PYDANTIC_MARKER,  # type: ignore
+    _MODEL_MARKER,  # type: ignore
     CheckpointState,
     Message,
     RunnerContext,
@@ -24,6 +21,9 @@ from ._runner_context import (
 )
 from ._shared_state import SharedState
 
+if TYPE_CHECKING:
+    from ._executor import RequestInfoExecutor
+
 logger = logging.getLogger(__name__)
 
 
@@ -168,7 +168,7 @@ class Runner:
                 data = message.data
                 if not isinstance(data, dict):
                     return
-                if _PYDANTIC_MARKER not in data and _DATACLASS_MARKER not in data:
+                if _MODEL_MARKER not in data and _DATACLASS_MARKER not in data:
                     return
                 try:
                     decoded = _decode_checkpoint_value(data)
@@ -226,6 +226,7 @@ class Runner:
                             continue
                     except Exception as exc:  # pragma: no cover
                         logger.debug("Terminal completion emission failed: %s", exc)
+
                     logger.warning(
                         f"Message {message} could not be delivered. "
                         "This may be due to type incompatibility or no matching targets."

python/packages/main/agent_framework/_workflow/_runner_context.py

@@ -1,11 +1,12 @@
 # Copyright (c) Microsoft. All rights reserved.
 
 import asyncio
+import contextlib
 import importlib
 import logging
 import sys
 import uuid
-from collections import defaultdict
+from copy import copy
 from dataclasses import dataclass, fields, is_dataclass
 from typing import Any, Protocol, TypedDict, TypeVar, cast, runtime_checkable
 
@@ -53,8 +54,9 @@ class CheckpointState(TypedDict):
 
 
 # Checkpoint serialization helpers
-_PYDANTIC_MARKER = "__af_pydantic_model__"
+_MODEL_MARKER = "__af_model__"
 _DATACLASS_MARKER = "__af_dataclass__"
+_AF_MARKER = "__af__"
 
 # Guards to prevent runaway recursion while encoding arbitrary user data
 _MAX_ENCODE_DEPTH = 100
@@ -78,9 +80,9 @@ def _instantiate_checkpoint_dataclass(cls: type[Any], payload: Any) -> Any | Non
         except Exception as exc:
             logger.debug(f"Checkpoint decoder could not allocate {cls.__name__} without __init__: {exc}")
             return None
-        for key, val in payload.items():
+        for key, val in payload.items():  # type: ignore[attr-defined]
             try:
-                setattr(instance, key, val)
+                setattr(instance, key, val)  # type: ignore[arg-type]
             except Exception as exc:
                 logger.debug(f"Checkpoint decoder could not set attribute {key} on {cls.__name__}: {exc}")
         return instance
@@ -94,22 +96,39 @@ def _instantiate_checkpoint_dataclass(cls: type[Any], payload: Any) -> Any | Non
     return None
 
 
-def _is_pydantic_model(obj: object) -> bool:
-    """Best-effort check for Pydantic models (e.g., AFBaseModel).
-
-    We avoid hard dependencies by duck-typing on model_dump/model_validate.
-    """
+def _supports_model_protocol(obj: object) -> bool:
+    """Detect objects that expose dictionary serialization hooks."""
     try:
         obj_type: type[Any] = type(obj)
-        return hasattr(obj, "model_dump") and hasattr(obj_type, "model_validate")
     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 _encode_checkpoint_value(value: Any) -> Any:
     """Recursively encode values into JSON-serializable structures.
 
-    - Pydantic models -> { _PYDANTIC_MARKER: "module:Class", value: model_dump(mode="json") }
+    - Objects exposing to_dict/to_json -> { _MODEL_MARKER: "module:Class", value: encoded }
     - dataclass instances -> { _DATACLASS_MARKER: "module:Class", value: {field: encoded} }
     - dict -> encode keys as str and values recursively
     - list/tuple/set -> list of encoded items
@@ -124,16 +143,31 @@ def _encode_checkpoint_value(value: Any) -> Any:
             logger.debug(f"Max encode depth reached at depth={depth} for type={type(v)}")
             return "<max_depth>"
 
-        # Pydantic (AFBaseModel) handling
-        if _is_pydantic_model(v):
+        # Structured model handling (objects exposing to_dict/to_json)
+        if _supports_model_protocol(v):
             cls = cast(type[Any], type(v))  # type: ignore
             try:
+                if hasattr(v, "to_dict") and callable(getattr(v, "to_dict", None)):
+                    raw = v.to_dict()  # type: ignore[attr-defined]
+                    strategy = "to_dict"
+                elif hasattr(v, "to_json") and callable(getattr(v, "to_json", None)):
+                    serialized = v.to_json()  # type: ignore[attr-defined]
+                    if isinstance(serialized, (bytes, bytearray)):
+                        try:
+                            serialized = serialized.decode()
+                        except Exception:
+                            serialized = serialized.decode(errors="replace")
+                    raw = serialized
+                    strategy = "to_json"
+                else:
+                    raise AttributeError("Structured model lacks serialization hooks")
                 return {
-                    _PYDANTIC_MARKER: f"{cls.__module__}:{cls.__name__}",
-                    "value": v.model_dump(mode="json"),
+                    _MODEL_MARKER: f"{cls.__module__}:{cls.__name__}",
+                    "strategy": strategy,
+                    "value": _enc(raw, stack, depth + 1),
                 }
             except Exception as exc:  # best-effort fallback
-                logger.debug(f"Pydantic model_dump failed for {cls}: {exc}")
+                logger.debug(f"Structured model serialization failed for {cls}: {exc}")
                 return str(v)
 
         # Dataclasses (instances only)
@@ -205,21 +239,31 @@ def _decode_checkpoint_value(value: Any) -> Any:
     """Recursively decode values previously encoded by _encode_checkpoint_value."""
     if isinstance(value, dict):
         value_dict = cast(dict[str, Any], value)  # encoded form always uses string keys
-        # Pydantic marker handling
-        if _PYDANTIC_MARKER in value_dict and "value" in value_dict:
-            type_key: str | None = value_dict.get(_PYDANTIC_MARKER)  # type: ignore[assignment]
-            raw: Any = value_dict.get("value")
+        # Structured model marker handling
+        if _MODEL_MARKER in value_dict and "value" in value_dict:
+            type_key: str | None = value_dict.get(_MODEL_MARKER)  # type: ignore[assignment]
+            strategy: str | None = value_dict.get("strategy")  # type: ignore[assignment]
+            raw_encoded: Any = value_dict.get("value")
+            decoded_payload = _decode_checkpoint_value(raw_encoded)
             if isinstance(type_key, str):
                 try:
-                    module_name, class_name = type_key.split(":", 1)
-                    module = sys.modules.get(module_name)
-                    if module is None:
-                        module = importlib.import_module(module_name)
-                    cls: Any = getattr(module, class_name)
-                    if hasattr(cls, "model_validate"):
-                        return cls.model_validate(raw)
+                    cls = _import_qualified_name(type_key)
                 except Exception as exc:
-                    logger.debug(f"Failed to decode pydantic model {type_key}: {exc}; returning raw value")
+                    logger.debug(f"Failed to import structured model {type_key}: {exc}")
+                    cls = None
+
+                if cls is not None:
+                    if strategy == "to_dict" and hasattr(cls, "from_dict"):
+                        with contextlib.suppress(Exception):
+                            return cls.from_dict(decoded_payload)
+                    if strategy == "to_json" and hasattr(cls, "from_json"):
+                        if isinstance(decoded_payload, (str, bytes, bytearray)):
+                            with contextlib.suppress(Exception):
+                                return cls.from_json(decoded_payload)
+                        if isinstance(decoded_payload, dict) and hasattr(cls, "from_dict"):
+                            with contextlib.suppress(Exception):
+                                return cls.from_dict(decoded_payload)
+            return decoded_payload
         # Dataclass marker handling
         if _DATACLASS_MARKER in value_dict and "value" in value_dict:
             type_key_dc: str | None = value_dict.get(_DATACLASS_MARKER)  # type: ignore[assignment]
@@ -394,7 +438,7 @@ class InProcRunnerContext:
         Args:
             checkpoint_storage: Optional storage to enable checkpointing.
         """
-        self._messages: defaultdict[str, list[Message]] = defaultdict(list)
+        self._messages: dict[str, list[Message]] = {}
         # Event queue for immediate streaming of events (e.g., AgentRunUpdateEvent)
         self._event_queue: asyncio.Queue[WorkflowEvent] = asyncio.Queue()
 
@@ -407,10 +451,11 @@ class InProcRunnerContext:
         self._max_iterations: int = 100
 
     async def send_message(self, message: Message) -> None:
+        self._messages.setdefault(message.source_id, [])
         self._messages[message.source_id].append(message)
 
     async def drain_messages(self) -> dict[str, list[Message]]:
-        messages = dict(self._messages)
+        messages = copy(self._messages)
         self._messages.clear()
         return messages
 

python/packages/main/agent_framework/_workflow/_workflow.py

@@ -9,10 +9,7 @@ import uuid
 from collections.abc import AsyncIterable, Awaitable, Callable, Sequence
 from typing import Any
 
-from pydantic import Field
-
 from .._agents import AgentProtocol
-from .._pydantic import AFBaseModel
 from ..observability import OtelAttr, capture_exception, create_workflow_span
 from ._agent import WorkflowAgent
 from ._checkpoint import CheckpointStorage
@@ -40,6 +37,7 @@ from ._events import (
     _framework_event_origin,  # type: ignore
 )
 from ._executor import AgentExecutor, Executor, RequestInfoExecutor
+from ._model_utils import DictConvertible
 from ._runner import Runner
 from ._runner_context import InProcRunnerContext, RunnerContext
 from ._shared_state import SharedState
@@ -116,7 +114,7 @@ class WorkflowRunResult(list[WorkflowEvent]):
 # region Workflow
 
 
-class Workflow(AFBaseModel):
+class Workflow(DictConvertible):
     """A graph-based execution engine that orchestrates connected executors.
 
     ## Overview
@@ -167,20 +165,6 @@ class Workflow(AFBaseModel):
     When invoked, the WorkflowExecutor runs the nested workflow to completion and processes its outputs.
     """
 
-    edge_groups: list[EdgeGroup] = Field(
-        default_factory=list, description="List of edge groups that define the workflow edges"
-    )
-    executors: dict[str, Executor] = Field(
-        default_factory=dict, description="Dictionary mapping executor IDs to Executor instances"
-    )
-    start_executor_id: str = Field(min_length=1, description="The ID of the starting executor for the workflow")
-    max_iterations: int = Field(
-        default=DEFAULT_MAX_ITERATIONS, description="Maximum number of iterations the workflow will run"
-    )
-    id: str = Field(
-        default_factory=lambda: str(uuid.uuid4()), description="Unique identifier for this workflow instance"
-    )
-
     def __init__(
         self,
         edge_groups: list[EdgeGroup],
@@ -205,15 +189,11 @@ class Workflow(AFBaseModel):
 
         id = str(uuid.uuid4())
 
-        kwargs.update({
-            "edge_groups": edge_groups,
-            "executors": executors,
-            "start_executor_id": start_executor_id,
-            "max_iterations": max_iterations,
-            "id": id,
-        })
-
-        super().__init__(**kwargs)
+        self.edge_groups = list(edge_groups)
+        self.executors = dict(executors)
+        self.start_executor_id = start_executor_id
+        self.max_iterations = max_iterations
+        self.id = id
 
         # Store non-serializable runtime objects as private attributes
         self._runner_context = runner_context
@@ -246,35 +226,35 @@ class Workflow(AFBaseModel):
         """Reset the running flag."""
         self._is_running = False
 
-    def model_dump(self, **kwargs: Any) -> dict[str, Any]:
-        """Custom serialization that properly handles WorkflowExecutor nested workflows."""
-        data = super().model_dump(**kwargs)
+    def to_dict(self) -> dict[str, Any]:
+        """Serialize the workflow definition into a JSON-ready dictionary."""
+        data: dict[str, Any] = {
+            "id": self.id,
+            "start_executor_id": self.start_executor_id,
+            "max_iterations": self.max_iterations,
+            "edge_groups": [group.to_dict() for group in self.edge_groups],
+            "executors": {executor_id: executor.to_dict() for executor_id, executor in self.executors.items()},
+        }
 
-        # Ensure WorkflowExecutor instances have their workflow field serialized
-        if "executors" in data:
-            executors_data = data["executors"]
-            for executor_id, executor_data in executors_data.items():
-                # Check if this is a WorkflowExecutor that might be missing its workflow field
-                if (
-                    isinstance(executor_data, dict)
-                    and executor_data.get("type") == "WorkflowExecutor"
-                    and "workflow" not in executor_data
-                ):
-                    # Get the original executor object and serialize its workflow
-                    original_executor = self.executors.get(executor_id)
-                    if original_executor and hasattr(original_executor, "workflow"):
-                        from ._workflow_executor import WorkflowExecutor
+        executors_data: dict[str, dict[str, Any]] = data.get("executors", {})
+        for executor_id, executor_payload in executors_data.items():
+            if (
+                isinstance(executor_payload, dict)
+                and executor_payload.get("type") == "WorkflowExecutor"
+                and "workflow" not in executor_payload
+            ):
+                original_executor = self.executors.get(executor_id)
+                if original_executor and hasattr(original_executor, "workflow"):
+                    from ._workflow_executor import WorkflowExecutor
 
-                        if isinstance(original_executor, WorkflowExecutor):
-                            executor_data["workflow"] = original_executor.workflow.model_dump(**kwargs)
+                    if isinstance(original_executor, WorkflowExecutor):
+                        executor_payload["workflow"] = original_executor.workflow.to_dict()
 
         return data
 
-    def model_dump_json(self, **kwargs: Any) -> str:
-        """Custom JSON serialization that properly handles WorkflowExecutor nested workflows."""
-        import json
-
-        return json.dumps(self.model_dump(**kwargs))
+    def to_json(self) -> str:
+        """Serialize the workflow definition to JSON."""
+        return json.dumps(self.to_dict())
 
     def get_start_executor(self) -> Executor:
         """Get the starting executor of the workflow.
@@ -567,7 +547,7 @@ class Workflow(AFBaseModel):
             self._reset_running_flag()
 
         # Coalesce streaming update events into a single AgentRunEvent per executor sequence.
-        coalesced: list[WorkflowEvent] = []  # type: ignore[name-defined]
+        coalesced: list[WorkflowEvent] = []
         pending_updates: list[AgentRunResponseUpdate] = []
         pending_executor: str | None = None
         status_events: list[WorkflowStatusEvent] = []
@@ -810,7 +790,7 @@ class Workflow(AFBaseModel):
             }
 
             if isinstance(group, FanOutEdgeGroup):
-                group_info["selection_func"] = group.selection_func_name
+                group_info["selection_func"] = getattr(group, "selection_func_name", None)
 
             edge_groups_signature.append(group_info)
 
@@ -975,7 +955,7 @@ class WorkflowBuilder:
         target_exec = self._maybe_wrap_agent(target)
         source_id = self._add_executor(source_exec)
         target_id = self._add_executor(target_exec)
-        self._edge_groups.append(SingleEdgeGroup(source_id, target_id, condition))
+        self._edge_groups.append(SingleEdgeGroup(source_id, target_id, condition))  # type: ignore[call-arg]
         return self
 
     def add_fan_out_edges(
@@ -995,7 +975,7 @@ class WorkflowBuilder:
         target_execs = [self._maybe_wrap_agent(t) for t in targets]
         source_id = self._add_executor(source_exec)
         target_ids = [self._add_executor(t) for t in target_execs]
-        self._edge_groups.append(FanOutEdgeGroup(source_id, target_ids))
+        self._edge_groups.append(FanOutEdgeGroup(source_id, target_ids))  # type: ignore[call-arg]
 
         return self
 
@@ -1033,7 +1013,7 @@ class WorkflowBuilder:
                 internal_cases.append(SwitchCaseEdgeGroupDefault(target_id=case.target.id))
             else:
                 internal_cases.append(SwitchCaseEdgeGroupCase(condition=case.condition, target_id=case.target.id))
-        self._edge_groups.append(SwitchCaseEdgeGroup(source_id, internal_cases))
+        self._edge_groups.append(SwitchCaseEdgeGroup(source_id, internal_cases))  # type: ignore[call-arg]
 
         return self
 
@@ -1061,7 +1041,7 @@ class WorkflowBuilder:
         target_execs = [self._maybe_wrap_agent(t) for t in targets]
         source_id = self._add_executor(source_exec)
         target_ids = [self._add_executor(t) for t in target_execs]
-        self._edge_groups.append(FanOutEdgeGroup(source_id, target_ids, selection_func))
+        self._edge_groups.append(FanOutEdgeGroup(source_id, target_ids, selection_func))  # type: ignore[call-arg]
 
         return self
 
@@ -1107,7 +1087,7 @@ class WorkflowBuilder:
         target_exec = self._maybe_wrap_agent(target)
         source_ids = [self._add_executor(s) for s in source_execs]
         target_id = self._add_executor(target_exec)
-        self._edge_groups.append(FanInEdgeGroup(source_ids, target_id))
+        self._edge_groups.append(FanInEdgeGroup(source_ids, target_id))  # type: ignore[call-arg]
 
         return self
 
@@ -1209,7 +1189,7 @@ class WorkflowBuilder:
                 )
                 span.set_attributes({
                     OtelAttr.WORKFLOW_ID: workflow.id,
-                    OtelAttr.WORKFLOW_DEFINITION: workflow.model_dump_json(by_alias=True),
+                    OtelAttr.WORKFLOW_DEFINITION: workflow.to_json(),
                 })
 
                 # Add workflow build completed event

python/packages/main/agent_framework/_workflow/_workflow_context.py

@@ -1,7 +1,5 @@
 # Copyright (c) Microsoft. All rights reserved.
 
-from __future__ import annotations
-
 import inspect
 import logging
 from collections.abc import Callable

python/packages/main/agent_framework/_workflow/_workflow_executor.py

@@ -11,8 +11,6 @@ from typing import TYPE_CHECKING, Any
 if TYPE_CHECKING:
     from ._workflow import Workflow
 
-from pydantic import Field
-
 from ._events import (
     RequestInfoEvent,
     WorkflowErrorEvent,
@@ -199,8 +197,6 @@ class WorkflowExecutor(Executor):
     - Concurrent executions are fully isolated and do not interfere with each other
     """
 
-    workflow: "Workflow" = Field(description="The workflow to execute as a sub-workflow")
-
     def __init__(self, workflow: "Workflow", id: str, **kwargs: Any):
         """Initialize the WorkflowExecutor.
 
@@ -209,8 +205,8 @@ class WorkflowExecutor(Executor):
             id: Unique identifier for this executor.
             **kwargs: Additional keyword arguments passed to the parent constructor.
         """
-        kwargs.update({"workflow": workflow})
         super().__init__(id, **kwargs)
+        self.workflow = workflow
 
         # Track execution contexts for concurrent sub-workflow executions
         self._execution_contexts: dict[str, ExecutionContext] = {}  # execution_id -> ExecutionContext
@@ -264,6 +260,11 @@ class WorkflowExecutor(Executor):
 
         return output_types
 
+    def to_dict(self) -> dict[str, Any]:
+        data = super().to_dict()
+        data["workflow"] = self.workflow.to_dict()
+        return data
+
     def can_handle(self, message: Any) -> bool:
         """Override can_handle to only accept messages that the wrapped workflow can handle.
 

python/packages/main/agent_framework/azure/__init__.pyi

@@ -0,0 +1,19 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from agent_framework_azure_ai import AzureAIAgentClient, AzureAISettings
+
+from agent_framework.azure._assistants_client import AzureOpenAIAssistantsClient
+from agent_framework.azure._chat_client import AzureOpenAIChatClient
+from agent_framework.azure._entra_id_authentication import get_entra_auth_token
+from agent_framework.azure._responses_client import AzureOpenAIResponsesClient
+from agent_framework.azure._shared import AzureOpenAISettings
+
+__all__ = [
+    "AzureAIAgentClient",
+    "AzureAISettings",
+    "AzureOpenAIAssistantsClient",
+    "AzureOpenAIChatClient",
+    "AzureOpenAIResponsesClient",
+    "AzureOpenAISettings",
+    "get_entra_auth_token",
+]

python/packages/main/agent_framework/observability.py

@@ -564,7 +564,11 @@ def get_meter(
         schema_url: Optional. Specifies the Schema URL of the emitted telemetry.
         attributes: Optional. Attributes that are associated with the emitted telemetry.
     """
-    return metrics.get_meter(name=name, version=version, schema_url=schema_url, attributes=attributes)
+    try:
+        return metrics.get_meter(name=name, version=version, schema_url=schema_url, attributes=attributes)
+    except TypeError:
+        # Older OpenTelemetry releases do not support the attributes parameter.
+        return metrics.get_meter(name=name, version=version, schema_url=schema_url)
 
 
 global OBSERVABILITY_SETTINGS
@@ -772,7 +776,11 @@ def _trace_get_response(
                 self.additional_properties["token_usage_histogram"] = _get_token_usage_histogram()
             if "operation_duration_histogram" not in self.additional_properties:
                 self.additional_properties["operation_duration_histogram"] = _get_duration_histogram()
-            model_id = str(kwargs.get("ai_model_id") or getattr(self, "ai_model_id", "unknown"))
+            model_id = (
+                kwargs.get("model")
+                or (chat_options.model_id if (chat_options := kwargs.get("chat_options")) else None)
+                or getattr(self, "model_id", None)
+            )
             service_url = str(
                 service_url_func()
                 if (service_url_func := getattr(self, "service_url", None)) and callable(service_url_func)
@@ -853,7 +861,11 @@ def _trace_get_streaming_response(
             if "operation_duration_histogram" not in self.additional_properties:
                 self.additional_properties["operation_duration_histogram"] = _get_duration_histogram()
 
-            model_id = kwargs.get("ai_model_id") or getattr(self, "ai_model_id", None)
+            model_id = (
+                kwargs.get("model")
+                or (chat_options.model_id if (chat_options := kwargs.get("chat_options")) else None)
+                or getattr(self, "model_id", None)
+            )
             service_url = str(
                 service_url_func()
                 if (service_url_func := getattr(self, "service_url", None)) and callable(service_url_func)
@@ -1244,7 +1256,7 @@ def _capture_messages(
     for index, message in enumerate(prepped):
         otel_messages.append(_to_otel_message(message))
         try:
-            message_data = message.model_dump(exclude_none=True)
+            message_data = message.to_dict(exclude_none=True)
         except Exception:
             message_data = {"role": message.role.value, "contents": message.contents}
         logger.info(
@@ -1298,7 +1310,7 @@ def _to_otel_part(content: "Contents") -> dict[str, Any] | None:
         case _:
             # GenericPart in otel output messages json spec.
             # just required type, and arbitrary other fields.
-            return content.model_dump(exclude_none=True)
+            return content.to_dict(exclude_none=True)
     return None
 
 
@@ -1317,8 +1329,8 @@ def _get_response_attributes(
         )
     if finish_reason:
         attributes[OtelAttr.FINISH_REASONS] = json.dumps([finish_reason.value])
-    if ai_model_id := getattr(response, "ai_model_id", None):
-        attributes[SpanAttributes.LLM_RESPONSE_MODEL] = ai_model_id
+    if model_id := getattr(response, "model_id", None):
+        attributes[SpanAttributes.LLM_RESPONSE_MODEL] = model_id
     if usage := response.usage_details:
         if usage.input_token_count:
             attributes[OtelAttr.INPUT_TOKENS] = usage.input_token_count

python/packages/main/agent_framework/openai/_assistants_client.py

@@ -28,12 +28,12 @@ from .._types import (
     ChatOptions,
     ChatResponse,
     ChatResponseUpdate,
-    ChatToolMode,
     Contents,
     FunctionCallContent,
     FunctionResultContent,
     Role,
     TextContent,
+    ToolMode,
     UriContent,
     UsageContent,
     UsageDetails,
@@ -115,7 +115,7 @@ class OpenAIAssistantsClient(OpenAIConfigMixin, BaseChatClient):
         if not openai_settings.chat_model_id:
             raise ServiceInitializationError(
                 "OpenAI model ID is required. "
-                "Set via 'ai_model_id' parameter or 'OPENAI_CHAT_MODEL_ID' environment variable."
+                "Set via 'model_id' parameter or 'OPENAI_CHAT_MODEL_ID' environment variable."
             )
 
         super().__init__(
@@ -361,7 +361,7 @@ class OpenAIAssistantsClient(OpenAIConfigMixin, BaseChatClient):
 
         if chat_options is not None:
             run_options["max_completion_tokens"] = chat_options.max_tokens
-            run_options["model"] = chat_options.ai_model_id
+            run_options["model"] = chat_options.model_id
             run_options["top_p"] = chat_options.top_p
             run_options["temperature"] = chat_options.temperature
 
@@ -392,7 +392,7 @@ class OpenAIAssistantsClient(OpenAIConfigMixin, BaseChatClient):
                 if chat_options.tool_choice == "none" or chat_options.tool_choice == "auto":
                     run_options["tool_choice"] = chat_options.tool_choice
                 elif (
-                    isinstance(chat_options.tool_choice, ChatToolMode)
+                    isinstance(chat_options.tool_choice, ToolMode)
                     and chat_options.tool_choice == "required"
                     and chat_options.tool_choice.required_function_name is not None
                 ):

python/packages/main/agent_framework/openai/_chat_client.py

@@ -217,7 +217,7 @@ class OpenAIBaseChatClient(OpenAIBase, BaseChatClient):
             return ChatResponseUpdate(
                 role=Role.ASSISTANT,
                 contents=[UsageContent(details=self._usage_details_from_openai(chunk.usage), raw_representation=chunk)],
-                ai_model_id=chunk.model,
+                model_id=chunk.model,
                 additional_properties=chunk_metadata,
                 response_id=chunk.id,
                 message_id=chunk.id,
@@ -236,7 +236,7 @@ class OpenAIBaseChatClient(OpenAIBase, BaseChatClient):
             created_at=datetime.fromtimestamp(chunk.created).strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
             contents=contents,
             role=Role.ASSISTANT,
-            ai_model_id=chunk.model,
+            model_id=chunk.model,
             additional_properties=chunk_metadata,
             finish_reason=finish_reason,
             raw_representation=chunk,
@@ -402,8 +402,8 @@ class OpenAIBaseChatClient(OpenAIBase, BaseChatClient):
                 elif content.media_type and "mp3" in content.media_type:
                     audio_format = "mp3"
                 else:
-                    # Fallback to default model_dump for unsupported audio formats
-                    return content.model_dump(exclude_none=True)
+                    # Fallback to default to_dict for unsupported audio formats
+                    return content.to_dict(exclude_none=True)
 
                 # Extract base64 data from data URI
                 audio_data = content.uri
@@ -435,11 +435,11 @@ class OpenAIBaseChatClient(OpenAIBase, BaseChatClient):
                             },
                         }
 
-                    return content.model_dump(exclude_none=True)
+                    return content.to_dict(exclude_none=True)
 
-                return content.model_dump(exclude_none=True)
+                return content.to_dict(exclude_none=True)
             case _:
-                return content.model_dump(exclude_none=True)
+                return content.to_dict(exclude_none=True)
 
     @override
     def service_url(self) -> str:

python/packages/main/agent_framework/openai/_responses_client.py

@@ -905,7 +905,7 @@ class OpenAIBaseResponsesClient(OpenAIBase, BaseChatClient):
             contents=contents,
             conversation_id=conversation_id,
             role=Role.ASSISTANT,
-            ai_model_id=model,
+            model_id=model,
             additional_properties=metadata,
             raw_representation=event,
         )

python/packages/main/agent_framework/openai/_shared.py

@@ -17,13 +17,13 @@ from openai.types.chat import ChatCompletion, ChatCompletionChunk
 from openai.types.images_response import ImagesResponse
 from openai.types.responses.response import Response
 from openai.types.responses.response_stream_event import ResponseStreamEvent
-from pydantic import BaseModel, ConfigDict, Field, SecretStr, validate_call
+from pydantic import ConfigDict, Field, SecretStr, validate_call
 from pydantic.types import StringConstraints
 
 from .._logging import get_logger
 from .._pydantic import AFBaseModel, AFBaseSettings
 from .._telemetry import APP_INFO, USER_AGENT_KEY, prepend_agent_framework_to_user_agent
-from .._types import ChatOptions, Contents, SpeechToTextOptions, TextToSpeechOptions
+from .._types import ChatOptions, Contents
 from ..exceptions import ServiceInitializationError
 
 logger: logging.Logger = get_logger("agent_framework.openai")
@@ -42,7 +42,7 @@ RESPONSE_TYPE = Union[
     _legacy_response.HttpxBinaryResponseContent,
 ]
 
-OPTION_TYPE = Union[ChatOptions, SpeechToTextOptions, TextToSpeechOptions, dict[str, Any]]
+OPTION_TYPE = Union[ChatOptions, dict[str, Any]]
 
 
 __all__ = [
@@ -52,20 +52,20 @@ __all__ = [
 
 def _prepare_function_call_results_as_dumpable(content: Contents | Any | list[Contents | Any]) -> Any:
     if isinstance(content, list):
-        # Particularly deal with lists of BaseModel
+        # Particularly deal with lists of Content
         return [_prepare_function_call_results_as_dumpable(item) for item in content]
     if isinstance(content, dict):
         return {k: _prepare_function_call_results_as_dumpable(v) for k, v in content.items()}
-    if isinstance(content, BaseModel):
-        return content.model_dump(exclude={"raw_representation", "additional_properties"})
+    if hasattr(content, "to_dict"):
+        return content.to_dict(exclude={"raw_representation", "additional_properties"})
     return content
 
 
 def prepare_function_call_results(content: Contents | Any | list[Contents | Any]) -> str | list[str]:
     """Prepare the values of the function call results."""
-    if isinstance(content, BaseModel):
-        # BaseModel is already dumpable, shortcut for performance
-        return content.model_dump_json(exclude={"raw_representation", "additional_properties"})
+    if isinstance(content, Contents):
+        # For BaseContent objects, use to_dict and serialize to JSON
+        return json.dumps(content.to_dict(exclude={"raw_representation", "additional_properties"}))
 
     dumpable = _prepare_function_call_results_as_dumpable(content)
     if isinstance(dumpable, str):