Python: Add sample to show handoff as agent with HITL (#2534)

* Add sample to show handoff as agent with HITL

* Update uv.lock with latest pkg versions. Fix lint error.

* Upgrade grpcio to 1.76.0

* Handle grpcio versions

* Case insensitive compare for declarative

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

@@ -149,9 +149,9 @@ class WorkflowGraphValidator:
         # check only when there is at least one edge group defined.
         if self._edges:  # Only evaluate when the workflow defines edges
             edge_executor_ids: set[str] = set()
-            for _e in self._edges:
-                edge_executor_ids.add(_e.source_id)
-                edge_executor_ids.add(_e.target_id)
+            for e in self._edges:
+                edge_executor_ids.add(e.source_id)
+                edge_executor_ids.add(e.target_id)
             if start_executor_id not in edge_executor_ids:
                 raise GraphConnectivityError(
                     f"Start executor '{start_executor_id}' is not present in the workflow graph"

python/packages/declarative/agent_framework_declarative/_models.py

@@ -253,7 +253,7 @@ class Connection(SerializationMixin):
             # We're being called on a subclass, use the normal from_dict
             return SerializationMixin.from_dict.__func__(cls, value, dependencies=dependencies)  # type: ignore[misc]
 
-        kind = value.get("kind", "")
+        kind = value.get("kind", "").lower()
         if kind == "reference":
             return SerializationMixin.from_dict.__func__(  # type: ignore[misc]
                 ReferenceConnection, value, dependencies=dependencies
@@ -262,7 +262,7 @@ class Connection(SerializationMixin):
             return SerializationMixin.from_dict.__func__(  # type: ignore[misc]
                 RemoteConnection, value, dependencies=dependencies
             )
-        if kind == "key":
+        if kind in ("key", "apikey"):
             return SerializationMixin.from_dict.__func__(  # type: ignore[misc]
                 ApiKeyConnection, value, dependencies=dependencies
             )

python/pyproject.toml

@@ -71,6 +71,11 @@ override-dependencies = [
   "uvicorn==0.38.0",
   # Similar problem with websockets, which is a dependency conflict between litellm[proxy] and mcp
   "websockets==15.0.1",
+  # grpcio 1.67.x has no Python 3.14 wheels; grpcio 1.76.0+ supports Python 3.14
+  # litellm constrains grpcio<1.68.0 due to resource exhaustion bug (https://github.com/grpc/grpc/issues/38290)
+  # Use version-specific overrides to satisfy both constraints
+  "grpcio>=1.76.0; python_version >= '3.14'",
+  "grpcio>=1.62.3,<1.68.0; python_version < '3.14'",
 ]
 
 [tool.uv.workspace]

python/samples/getting_started/workflows/README.md

@@ -44,6 +44,7 @@ Once comfortable with these, explore the rest of the samples below.
 | Magentic Workflow as Agent | [agents/magentic_workflow_as_agent.py](./agents/magentic_workflow_as_agent.py) | Configure Magentic orchestration with callbacks, then expose the workflow as an agent |
 | Workflow as Agent (Reflection Pattern) | [agents/workflow_as_agent_reflection_pattern.py](./agents/workflow_as_agent_reflection_pattern.py) | Wrap a workflow so it can behave like an agent (reflection pattern) |
 | Workflow as Agent + HITL | [agents/workflow_as_agent_human_in_the_loop.py](./agents/workflow_as_agent_human_in_the_loop.py) | Extend workflow-as-agent with human-in-the-loop capability |
+| Handoff Workflow as Agent | [agents/handoff_workflow_as_agent.py](./agents/handoff_workflow_as_agent.py) | Use a HandoffBuilder workflow as an agent with HITL via FunctionCallContent/FunctionResultContent |
 
 ### checkpoint
 

python/samples/getting_started/workflows/agents/handoff_workflow_as_agent.py

@@ -0,0 +1,230 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+import asyncio
+from collections.abc import Mapping
+from typing import Any
+
+from agent_framework import (
+    ChatAgent,
+    ChatMessage,
+    FunctionCallContent,
+    FunctionResultContent,
+    HandoffBuilder,
+    HandoffUserInputRequest,
+    Role,
+    WorkflowAgent,
+)
+from agent_framework.azure import AzureOpenAIChatClient
+from azure.identity import AzureCliCredential
+
+"""
+Sample: Handoff Workflow as Agent with Human-in-the-Loop
+
+Purpose:
+This sample demonstrates how to use a HandoffBuilder workflow as an agent via
+`.as_agent()`, enabling human-in-the-loop interactions through the standard
+agent interface. The handoff pattern routes user requests through a triage agent
+to specialist agents, with the workflow requesting user input as needed.
+
+When using a handoff workflow as an agent:
+1. The workflow emits `HandoffUserInputRequest` when it needs user input
+2. `WorkflowAgent` converts this to a `FunctionCallContent` named "request_info"
+3. The caller extracts `HandoffUserInputRequest` from the function call arguments
+4. The caller provides a response via `FunctionResultContent`
+
+This differs from running the workflow directly:
+- Direct workflow: Use `workflow.run_stream()` and `workflow.send_responses_streaming()`
+- As agent: Use `agent.run()` with `FunctionCallContent`/`FunctionResultContent` messages
+
+Key Concepts:
+- HandoffBuilder: Creates triage-to-specialist routing workflows
+- WorkflowAgent: Wraps workflows to expose them as standard agents
+- HandoffUserInputRequest: Contains conversation context and the awaiting agent
+- FunctionCallContent/FunctionResultContent: Standard agent interface for HITL
+
+Prerequisites:
+- `az login` (Azure CLI authentication)
+- Environment variables configured for AzureOpenAIChatClient (AZURE_OPENAI_ENDPOINT, etc.)
+"""
+
+
+def create_agents(chat_client: AzureOpenAIChatClient) -> tuple[ChatAgent, ChatAgent, ChatAgent, ChatAgent]:
+    """Create and configure the triage and specialist agents.
+
+    The triage agent dispatches requests to the appropriate specialist.
+    Specialists handle their domain-specific queries.
+
+    Returns:
+        Tuple of (triage_agent, refund_agent, order_agent, support_agent)
+    """
+    triage = chat_client.create_agent(
+        instructions=(
+            "You are frontline support triage. Read the latest user message and decide whether "
+            "to hand off to refund_agent, order_agent, or support_agent. Provide a brief natural-language "
+            "response for the user. When delegation is required, call the matching handoff tool "
+            "(`handoff_to_refund_agent`, `handoff_to_order_agent`, or `handoff_to_support_agent`)."
+        ),
+        name="triage_agent",
+    )
+
+    refund = chat_client.create_agent(
+        instructions=(
+            "You handle refund workflows. Ask for any order identifiers you require and outline the refund steps."
+        ),
+        name="refund_agent",
+    )
+
+    order = chat_client.create_agent(
+        instructions=(
+            "You resolve shipping and fulfillment issues. Clarify the delivery problem and describe the actions "
+            "you will take to remedy it."
+        ),
+        name="order_agent",
+    )
+
+    support = chat_client.create_agent(
+        instructions=(
+            "You are a general support agent. Offer empathetic troubleshooting and gather missing details if the "
+            "issue does not match other specialists."
+        ),
+        name="support_agent",
+    )
+
+    return triage, refund, order, support
+
+
+def extract_handoff_request(
+    response_messages: list[ChatMessage],
+) -> tuple[FunctionCallContent, HandoffUserInputRequest]:
+    """Extract the HandoffUserInputRequest from agent response messages.
+
+    When a handoff workflow running as an agent needs user input, it emits a
+    FunctionCallContent with name="request_info" containing the HandoffUserInputRequest.
+
+    Args:
+        response_messages: Messages from the agent response
+
+    Returns:
+        Tuple of (function_call, handoff_request)
+
+    Raises:
+        ValueError: If no request_info function call is found or payload is invalid
+    """
+    for message in response_messages:
+        for content in message.contents:
+            if isinstance(content, FunctionCallContent) and content.name == WorkflowAgent.REQUEST_INFO_FUNCTION_NAME:
+                # Parse the function arguments to extract the HandoffUserInputRequest
+                args = content.arguments
+                if isinstance(args, str):
+                    request_args = WorkflowAgent.RequestInfoFunctionArgs.from_json(args)
+                elif isinstance(args, Mapping):
+                    request_args = WorkflowAgent.RequestInfoFunctionArgs.from_dict(dict(args))
+                else:
+                    raise ValueError("Unexpected argument type for request_info function call.")
+
+                payload: Any = request_args.data
+                if not isinstance(payload, HandoffUserInputRequest):
+                    raise ValueError(
+                        f"Expected HandoffUserInputRequest in request_info payload, got {type(payload).__name__}"
+                    )
+
+                return content, payload
+
+    raise ValueError("No request_info function call found in response messages.")
+
+
+def print_conversation(request: HandoffUserInputRequest) -> None:
+    """Display the conversation history from a HandoffUserInputRequest."""
+    print("\n=== Conversation History ===")
+    for message in request.conversation:
+        speaker = message.author_name or message.role.value
+        print(f"  [{speaker}]: {message.text}")
+    print(f"  [Awaiting]: {request.awaiting_agent_id}")
+    print("============================")
+
+
+async def main() -> None:
+    """Main entry point demonstrating handoff workflow as agent.
+
+    This demo:
+    1. Builds a handoff workflow with triage and specialist agents
+    2. Converts it to an agent using .as_agent()
+    3. Runs a multi-turn conversation with scripted user responses
+    4. Demonstrates the FunctionCallContent/FunctionResultContent pattern for HITL
+    """
+    print("Starting Handoff Workflow as Agent Demo")
+    print("=" * 55)
+
+    # Initialize the Azure OpenAI chat client
+    chat_client = AzureOpenAIChatClient(credential=AzureCliCredential())
+
+    # Create agents
+    triage, refund, order, support = create_agents(chat_client)
+
+    # Build the handoff workflow and convert to agent
+    # Termination condition: stop after 4 user messages
+    agent = (
+        HandoffBuilder(
+            name="customer_support_handoff",
+            participants=[triage, refund, order, support],
+        )
+        .set_coordinator("triage_agent")
+        .with_termination_condition(lambda conv: sum(1 for msg in conv if msg.role.value == "user") >= 4)
+        .build()
+        .as_agent()  # Convert workflow to agent interface
+    )
+
+    # Scripted user responses for reproducible demo
+    scripted_responses = [
+        "My order 1234 arrived damaged and the packaging was destroyed.",
+        "Yes, I'd like a refund if that's possible.",
+        "Thanks for your help!",
+    ]
+
+    # Start the conversation
+    print("\n[User]: Hello, I need assistance with my recent purchase.")
+    response = await agent.run("Hello, I need assistance with my recent purchase.")
+
+    # Process conversation turns until workflow completes or responses exhausted
+    while True:
+        # Check if the agent is requesting user input
+        try:
+            function_call, handoff_request = extract_handoff_request(response.messages)
+        except ValueError:
+            # No request_info call found - workflow has completed
+            print("\n[Workflow completed - no pending requests]")
+            if response.messages:
+                final_text = response.messages[-1].text
+                if final_text:
+                    print(f"[Final response]: {final_text}")
+            break
+
+        # Display the conversation context
+        print_conversation(handoff_request)
+
+        # Get the next scripted response
+        if not scripted_responses:
+            print("\n[No more scripted responses - ending conversation]")
+            break
+
+        user_input = scripted_responses.pop(0)
+
+        print(f"\n[User responding]: {user_input}")
+
+        # Create the function result to send back to the agent
+        # The result is the user's text response which gets converted to ChatMessage
+        function_result = FunctionResultContent(
+            call_id=function_call.call_id,
+            result=user_input,
+        )
+
+        # Send the response back to the agent
+        response = await agent.run(ChatMessage(role=Role.TOOL, contents=[function_result]))
+
+    print("\n" + "=" * 55)
+    print("Demo completed!")
+
+
+if __name__ == "__main__":
+    print("Initializing Handoff Workflow as Agent Sample...")
+    asyncio.run(main())