Python: Align HandoffBuilder participant type signatures with runtime Agent-only requirement

python/packages/orchestrations/agent_framework_orchestrations/_handoff.py

@@ -197,7 +197,7 @@ class HandoffAgentExecutor(AgentExecutor):
 
     def __init__(
         self,
-        agent: SupportsAgentRun,
+        agent: Agent,
         handoffs: Sequence[HandoffConfiguration],
         *,
         agent_session: AgentSession | None = None,
@@ -210,7 +210,7 @@ def __init__(
         """Initialize the HandoffAgentExecutor.
 
         Args:
-            agent: The agent to execute
+            agent: The ``Agent`` instance to execute
             handoffs: Sequence of handoff configurations defining target agents
             agent_session: Optional AgentSession that manages the agent's execution context
             is_start_agent: Whether this agent is the starting agent in the handoff workflow.
@@ -240,20 +240,18 @@ def __init__(
 
     def _prepare_agent_with_handoffs(
         self,
-        agent: SupportsAgentRun,
+        agent: Agent,
         handoffs: Sequence[HandoffConfiguration],
-    ) -> SupportsAgentRun:
+    ) -> Agent:
         """Prepare an agent by adding handoff tools for the specified target agents.
 
         Args:
-            agent: The agent to prepare
+            agent: The ``Agent`` instance to prepare
             handoffs: Sequence of handoff configurations defining target agents
 
         Returns:
-            A new AgentExecutor instance with handoff tools added
+            A cloned ``Agent`` instance with handoff tools added
         """
-        if not isinstance(agent, Agent):
-            raise TypeError("Handoff can only be applied to Agent. Please ensure the agent is a Agent instance.")
 
         # Clone the agent to avoid mutating the original
         cloned_agent = self._clone_chat_agent(agent)  # type: ignore
@@ -701,13 +699,15 @@ class HandoffBuilder:
     approach to multi-agent collaboration. Handoffs can be configured using `.add_handoff`. If
     none are specified, all agents can hand off to all others by default (making a mesh topology).
 
-    Participants must be agents. Support for custom executors is not available in handoff workflows.
+    Participants must be ``Agent`` instances. ``SupportsAgentRun`` protocol implementors that
+    are not ``Agent`` subclasses are not supported because handoff workflows require cloning,
+    tool injection, and middleware — capabilities only available on ``Agent``.
 
     Outputs:
     The final conversation history as a list of Message once the group chat completes.
 
     Note:
-    1. Agents in handoff workflows must be Agent instances and support local tool calls.
+    1. Agents in handoff workflows must be ``Agent`` instances and support local tool calls.
     2. Handoff doesn't support intermediate outputs from agents. All outputs are returned as
        they become available. This is because agents in handoff workflows are not considered
        sub-agents of a central orchestrator, thus all outputs are directly emitted.
@@ -717,7 +717,7 @@ def __init__(
         self,
         *,
         name: str | None = None,
-        participants: Sequence[SupportsAgentRun] | None = None,
+        participants: Sequence[Agent] | None = None,
         description: str | None = None,
         checkpoint_storage: CheckpointStorage | None = None,
         termination_condition: TerminationCondition | None = None,
@@ -734,7 +734,7 @@ def __init__(
         Args:
             name: Optional workflow identifier used in logging and debugging.
                   If not provided, a default name will be generated.
-            participants: Optional list of agents that will participate in the handoff workflow.
+            participants: Optional list of ``Agent`` instances that will participate in the handoff workflow.
                           You can also call `.participants([...])` later. Each participant must have a
                           unique identifier (`.name` is preferred if set, otherwise `.id` is used).
             description: Optional human-readable description explaining the workflow's
@@ -747,7 +747,7 @@ def __init__(
         self._description = description
 
         # Participant related members
-        self._participants: dict[str, SupportsAgentRun] = {}
+        self._participants: dict[str, Agent] = {}
         self._start_id: str | None = None
 
         if participants:
@@ -768,11 +768,11 @@ def __init__(
         # Termination related members
         self._termination_condition: Callable[[list[Message]], bool | Awaitable[bool]] | None = termination_condition
 
-    def participants(self, participants: Sequence[SupportsAgentRun]) -> "HandoffBuilder":
+    def participants(self, participants: Sequence[Agent]) -> "HandoffBuilder":
         """Register the agents that will participate in the handoff workflow.
 
         Args:
-            participants: Sequence of SupportsAgentRun instances. Each must have a unique identifier.
+            participants: Sequence of ``Agent`` instances. Each must have a unique identifier.
                 (`.name` is preferred if set, otherwise `.id` is used).
 
         Returns:
@@ -781,7 +781,7 @@ def participants(self, participants: Sequence[SupportsAgentRun]) -> "HandoffBuil
         Raises:
             ValueError: If participants is empty, contains duplicates, or `.participants()`
                         has already been called.
-            TypeError: If participants are not SupportsAgentRun instances.
+            TypeError: If participants are not ``Agent`` instances.
 
         Example:
 
@@ -804,14 +804,15 @@ def participants(self, participants: Sequence[SupportsAgentRun]) -> "HandoffBuil
         if not participants:
             raise ValueError("participants cannot be empty")
 
-        named: dict[str, SupportsAgentRun] = {}
+        named: dict[str, Agent] = {}
         for participant in participants:
-            if isinstance(participant, SupportsAgentRun):
-                resolved_id = self._resolve_to_id(participant)
-            else:
+            if not isinstance(participant, Agent):
                 raise TypeError(
-                    f"Participants must be SupportsAgentRun or Executor instances. Got {type(participant).__name__}."
+                    f"Participants must be Agent instances. Got {type(participant).__name__}. "
+                    "Handoff workflows require Agent because they rely on cloning, tool injection, "
+                    "and middleware capabilities."
                 )
+            resolved_id = self._resolve_to_id(participant)
 
             if resolved_id in named:
                 raise ValueError(f"Duplicate participant name '{resolved_id}' detected")
@@ -823,8 +824,8 @@ def participants(self, participants: Sequence[SupportsAgentRun]) -> "HandoffBuil
 
     def add_handoff(
         self,
-        source: SupportsAgentRun,
-        targets: Sequence[SupportsAgentRun],
+        source: Agent,
+        targets: Sequence[Agent],
         *,
         description: str | None = None,
     ) -> "HandoffBuilder":
@@ -905,7 +906,7 @@ def add_handoff(
 
         return self
 
-    def with_start_agent(self, agent: SupportsAgentRun) -> "HandoffBuilder":
+    def with_start_agent(self, agent: Agent) -> "HandoffBuilder":
         """Set the agent that will initiate the handoff workflow.
 
         If not specified, the first registered participant will be used as the starting agent.
@@ -929,7 +930,7 @@ def with_start_agent(self, agent: SupportsAgentRun) -> "HandoffBuilder":
     def with_autonomous_mode(
         self,
         *,
-        agents: Sequence[SupportsAgentRun] | Sequence[str] | None = None,
+        agents: Sequence[Agent] | Sequence[str] | None = None,
         prompts: dict[str, str] | None = None,
         turn_limits: dict[str, int] | None = None,
     ) -> "HandoffBuilder":
@@ -943,7 +944,7 @@ def with_autonomous_mode(
         Args:
             agents: Optional list of agents to enable autonomous mode for. Can be:
                     - Factory names (str): If using participant factories
-                    - SupportsAgentRun instances: The actual agent objects
+                    - SupportsAgentRun / Agent instances: The actual agent objects
                     - If not provided, all agents will operate in autonomous mode.
             prompts: Optional mapping of agent identifiers/factory names to custom prompts to use when continuing
                      in autonomous mode. If not provided, a default prompt will be used.
@@ -1092,22 +1093,22 @@ def build(self) -> Workflow:
 
     # region Internal Helper Methods
 
-    def _resolve_agents(self) -> dict[str, SupportsAgentRun]:
+    def _resolve_agents(self) -> dict[str, Agent]:
         """Resolve participant instances into agent instances.
 
         Returns:
-            Map of executor IDs to `SupportsAgentRun` instances
+            Map of executor IDs to ``Agent`` instances
         """
         if not self._participants:
             raise ValueError("No participants provided. Call .participants() first.")
 
         return self._participants
 
-    def _resolve_handoffs(self, agents: dict[str, SupportsAgentRun]) -> dict[str, list[HandoffConfiguration]]:
+    def _resolve_handoffs(self, agents: dict[str, Agent]) -> dict[str, list[HandoffConfiguration]]:
         """Resolve handoff configurations to executor IDs.
 
         Args:
-            agents: Map of agent IDs to `SupportsAgentRun` instances
+            agents: Map of agent IDs to ``Agent`` instances
 
         Returns:
             Map of executor IDs to list of HandoffConfiguration instances
@@ -1154,13 +1155,13 @@ def _resolve_handoffs(self, agents: dict[str, SupportsAgentRun]) -> dict[str, li
 
     def _resolve_executors(
         self,
-        agents: dict[str, SupportsAgentRun],
+        agents: dict[str, Agent],
         handoffs: dict[str, list[HandoffConfiguration]],
     ) -> dict[str, HandoffAgentExecutor]:
         """Resolve agents into HandoffAgentExecutors.
 
         Args:
-            agents: Map of agent IDs to `SupportsAgentRun` instances
+            agents: Map of agent IDs to ``Agent`` instances
             handoffs: Map of executor IDs to list of HandoffConfiguration instances
 
         Returns:

python/packages/orchestrations/tests/test_handoff.py

@@ -1091,3 +1091,29 @@ def regular_tool() -> str:
 
     call_next.assert_awaited_once()
     assert context.result is None
+
+
+def test_handoff_builder_rejects_non_agent_supports_agent_run():
+    """Verify that participants() rejects SupportsAgentRun implementations that are not Agent instances."""
+    from agent_framework import AgentResponse, AgentSession, SupportsAgentRun
+
+    class FakeAgentRun:
+        def __init__(self, id, name):
+            self.id = id
+            self.name = name
+            self.description = "d"
+
+        async def run(self, messages=None, *, stream=False, session=None, **kwargs):
+            return AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text("ok")])])
+
+        def create_session(self, **kwargs):
+            return AgentSession()
+
+        def get_session(self, *, service_session_id, **kwargs):
+            return AgentSession(service_session_id=service_session_id)
+
+    fake = FakeAgentRun("a", "A")
+    assert isinstance(fake, SupportsAgentRun)
+
+    with pytest.raises(TypeError, match="Participants must be Agent instances"):
+        HandoffBuilder().participants([fake])