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

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

@@ -80,6 +80,7 @@
     RequestInfoMessage,
     RequestResponse,
 )
+from ._request_info_mixin import response_handler
 from ._runner import Runner
 from ._runner_context import (
     InProcRunnerContext,
@@ -90,7 +91,6 @@
 from ._shared_state import SharedState
 from ._validation import (
     EdgeDuplicationError,
-    ExecutorDuplicationError,
     GraphConnectivityError,
     TypeCompatibilityError,
     ValidationTypeEnum,
@@ -117,7 +117,6 @@
     "EdgeDuplicationError",
     "Executor",
     "ExecutorCompletedEvent",
-    "ExecutorDuplicationError",
     "ExecutorEvent",
     "ExecutorFailedEvent",
     "ExecutorInvokedEvent",
@@ -187,5 +186,6 @@
     "executor",
     "get_checkpoint_summary",
     "handler",
+    "response_handler",
     "validate_workflow_graph",
 ]

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

@@ -78,6 +78,7 @@ from ._request_info_executor import (
     RequestInfoMessage,
     RequestResponse,
 )
+from ._request_info_mixin import response_handler
 from ._runner import Runner
 from ._runner_context import (
     InProcRunnerContext,
@@ -88,7 +89,6 @@ from ._sequential import SequentialBuilder
 from ._shared_state import SharedState
 from ._validation import (
     EdgeDuplicationError,
-    ExecutorDuplicationError,
     GraphConnectivityError,
     TypeCompatibilityError,
     ValidationTypeEnum,
@@ -115,7 +115,6 @@ __all__ = [
     "EdgeDuplicationError",
     "Executor",
     "ExecutorCompletedEvent",
-    "ExecutorDuplicationError",
     "ExecutorEvent",
     "ExecutorFailedEvent",
     "ExecutorInvokedEvent",
@@ -185,5 +184,6 @@ __all__ = [
     "executor",
     "get_checkpoint_summary",
     "handler",
+    "response_handler",
     "validate_workflow_graph",
 ]

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

@@ -1,3 +1,10 @@
 # Copyright (c) Microsoft. All rights reserved.
 
 DEFAULT_MAX_ITERATIONS = 100  # Default maximum iterations for workflow execution.
+
+INTERNAL_SOURCE_PREFIX = "internal"  # Source identifier for internal workflow messages.
+
+
+def INTERNAL_SOURCE_ID(executor_id: str) -> str:
+    """Generate an internal source ID for a given executor."""
+    return f"{INTERNAL_SOURCE_PREFIX}:{executor_id}"

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

@@ -6,6 +6,7 @@
 from dataclasses import dataclass, field
 from typing import Any, ClassVar
 
+from ._const import INTERNAL_SOURCE_ID
 from ._executor import Executor
 from ._model_utils import DictConvertible, encode_value
 
@@ -865,3 +866,27 @@ def to_dict(self) -> dict[str, Any]:
         payload = super().to_dict()
         payload["cases"] = [encode_value(case) for case in self.cases]
         return payload
+
+
+class InternalEdgeGroup(EdgeGroup):
+    """Special edge group used to route internal messages to executors.
+
+    This group is not serialized and is only used at runtime to link internal
+    executors that should not be exposed as part of the public workflow graph.
+    """
+
+    def __init__(self, executor_id: str) -> None:
+        """Create an internal edge group from the given edges.
+
+        Parameters
+        ----------
+        executor_id:
+            Identifier of the internal executor that should receive messages.
+
+        Examples:
+            .. code-block:: python
+
+                edge_group = InternalEdgeGroup("executor_a")
+        """
+        edge = Edge(source_id=INTERNAL_SOURCE_ID(executor_id), target_id=executor_id)
+        super().__init__([edge])

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

@@ -8,7 +8,15 @@
 from typing import Any, cast
 
 from ..observability import EdgeGroupDeliveryStatus, OtelAttr, create_edge_group_processing_span
-from ._edge import Edge, EdgeGroup, FanInEdgeGroup, FanOutEdgeGroup, SingleEdgeGroup, SwitchCaseEdgeGroup
+from ._edge import (
+    Edge,
+    EdgeGroup,
+    FanInEdgeGroup,
+    FanOutEdgeGroup,
+    InternalEdgeGroup,
+    SingleEdgeGroup,
+    SwitchCaseEdgeGroup,
+)
 from ._executor import Executor
 from ._runner_context import Message, RunnerContext
 from ._shared_state import SharedState
@@ -44,11 +52,11 @@ async def send_message(self, message: Message, shared_state: SharedState, ctx: R
         """
         raise NotImplementedError
 
-    def _can_handle(self, executor_id: str, message_data: Any) -> bool:
+    def _can_handle(self, executor_id: str, message: Message) -> bool:
         """Check if an executor can handle the given message data."""
         if executor_id not in self._executors:
             return False
-        return self._executors[executor_id].can_handle(message_data)
+        return self._executors[executor_id].can_handle(message)
 
     async def _execute_on_target(
         self,
@@ -66,7 +74,7 @@ async def _execute_on_target(
 
         # Execute with trace context parameters
         await target_executor.execute(
-            message.data,
+            message,
             source_ids,  # source_executor_ids
             shared_state,  # shared_state
             ctx,  # runner_context
@@ -103,7 +111,7 @@ async def send_message(self, message: Message, shared_state: SharedState, ctx: R
                     })
                     return False
 
-                if self._can_handle(self._edge.target_id, message.data):
+                if self._can_handle(self._edge.target_id, message):
                     if self._edge.should_route(message.data):
                         span.set_attributes({
                             OtelAttr.EDGE_GROUP_DELIVERED: True,
@@ -183,7 +191,7 @@ async def send_message(self, message: Message, shared_state: SharedState, ctx: R
                     # If the target ID is specified and the selection result contains it, send the message to that edge
                     if message.target_id in selection_results:
                         edge = self._target_map.get(message.target_id)
-                        if edge and self._can_handle(edge.target_id, message.data):
+                        if edge and self._can_handle(edge.target_id, message):
                             if edge.should_route(message.data):
                                 span.set_attributes({
                                     OtelAttr.EDGE_GROUP_DELIVERED: True,
@@ -215,7 +223,7 @@ async def send_message(self, message: Message, shared_state: SharedState, ctx: R
                     # If no target ID, send the message to the selected targets
                     for target_id in selection_results:
                         edge = self._target_map[target_id]
-                        if self._can_handle(edge.target_id, message.data) and edge.should_route(message.data):
+                        if self._can_handle(edge.target_id, message) and edge.should_route(message.data):
                             deliverable_edges.append(edge)
 
                     if len(deliverable_edges) > 0:
@@ -291,7 +299,9 @@ async def send_message(self, message: Message, shared_state: SharedState, ctx: R
                     return False
 
                 # Check if target can handle list of message data (fan-in aggregates multiple messages)
-                if self._can_handle(self._edges[0].target_id, [message.data]):
+                if self._can_handle(
+                    self._edges[0].target_id, Message(data=[message.data], source_id=message.source_id)
+                ):
                     # If the edge can handle the data, buffer the message
                     self._buffer[message.source_id].append(message)
                     span.set_attributes({
@@ -374,7 +384,7 @@ def create_edge_runner(edge_group: EdgeGroup, executors: dict[str, Executor]) ->
     Returns:
         The appropriate EdgeRunner instance.
     """
-    if isinstance(edge_group, SingleEdgeGroup):
+    if isinstance(edge_group, (SingleEdgeGroup, InternalEdgeGroup)):
         return SingleEdgeRunner(edge_group, executors)
     if isinstance(edge_group, SwitchCaseEdgeGroup):
         return SwitchCaseEdgeRunner(edge_group, executors)

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

@@ -212,6 +212,7 @@ def __init__(
         source_executor_id: str,
         request_type: type,
         request_data: "RequestInfoMessage",
+        response_type: type,
     ):
         """Initialize the request info event.
 
@@ -220,11 +221,13 @@ def __init__(
             source_executor_id: ID of the executor that made the request.
             request_type: Type of the request (e.g., a specific data type).
             request_data: The data associated with the request.
+            response_type: Expected type of the response.
         """
         super().__init__(request_data)
         self.request_id = request_id
         self.source_executor_id = source_executor_id
         self.request_type = request_type
+        self.response_type = response_type
 
     def __repr__(self) -> str:
         """Return a string representation of the request info event."""
@@ -233,7 +236,8 @@ def __repr__(self) -> str:
             f"request_id={self.request_id}, "
             f"source_executor_id={self.source_executor_id}, "
             f"request_type={self.request_type.__name__}, "
-            f"data={self.data})"
+            f"data={self.data}, "
+            f"response_type={self.response_type.__name__})"
         )
 
 

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

@@ -16,7 +16,8 @@
     _framework_event_origin,  # type: ignore[reportPrivateUsage]
 )
 from ._model_utils import DictConvertible
-from ._runner_context import Message, RunnerContext  # type: ignore
+from ._request_info_mixin import RequestInfoMixin
+from ._runner_context import Message, MessageType, RunnerContext
 from ._shared_state import SharedState
 from ._typing_utils import is_instance_of
 from ._workflow_context import WorkflowContext, validate_function_signature
@@ -25,7 +26,7 @@
 
 
 # region Executor
-class Executor(DictConvertible):
+class Executor(RequestInfoMixin, DictConvertible):
     """Base class for all workflow executors that process messages and perform computations.
 
     ## Overview
@@ -204,6 +205,9 @@ def __init__(
                     "Please define at least one handler using the @handler decorator."
                 )
 
+            # Initialize RequestInfoMixin to discover response handlers
+            self._discover_response_handlers()
+
     async def execute(
         self,
         message: Any,
@@ -229,12 +233,17 @@ async def execute(
         Returns:
             An awaitable that resolves to the result of the execution.
         """
-        # Create processing span for tracing (gracefully handles disabled tracing)
+        # Default to find handler in regular handlers
+        target_handlers = self._handlers
 
         # Handle case where Message wrapper is passed instead of raw data
         if isinstance(message, Message):
+            if message.message_type == MessageType.RESPONSE:
+                # Switch to response handlers if message is a response
+                target_handlers = self._response_handlers
             message = message.data
 
+        # Create processing span for tracing (gracefully handles disabled tracing)
         with create_processing_span(
             self.id,
             self.__class__.__name__,
@@ -244,15 +253,9 @@ async def execute(
         ):
             # Find the handler and handler spec that matches the message type.
             handler: Callable[[Any, WorkflowContext[Any, Any]], Awaitable[None]] | None = None
-            ctx_annotation = None
-            for message_type in self._handlers:
+            for message_type in target_handlers:
                 if is_instance_of(message, message_type):
-                    handler = self._handlers[message_type]
-                    # Find the corresponding handler spec for context annotation
-                    for spec in self._handler_specs:
-                        if spec.get("message_type") == message_type:
-                            ctx_annotation = spec.get("ctx_annotation")
-                            break
+                    handler = target_handlers[message_type]
                     break
 
             if handler is None:
@@ -263,7 +266,6 @@ async def execute(
                 source_executor_ids=source_executor_ids,
                 shared_state=shared_state,
                 runner_context=runner_context,
-                ctx_annotation=ctx_annotation,
                 trace_contexts=trace_contexts,
                 source_span_ids=source_span_ids,
             )
@@ -289,7 +291,6 @@ def _create_context_for_handler(
         source_executor_ids: list[str],
         shared_state: SharedState,
         runner_context: RunnerContext,
-        ctx_annotation: Any,
         trace_contexts: list[dict[str, str]] | None = None,
         source_span_ids: list[str] | None = None,
     ) -> WorkflowContext[Any]:
@@ -299,7 +300,6 @@ def _create_context_for_handler(
             source_executor_ids: The IDs of the source executors that sent messages to this executor.
             shared_state: The shared state for the workflow.
             runner_context: The runner context that provides methods to send messages and events.
-            ctx_annotation: The context annotation from the handler spec to determine which context type to create.
             trace_contexts: Optional trace contexts from multiple sources for OpenTelemetry propagation.
             source_span_ids: Optional source span IDs from multiple sources for linking.
 
@@ -350,7 +350,7 @@ def _discover_handlers(self) -> None:
                 # Skip attributes that may not be accessible
                 continue
 
-    def can_handle(self, message: Any) -> bool:
+    def can_handle(self, message: Message) -> bool:
         """Check if the executor can handle a given message type.
 
         Args:
@@ -359,7 +359,11 @@ def can_handle(self, message: Any) -> bool:
         Returns:
             True if the executor can handle the message type, False otherwise.
         """
-        return any(is_instance_of(message, message_type) for message_type in self._handlers)
+        if message.message_type == MessageType.REGULAR:
+            return any(is_instance_of(message.data, message_type) for message_type in self._handlers)
+        if message.message_type == MessageType.RESPONSE:
+            return any(is_instance_of(message.data, message_type) for message_type in self._response_handlers)
+        return False
 
     def _register_instance_handler(
         self,