agent/agent-framework
1
0
Fork 0
mirror of https://github.com/microsoft/agent-framework.git synced 2026-06-16 21:04:09 +08:00
Code Issues Packages Projects Releases Wiki Activity

Python: [BREAKING] Remove Request Interceptor Architecture - Simplify Sub-workflow Communication (#898)

Browse Source 
* removed intercepts_request and simplified how interception is handled

* parameterize SubWorkflowRequestInfo

* revert back the field rename of RequestResponse

* remove duplicate tests

* ignore type error

* remove SubWorkflowResponse

* Remove SubWorkflowRequestInfo and update RequestInfoMessage with source_executor_id for correlation
This commit is contained in:
Eric Zhu
2025-09-24 20:43:37 -07:00
committed by GitHub
Unverified
parent 39e071c430
commit 366a7f7d47
14 changed files with 397 additions and 1131 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/packages/main/agent_framework/_workflow/__init__.py
-6
View File
@@ -51,10 +51,7 @@ from ._executor import (
RequestInfoExecutor,
RequestInfoMessage,
RequestResponse,
SubWorkflowRequestInfo,
SubWorkflowResponse,
handler,
intercepts_request,
)
from ._function_executor import FunctionExecutor, executor
from ._magentic import (
@@ -157,8 +154,6 @@ __all__ = [
"SharedState",
"SingleEdgeGroup",
"StandardMagenticManager",
"SubWorkflowRequestInfo",
"SubWorkflowResponse",
"SwitchCaseEdgeGroup",
"SwitchCaseEdgeGroupCase",
"SwitchCaseEdgeGroupDefault",
@@ -185,7 +180,6 @@ __all__ = [
"create_edge_runner",
"executor",
"handler",
"intercepts_request",
"validate_workflow_graph",
]
python/packages/main/agent_framework/_workflow/__init__.pyi
-6
View File
@@ -47,10 +47,7 @@ from ._executor import (
RequestInfoExecutor,
RequestInfoMessage,
RequestResponse,
SubWorkflowRequestInfo,
SubWorkflowResponse,
handler,
intercepts_request,
)
from ._function_executor import FunctionExecutor, executor
from ._magentic import (
@@ -153,8 +150,6 @@ __all__ = [
"SharedState",
"SingleEdgeGroup",
"StandardMagenticManager",
"SubWorkflowRequestInfo",
"SubWorkflowResponse",
"SwitchCaseEdgeGroup",
"SwitchCaseEdgeGroupCase",
"SwitchCaseEdgeGroupDefault",
@@ -181,6 +176,5 @@ __all__ = [
"create_edge_runner",
"executor",
"handler",
"intercepts_request",
"validate_workflow_graph",
]
python/packages/main/agent_framework/_workflow/_executor.py
+60 -494
View File
@@ -9,7 +9,7 @@ import uuid
from collections.abc import Awaitable, Callable, Iterable, Mapping, Sequence
from dataclasses import asdict, dataclass, field, fields, is_dataclass
from textwrap import shorten
from typing import Any, ClassVar, Generic, TypeVar, cast, overload
from typing import Any, ClassVar, Generic, TypeVar, cast
from pydantic import Field
@@ -104,20 +104,6 @@ class Executor(AFBaseModel):
```
Access via the `workflow_output_types` property.
### Request Types
The types of sub-workflow requests an executor can intercept via `@intercepts_request`:
```python
class ParentExecutor(Executor):
@intercepts_request
async def handle_request(
self,
request: MyRequest,
ctx: WorkflowContext,
) -> RequestResponse[MyRequest, str]:
# Can intercept 'MyRequest' from sub-workflows
```
Access via the `request_types` property.
## Handler Discovery
Executors discover their capabilities through decorated methods:
@@ -130,17 +116,21 @@ class Executor(AFBaseModel):
await ctx.send_message(message.upper())
```
### @intercepts_request Decorator
Marks methods that intercept sub-workflow requests:
### Sub-workflow Request Interception
Use @handler methods to intercept sub-workflow requests:
```python
class ParentExecutor(Executor):
@intercepts_request
async def check_domain(
self, request: DomainRequest, ctx: WorkflowContext
) -> RequestResponse[DomainRequest, bool]:
@handler
async def handle_domain_request(
self,
request: DomainRequest, # Subclass of RequestInfoMessage
ctx: WorkflowContext[RequestResponse[RequestInfoMessage, Any] | DomainRequest],
) -> None:
if self.is_allowed(request.domain):
return RequestResponse.handled(True)
return RequestResponse.forward()
response = RequestResponse(data=True, original_request=request, request_id=request.request_id)
await ctx.send_message(response, target_id=request.source_executor_id)
else:
await ctx.send_message(request) # Forward to external
```
## Context Types
@@ -196,7 +186,7 @@ class Executor(AFBaseModel):
## Implementation Notes
- Do not call `execute()` directly - it's invoked by the workflow engine
- Do not override `execute()` - define handlers using decorators instead
- Each executor must have at least one `@handler` or `@intercepts_request` method
- Each executor must have at least one `@handler` method
- Handler method signatures are validated at initialization time
"""
@@ -226,15 +216,13 @@ class Executor(AFBaseModel):
super().__init__(**kwargs)
self._handlers: dict[type, Callable[[Any, WorkflowContext[Any, Any]], Awaitable[None]]] = {}
self._request_interceptors: dict[type | str, list[dict[str, Any]]] = {}
self._handler_specs: list[dict[str, Any]] = []
self._discover_handlers()
if not self._handlers and not self._request_interceptors:
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 "
"or @intercepts_request decorator."
"Please define at least one handler using the @handler decorator."
)
async def execute(
@@ -276,7 +264,6 @@ class Executor(AFBaseModel):
source_span_ids=source_span_ids,
):
# Find the handler and handler spec that matches the message type.
# This includes the self._handle_sub_workflow_request handler for SubWorkflowRequestInfo.
handler: Callable[[Any, WorkflowContext[Any, Any]], Awaitable[None]] | None = None
ctx_annotation = None
for message_type in self._handlers:
@@ -351,202 +338,39 @@ class Executor(AFBaseModel):
)
def _discover_handlers(self) -> None:
"""Discover message handlers and request interceptors in the executor class."""
"""Discover message handlers in the executor class."""
# Use __class__.__dict__ to avoid accessing pydantic's dynamic attributes
for attr_name in dir(self.__class__):
try:
attr = getattr(self.__class__, attr_name)
if callable(attr):
# Discover @handler methods
if hasattr(attr, "_handler_spec"):
handler_spec = attr._handler_spec # type: ignore
message_type = handler_spec["message_type"]
# Discover @handler methods
if callable(attr) and hasattr(attr, "_handler_spec"):
handler_spec = attr._handler_spec # type: ignore
message_type = handler_spec["message_type"]
# Keep full generic types for handler registration to avoid conflicts
# Different RequestResponse[T, U] specializations are distinct handler types
# Keep full generic types for handler registration to avoid conflicts
# Different RequestResponse[T, U] specializations are distinct handler types
if self._handlers.get(message_type) is not None:
raise ValueError(f"Duplicate handler for type {message_type} in {self.__class__.__name__}")
if self._handlers.get(message_type) is not None:
raise ValueError(f"Duplicate handler for type {message_type} in {self.__class__.__name__}")
# RequestInfoExecutor is allowed to handle SubWorkflowRequestInfo directly
# but other executors should use @intercepts_request
if message_type is SubWorkflowRequestInfo and not isinstance(self, RequestInfoExecutor):
raise ValueError(
f"Executor {self.__class__.__name__} cannot define a handler for "
"SubWorkflowRequestInfo directly. "
f"Use @intercepts_request decorator to intercept sub-workflow requests."
)
# Get the bound method
bound_method = getattr(self, attr_name)
self._handlers[message_type] = bound_method
# Get the bound method
bound_method = getattr(self, attr_name)
self._handlers[message_type] = bound_method
# Add to unified handler specs list
self._handler_specs.append({
"name": handler_spec["name"],
"message_type": message_type,
"output_types": handler_spec.get("output_types", []),
"workflow_output_types": handler_spec.get("workflow_output_types", []),
"ctx_annotation": handler_spec.get("ctx_annotation"),
"source": "class_method", # Distinguish from instance handlers if needed
})
# Discover @intercepts_request methods
if hasattr(attr, "_intercepts_request"):
# Get the bound method for interceptors
bound_method = getattr(self, attr_name)
interceptor_info = {
"method": bound_method,
"from_workflow": getattr(attr, "_from_workflow", None),
"condition": getattr(attr, "_intercept_condition", None),
}
request_type = attr._intercepts_request # type: ignore
if request_type not in self._request_interceptors:
self._request_interceptors[request_type] = []
self._request_interceptors[request_type].append(interceptor_info)
# We need to register the handler for SubWorkflowRequestInfo
# if not already registered.
if SubWorkflowRequestInfo not in self._handlers:
self._handlers[SubWorkflowRequestInfo] = self._handle_sub_workflow_request
self._handler_specs.append({
"name": attr_name,
"message_type": SubWorkflowRequestInfo,
"output_types": [SubWorkflowResponse, SubWorkflowRequestInfo, RequestInfoMessage],
"workflow_output_types": [],
"ctx_annotation": WorkflowContext[
SubWorkflowRequestInfo | SubWorkflowResponse | RequestInfoMessage
],
"source": "class_method",
})
# Add to unified handler specs list
self._handler_specs.append({
"name": handler_spec["name"],
"message_type": message_type,
"output_types": handler_spec.get("output_types", []),
"workflow_output_types": handler_spec.get("workflow_output_types", []),
"ctx_annotation": handler_spec.get("ctx_annotation"),
"source": "class_method", # Distinguish from instance handlers if needed
})
except AttributeError:
# Skip attributes that may not be accessible
continue
async def _handle_sub_workflow_request(
self,
request: "SubWorkflowRequestInfo",
ctx: WorkflowContext["SubWorkflowResponse | SubWorkflowRequestInfo | RequestInfoMessage"],
) -> None:
"""Automatic routing to @intercepts_request methods.
This is only active for executors that have @intercepts_request methods.
"""
# Try to find a matching interceptor for the request and execute it
for request_type, interceptor_list in self._request_interceptors.items():
if self._does_request_match_type(request.data, request_type):
for interceptor_info in interceptor_list:
if self._does_interceptor_apply_to_request(request, interceptor_info):
logger.debug(
f"Executor {self.id} intercepting request {request.request_id} "
f"of type {type(request.data).__name__} from sub-workflow {request.sub_workflow_id}"
)
await self._execute_interceptor(request, interceptor_info, ctx)
return # Only the first matching interceptor is executed
logger.debug(
f"Executor {self.id} has no matching interceptor for request {request.request_id} "
f"of type {type(request.data).__name__} from sub-workflow {request.sub_workflow_id}; forwarding "
"inner request to RequestInfoExecutor for external handling."
)
# No interceptor found - forward inner request to RequestInfoExecutor
await ctx.send_message(request.data)
def _does_request_match_type(self, request_data: Any, request_type: type | str) -> bool:
"""Check if request data matches the expected type.
Args:
request_data: The request data to check
request_type: The type to match against (can be a type or string)
Returns:
True if the request matches the type, False otherwise.
"""
if isinstance(request_type, type):
return is_instance_of(request_data, request_type)
return (
isinstance(request_type, str)
and hasattr(request_data, "__class__")
and request_data.__class__.__name__ == request_type
)
def _does_interceptor_apply_to_request(
self, request: "SubWorkflowRequestInfo", interceptor_info: dict[str, Any]
) -> bool:
"""Check if an interceptor applies to the given request.
Args:
request: The sub-workflow request
interceptor_info: Information about the interceptor
Returns:
True if the interceptor should handle this request, False otherwise.
"""
# Check workflow scope if specified
from_workflow = interceptor_info["from_workflow"]
if from_workflow and request.sub_workflow_id != from_workflow:
return False
# Check additional condition
condition = interceptor_info["condition"]
return not (condition and not condition(request))
async def _execute_interceptor(
self,
request: "SubWorkflowRequestInfo",
interceptor_info: dict[str, Any],
ctx: WorkflowContext["SubWorkflowResponse | SubWorkflowRequestInfo | RequestInfoMessage"],
) -> None:
"""Execute a single interceptor method.
Args:
request: The sub-workflow request
interceptor_info: Information about the interceptor method
ctx: The workflow context
"""
method = interceptor_info["method"]
response = await method(request.data, ctx)
if not isinstance(response, RequestResponse):
raise RuntimeError(
f"Interceptor method {method.__name__} must return RequestResponse, got {type(response)}"
)
# Add automatic correlation info to the response
correlated_response = RequestResponse[RequestInfoMessage, Any].with_correlation(
response,
request.data,
request.request_id,
)
if correlated_response.is_handled:
logger.debug(
f"Executor {self.id}'s interceptor handled request {request.request_id} "
f"of type {type(request.data).__name__} from sub-workflow {request.sub_workflow_id}. "
f"Sending response back to sub-workflow."
)
# Send response back to sub-workflow that made the request
await ctx.send_message(
SubWorkflowResponse(
request_id=request.request_id,
data=correlated_response.data,
),
target_id=request.sub_workflow_id,
)
else:
logger.debug(
f"Executor {self.id}'s interceptor did not handle request {request.request_id} "
f"of type {type(request.data).__name__} from sub-workflow {request.sub_workflow_id}. "
"Forwarding to RequestInfoExecutor for external handling."
)
# Forward the request with potentially modified data
# Update the data if interceptor provided a modified request
if correlated_response.forward_request:
request.data = correlated_response.forward_request
# Send the inner request to RequestInfoExecutor to create external request
await ctx.send_message(request)
def can_handle(self, message: Any) -> bool:
"""Check if the executor can handle a given message type.
@@ -577,13 +401,6 @@ class Executor(AFBaseModel):
output_types: List of output types for send_message()
workflow_output_types: List of workflow output types for yield_output()
"""
if message_type is SubWorkflowRequestInfo:
raise ValueError(
f"Executor {self.__class__.__name__} cannot define a handler for "
"SubWorkflowRequestInfo directly. "
f"Use @intercepts_request decorator to intercept sub-workflow requests."
)
if message_type in self._handlers:
raise ValueError(f"Handler for type {message_type} already registered in {self.__class__.__name__}")
@@ -638,21 +455,6 @@ class Executor(AFBaseModel):
return list(output_types)
@property
def request_types(self) -> list[type[Any]]:
"""Get the list of request types that this executor can intercept via @intercepts_request.
Returns:
A list of the request types that this executor's interceptors can handle.
"""
request_types: list[type[Any]] = []
for request_type in self._request_interceptors:
if isinstance(request_type, type):
request_types.append(request_type)
return request_types
# endregion: Executor
@@ -738,6 +540,11 @@ class RequestInfoMessage:
"""
request_id: str = field(default_factory=lambda: str(uuid.uuid4()))
"""Unique identifier for correlating requests and responses."""
source_executor_id: str | None = None
"""ID of the executor expecting a response to this request.
May differ from the executor that sent the request if intercepted and forwarded."""
TRequest = TypeVar("TRequest", bound="RequestInfoMessage")
@@ -746,217 +553,26 @@ TResponse = TypeVar("TResponse")
@dataclass
class RequestResponse(Generic[TRequest, TResponse]):
"""Response from @intercepts_request methods with automatic correlation support.
"""Response type for request/response correlation in workflows.
This type allows intercepting executors to indicate whether they handled
a request or whether it should be forwarded to external sources. When handled,
the framework automatically adds correlation info to link responses to requests.
This type is used by RequestInfoExecutor to create correlated responses
that include the original request context for proper message routing.
"""
is_handled: bool
data: TResponse | None = None
forward_request: TRequest | None = None
original_request: TRequest | None = None # Added for automatic correlation
request_id: str | None = None # Added for tracking
data: TResponse
"""The response data returned from handling the request."""
@classmethod
def handled(cls, data: TResponse) -> "RequestResponse[TRequest, TResponse]":
"""Create a response indicating the request was handled.
original_request: TRequest
"""The original request that this response corresponds to."""
Correlation info (original_request, request_id) will be added automatically
by the framework when processing intercepted requests.
"""
return cls(is_handled=True, data=data)
@classmethod
def forward(cls, modified_request: Any = None) -> "RequestResponse[TRequest, TResponse]":
"""Create a response indicating the request should be forwarded."""
return cls(is_handled=False, forward_request=modified_request)
@staticmethod
def with_correlation(
original_response: "RequestResponse[TRequest, TResponse]",
original_request: TRequest,
request_id: str,
) -> "RequestResponse[TRequest, TResponse]":
"""Add correlation info to a response.
This is called automatically by the framework when processing intercepted requests.
"""
return RequestResponse(
is_handled=original_response.is_handled,
data=original_response.data,
forward_request=original_response.forward_request,
original_request=original_request,
request_id=request_id,
)
@dataclass
class SubWorkflowRequestInfo:
"""Routes requests from sub-workflows to parent workflows.
This message type wraps requests from sub-workflows to add routing context,
allowing parent workflows to intercept and potentially handle the request.
"""
request_id: str # Original request ID from sub-workflow
sub_workflow_id: str # ID of the WorkflowExecutor that sent this
data: RequestInfoMessage # The actual request data
@dataclass
class SubWorkflowResponse:
"""Routes responses back to sub-workflows.
This message type is used to send responses back to sub-workflows that
made requests, ensuring the response reaches the correct sub-workflow.
"""
request_id: str # Matches the original request ID
data: Any # The actual response data
request_id: str
"""The ID of the original request."""
# endregion: Request/Response Types
# region Intercepts Request Decorator
# TypeVar for request type that must be a RequestInfoMessage subclass
RequestInfoMessageT = TypeVar("RequestInfoMessageT", bound="RequestInfoMessage")
@overload
def intercepts_request(
func: Callable[
[Any, RequestInfoMessageT, WorkflowContext[Any, Any]], Awaitable[RequestResponse[RequestInfoMessageT, Any]]
],
) -> Callable[
[Any, RequestInfoMessageT, WorkflowContext[Any, Any]], Awaitable[RequestResponse[RequestInfoMessageT, Any]]
]: ...
@overload
def intercepts_request(
*,
from_workflow: str | None = None,
condition: Callable[[Any], bool] | None = None,
) -> Callable[
[
Callable[
[Any, RequestInfoMessageT, WorkflowContext[Any, Any]], Awaitable[RequestResponse[RequestInfoMessageT, Any]]
]
],
Callable[
[Any, RequestInfoMessageT, WorkflowContext[Any, Any]], Awaitable[RequestResponse[RequestInfoMessageT, Any]]
],
]: ...
def intercepts_request(
func: Callable[..., Any] | None = None,
*,
from_workflow: str | None = None,
condition: Callable[[Any], bool] | None = None,
) -> Callable[..., Any]:
"""Decorator to mark methods that intercept sub-workflow requests.
The request type is automatically inferred from the method's second parameter type hint.
The type must be a subclass of RequestInfoMessage.
This decorator allows executors in parent workflows to intercept and handle requests from
sub-workflows before they are sent to external sources.
Args:
func: The function being decorated (automatically passed when used without parentheses).
from_workflow: Optional ID of specific sub-workflow to intercept from.
condition: Optional callable that must return True for interception.
Returns:
The decorated function with interception metadata.
Example:
@intercepts_request
async def check_domain(
self, request: DomainCheckRequest, ctx: WorkflowContext
) -> RequestResponse[DomainCheckRequest, bool]:
# Type automatically inferred as DomainCheckRequest from parameter annotation
if request.domain in self.approved_domains:
return RequestResponse.handled(True)
return RequestResponse.forward()
@intercepts_request(from_workflow="email_validator")
async def handle_specific(
self, request: EmailRequest, ctx: WorkflowContext
) -> RequestResponse[EmailRequest, str]:
# Only intercepts EmailRequest from the "email_validator" workflow
return RequestResponse.handled("handled by parent")
"""
def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
# Extract request type from method signature
sig = inspect.signature(func)
params = list(sig.parameters.values())
if len(params) < 2:
raise ValueError(f"Interceptor method '{func.__name__}' must have at least 2 parameters (self, request)")
request_param = params[1] # Second parameter (after self)
request_type = request_param.annotation
if request_type is inspect.Parameter.empty:
raise ValueError(f"Interceptor method '{func.__name__}' request parameter must have a type annotation")
# Runtime validation that it's a RequestInfoMessage subclass
if isinstance(request_type, type):
# We need to check if RequestInfoMessage is defined yet, since this runs at import time
try:
# Try to get RequestInfoMessage from the current module's globals
request_info_message_class = None
func_module = inspect.getmodule(func)
if func_module and hasattr(func_module, "RequestInfoMessage"):
request_info_message_class = func_module.RequestInfoMessage
else:
# Look in the current module (where this decorator is defined)
import sys
current_module = sys.modules[__name__]
if hasattr(current_module, "RequestInfoMessage"):
request_info_message_class = current_module.RequestInfoMessage
if request_info_message_class and not issubclass(request_type, request_info_message_class):
raise TypeError(
f"Interceptor method '{func.__name__}' can only handle RequestInfoMessage subclasses, "
f"got {request_type}. Make sure your request type inherits from RequestInfoMessage."
)
except (AttributeError, NameError):
# RequestInfoMessage might not be defined yet at import time, skip validation
# This will be caught later when the interceptor is actually called
pass
@functools.wraps(func)
async def wrapper(self: Any, request: RequestInfoMessage, ctx: WorkflowContext[Any, Any]) -> Any:
return await func(self, request, ctx)
# Add metadata for discovery - store the inferred type
wrapper._intercepts_request = request_type # type: ignore
wrapper._from_workflow = from_workflow # type: ignore
wrapper._intercept_condition = condition # type: ignore
return wrapper
# If func is provided, we're being called without parentheses: @intercepts_request
if func is not None:
return decorator(func)
# Otherwise, we're being called with parentheses: @intercepts_request(from_workflow="...")
return decorator
# endregion: Intercepts Request Decorator
# region Request Info Executor
class RequestInfoExecutor(Executor):
"""Built-in executor that handles request/response patterns in workflows.
@@ -975,12 +591,12 @@ class RequestInfoExecutor(Executor):
"""
super().__init__(id=id)
self._request_events: dict[str, RequestInfoEvent] = {}
self._sub_workflow_contexts: dict[str, dict[str, str]] = {}
@handler
async def run(self, message: RequestInfoMessage, ctx: WorkflowContext) -> None:
"""Run the RequestInfoExecutor with the given message."""
source_executor_id = ctx.get_source_executor_id()
# Use source_executor_id from message if available, otherwise fall back to context
source_executor_id = message.source_executor_id or ctx.get_source_executor_id()
event = RequestInfoEvent(
request_id=message.request_id,
@@ -992,41 +608,11 @@ class RequestInfoExecutor(Executor):
await self._record_pending_request_snapshot(message, source_executor_id, ctx)
await ctx.add_event(event)
@handler
async def handle_sub_workflow_request(
self,
message: SubWorkflowRequestInfo,
ctx: WorkflowContext,
) -> None:
"""Handle forwarded sub-workflow request.
This method handles requests that were forwarded from parent workflows
because they couldn't be handled locally.
"""
# When called directly from runner, we need to use the sub_workflow_id as the source
source_executor_id = message.sub_workflow_id
# Store context for routing response back
self._sub_workflow_contexts[message.request_id] = {
"sub_workflow_id": message.sub_workflow_id,
"parent_executor_id": source_executor_id,
}
# Create event for external handling - preserve the SubWorkflowRequestInfo wrapper
event = RequestInfoEvent(
request_id=message.request_id, # Use original request ID
source_executor_id=source_executor_id,
request_type=type(message.data), # Type of the wrapped data # type: ignore
request_data=message.data, # The wrapped request data
)
self._request_events[message.request_id] = event
await ctx.add_event(event)
async def handle_response(
self,
response_data: Any,
request_id: str,
ctx: WorkflowContext[SubWorkflowResponse | RequestResponse[RequestInfoMessage, Any]],
ctx: WorkflowContext[RequestResponse[RequestInfoMessage, Any]],
) -> None:
"""Handle a response to a request.
@@ -1043,31 +629,11 @@ class RequestInfoExecutor(Executor):
self._request_events.pop(request_id, None)
# Check if this was a forwarded sub-workflow request
if request_id in self._sub_workflow_contexts:
context = self._sub_workflow_contexts.pop(request_id)
# Send back to sub-workflow that made the original request
await ctx.send_message(
SubWorkflowResponse(
request_id=request_id,
data=response_data,
),
target_id=context["sub_workflow_id"],
)
else:
# Regular response - send directly back to source
# Create a correlated response that includes both the response data and original request
if not isinstance(event.data, RequestInfoMessage):
raise TypeError(f"Expected RequestInfoMessage, got {type(event.data)}")
correlated_response = RequestResponse[RequestInfoMessage, Any].handled(response_data)
correlated_response = RequestResponse[RequestInfoMessage, Any].with_correlation(
correlated_response,
event.data,
request_id,
)
await ctx.send_message(correlated_response, target_id=event.source_executor_id)
# Create a correlated response that includes both the response data and original request
if not isinstance(event.data, RequestInfoMessage):
raise TypeError(f"Expected RequestInfoMessage, got {type(event.data)}")
correlated_response = RequestResponse(data=response_data, original_request=event.data, request_id=request_id)
await ctx.send_message(correlated_response, target_id=event.source_executor_id)
await self._clear_pending_request_snapshot(request_id, ctx)
@@ -1194,7 +760,7 @@ class RequestInfoExecutor(Executor):
return repr(value)
async def has_pending_request(self, request_id: str, ctx: WorkflowContext[Any]) -> bool:
if request_id in self._request_events or request_id in self._sub_workflow_contexts:
if request_id in self._request_events:
return True
snapshot = await self._get_pending_request_snapshot(request_id, ctx)
return snapshot is not None
python/packages/main/agent_framework/_workflow/_runner.py
+2 -106
View File
@@ -22,7 +22,6 @@ from ._runner_context import (
_decode_checkpoint_value,
)
from ._shared_state import SharedState
from ._typing_utils import is_instance_of
logger = logging.getLogger(__name__)
@@ -160,90 +159,6 @@ class Runner:
async def _deliver_messages(source_executor_id: str, messages: list[Message]) -> None:
"""Outer loop to concurrently deliver messages from all sources to their targets."""
# Special handling for SubWorkflowRequestInfo messages
async def _deliver_sub_workflow_requests(messages: list[Message]) -> None:
from ._executor import SubWorkflowRequestInfo
# Handle SubWorkflowRequestInfo messages - only process those not already targeted
sub_workflow_messages: list[Message] = []
for msg in messages:
# Skip messages sent directly to RequestInfoExecutor - they are already forwarded
if self._is_message_to_request_info_executor(msg):
continue
if isinstance(msg.data, SubWorkflowRequestInfo):
sub_workflow_messages.append(msg)
for message in sub_workflow_messages:
# message.data is guaranteed to be SubWorkflowRequestInfo via filtering above
sub_request = message.data # type: ignore[assignment]
# Find executor that can intercept the wrapped type
interceptor_found = False
for executor in self._executors.values():
interceptors = getattr(executor, "_request_interceptors", [])
if interceptors and executor.id != message.source_id:
for registered_type in interceptors: # type: ignore[assignment]
# Check type matching - handle both type and string cases
matched = False
if (
isinstance(registered_type, type)
and is_instance_of(sub_request.data, registered_type)
) or (
isinstance(registered_type, str)
and hasattr(sub_request.data, "__class__")
and sub_request.data.__class__.__name__ == registered_type
):
matched = True
if matched:
# Send directly to the intercepting executor
logger.info(
f"Sending sub-workflow request of type '{sub_request.data.__class__.__name__}' "
f"from sub-workflow '{sub_request.sub_workflow_id}' "
f"to executor '{executor.id}' for interception."
)
await executor.execute(
sub_request,
[message.source_id], # source_executor_ids
self._shared_state, # shared_state
self._ctx, # runner_context
trace_contexts=[message.trace_context] if message.trace_context else None,
source_span_ids=[message.source_span_id] if message.source_span_id else None,
)
interceptor_found = True
break
if interceptor_found:
break
if not interceptor_found:
# No interceptor found - send directly to RequestInfoExecutor if available.
# Find the RequestInfoExecutor instance
request_info_executor = self._find_request_info_executor()
if request_info_executor:
logger.info(
f"Sending sub-workflow request of type '{sub_request.data.__class__.__name__}' "
f"from sub-workflow '{sub_request.sub_workflow_id}' to RequestInfoExecutor "
f"'{request_info_executor.id}'"
)
await request_info_executor.execute(
sub_request,
[message.source_id], # source_executor_ids
self._shared_state, # shared_state
self._ctx, # runner_context
trace_contexts=[message.trace_context] if message.trace_context else None,
source_span_ids=[message.source_span_id] if message.source_span_id else None,
)
else:
logger.warning(
f"Sub-workflow request of type '{sub_request.data.__class__.__name__}' "
f"from sub-workflow '{sub_request.sub_workflow_id}' could not be handled: "
f"no RequestInfoExecutor found in the workflow. Add a RequestInfoExecutor "
f"to handle external requests or add an @intercepts_request handler."
)
async def _deliver_message_inner(edge_runner: EdgeRunner, message: Message) -> bool:
"""Inner loop to deliver a single message through an edge runner."""
return await edge_runner.send_message(message, self._shared_state, self._ctx)
@@ -261,28 +176,9 @@ class Runner:
return
message.data = decoded
# Handle SubWorkflowRequestInfo messages specially
await _deliver_sub_workflow_requests(messages)
# Filter out SubWorkflowRequestInfo messages from normal edge routing
# since they were handled specially
from ._executor import SubWorkflowRequestInfo
non_sub_workflow_messages: list[Message] = []
for msg in messages:
# Keep messages sent directly to RequestInfoExecutor (forwarded messages)
if self._is_message_to_request_info_executor(msg):
non_sub_workflow_messages.append(msg)
continue
# Skip SubWorkflowRequestInfo messages (handled by special routing)
if isinstance(msg.data, SubWorkflowRequestInfo):
continue
non_sub_workflow_messages.append(msg)
# Route all messages through normal workflow edges
associated_edge_runners = self._edge_runner_map.get(source_executor_id, [])
for message in non_sub_workflow_messages:
for message in messages:
_normalize_message_payload(message)
# Deliver a message through all edge runners associated with the source executor concurrently.
tasks = [_deliver_message_inner(edge_runner, message) for edge_runner in associated_edge_runners]
python/packages/main/agent_framework/_workflow/_validation.py
+1 -51
View File
@@ -10,7 +10,6 @@ from typing import Any, Union, get_args, get_origin
from ._edge import Edge, EdgeGroup, FanInEdgeGroup
from ._executor import Executor, RequestInfoExecutor
from ._workflow_executor import WorkflowExecutor
logger = logging.getLogger(__name__)
@@ -187,7 +186,6 @@ class WorkflowGraphValidator:
self._validate_self_loops()
self._validate_dead_ends()
self._validate_cycles()
self._validate_interceptor_uniqueness()
def _validate_handler_output_annotations(self) -> None:
"""Validate that each handler's ctx parameter is annotated with WorkflowContext[T].
@@ -273,9 +271,6 @@ class WorkflowGraphValidator:
# Get output types from source executor
source_output_types = list(source_executor.output_types)
# Also include intercepted request types as potential outputs
# since @intercepts_request methods can forward requests
source_output_types.extend(source_executor.request_types)
# Get input types from target executor
target_input_types = target_executor.input_types
@@ -482,52 +477,6 @@ class WorkflowGraphValidator:
"Ensure proper termination conditions exist to prevent infinite loops."
)
def _validate_interceptor_uniqueness(self) -> None:
"""Validate that only one executor intercepts a given request type from a specific sub-workflow.
This prevents non-deterministic behavior where multiple executors could intercept
the same request type from the same sub-workflow.
"""
# Find all WorkflowExecutor instances in the workflow
workflow_executors: dict[str, WorkflowExecutor] = {}
for executor_id, executor in self._executors.items():
if isinstance(executor, WorkflowExecutor):
workflow_executors[executor_id] = executor
# For each WorkflowExecutor, check which executors can intercept its requests
for workflow_id, _workflow_executor in workflow_executors.items():
# Map of request_type -> list of intercepting executor IDs
interceptors_by_type: dict[type | str, list[str]] = {}
# Find all executors that have edges from this WorkflowExecutor
# These are potential interceptors
for edge in self._edges:
if edge.source_id == workflow_id:
target_executor = self._executors.get(edge.target_id)
if target_executor and hasattr(target_executor, "_request_interceptors"):
# Check what request types this executor intercepts
for request_type, interceptor_list in target_executor._request_interceptors.items():
# Check if any interceptor is scoped to this workflow or unscoped
for interceptor_info in interceptor_list:
from_workflow = interceptor_info.get("from_workflow")
# If unscoped or specifically scoped to this workflow
if from_workflow is None or from_workflow == workflow_id:
if request_type not in interceptors_by_type:
interceptors_by_type[request_type] = []
interceptors_by_type[request_type].append(edge.target_id)
# Check for duplicates
for request_type, executor_ids in interceptors_by_type.items():
unique_executors = list(set(executor_ids)) # Remove duplicates from same executor
if len(unique_executors) > 1:
type_name = request_type.__name__ if isinstance(request_type, type) else str(request_type)
raise InterceptorConflictError(
f"Multiple executors intercept the same request type '{type_name}' "
f"from sub-workflow '{workflow_id}': {', '.join(unique_executors)}. "
f"Only one executor should intercept a given request type from a specific sub-workflow "
f"to ensure deterministic behavior."
)
# endregion
# region Type Compatibility Utilities
@@ -572,6 +521,7 @@ class WorkflowGraphValidator:
for s_arg, t_arg in zip(source_args, target_args, strict=True)
)
# No other special compatibility cases
return False
# endregion
python/packages/main/agent_framework/_workflow/_workflow_executor.py
+68 -33
View File
@@ -1,5 +1,6 @@
# Copyright (c) Microsoft. All rights reserved.
import inspect
import logging
import uuid
from dataclasses import dataclass
@@ -19,10 +20,10 @@ from ._executor import (
Executor,
RequestInfoExecutor,
RequestInfoMessage,
SubWorkflowRequestInfo,
SubWorkflowResponse,
RequestResponse,
handler,
)
from ._typing_utils import is_instance_of
from ._workflow_context import WorkflowContext
logger = logging.getLogger(__name__)
@@ -76,9 +77,11 @@ class WorkflowExecutor(Executor):
request = MyDataRequest(query="user info")
# RequestInfoExecutor emits RequestInfoEvent
# WorkflowExecutor wraps and forwards to parent
wrapped = SubWorkflowRequestInfo(request_id="...", sub_workflow_id="child_workflow", data=request)
# Parent workflow can intercept via @intercepts_request
# WorkflowExecutor sets source_executor_id and forwards to parent
request.source_executor_id = "child_workflow_executor_id"
# Parent workflow can handle via @handler for RequestInfoMessage subclasses,
# or directly forward to external source via a RequestInfoExecutor in the parent
# workflow.
```
### State Management
@@ -103,8 +106,8 @@ class WorkflowExecutor(Executor):
Combines sub-workflow outputs with request coordination types:
```python
# Includes all sub-workflow output types
# Plus SubWorkflowRequestInfo if sub-workflow can make requests
output_types = workflow.output_types + [SubWorkflowRequestInfo] # if applicable
# Plus RequestInfoMessage if sub-workflow can make requests
output_types = workflow.output_types + [RequestInfoMessage] # if applicable
```
## Error Handling
@@ -169,14 +172,20 @@ class WorkflowExecutor(Executor):
Parent workflows can intercept sub-workflow requests:
```python
class ParentExecutor(Executor):
@intercepts_request
async def handle_child_request(
self, request: MyDataRequest, ctx: WorkflowContext[Any]
) -> RequestResponse[MyDataRequest, str]:
@handler
async def handle_request(
self,
request: MyRequestType, # Subclass of RequestInfoMessage
ctx: WorkflowContext[RequestResponse[RequestInfoMessage, Any] | RequestInfoMessage],
) -> None:
# Handle request locally or forward to external source
if self.can_handle_locally(request):
return RequestResponse.handled("local result")
return RequestResponse.forward() # Send to external handler
# Send response back to sub-workflow
response = RequestResponse(data="local result", original_request=request, request_id=request.request_id)
await ctx.send_message(response, target_id=request.source_executor_id)
else:
# Forward to external handler
await ctx.send_message(request)
```
## Implementation Notes
@@ -208,12 +217,18 @@ class WorkflowExecutor(Executor):
@property
def input_types(self) -> list[type[Any]]:
"""Get the input types based on the underlying workflow's input types.
"""Get the input types based on the underlying workflow's input types plus WorkflowExecutor-specific types.
Returns:
A list of input types that the underlying workflow can accept.
A list of input types that the WorkflowExecutor can accept.
"""
return self.workflow.input_types
input_types = list(self.workflow.input_types)
# WorkflowExecutor can also handle RequestResponse for sub-workflow responses
if RequestResponse not in input_types:
input_types.append(RequestResponse)
return input_types
@property
def output_types(self) -> list[type[Any]]:
@@ -221,20 +236,43 @@ class WorkflowExecutor(Executor):
Returns:
A list of output types that the underlying workflow can produce.
Includes SubWorkflowRequestInfo if the sub-workflow contains RequestInfoExecutor.
Includes specific RequestInfoMessage subtypes if the sub-workflow contains RequestInfoExecutor.
"""
output_types = list(self.workflow.output_types)
# Check if the sub-workflow contains a RequestInfoExecutor
# If so, this WorkflowExecutor can also output SubWorkflowRequestInfo messages
for executor in self.workflow.executors.values():
if isinstance(executor, RequestInfoExecutor):
if SubWorkflowRequestInfo not in output_types:
output_types.append(SubWorkflowRequestInfo)
break
# If so, collect the specific RequestInfoMessage subtypes from all executors
has_request_info_executor = any(
isinstance(executor, RequestInfoExecutor) for executor in self.workflow.executors.values()
)
if has_request_info_executor:
# Collect all RequestInfoMessage subtypes from executor output types
for executor in self.workflow.executors.values():
for output_type in executor.output_types:
# Check if this is a RequestInfoMessage subclass
if (
inspect.isclass(output_type)
and issubclass(output_type, RequestInfoMessage)
and output_type not in output_types
):
output_types.append(output_type)
return output_types
def can_handle(self, message: Any) -> bool:
"""Override can_handle to only accept messages that the wrapped workflow can handle.
This prevents the WorkflowExecutor from accepting messages that should go to other
executors (like RequestInfoExecutor).
"""
# Always handle RequestResponse (for the handle_response handler)
if isinstance(message, RequestResponse):
return True
# For other messages, only handle if the wrapped workflow can accept them as input
return any(is_instance_of(message, input_type) for input_type in self.workflow.input_types)
@handler # No output_types - can send any completion data type
async def process_workflow(self, input_data: object, ctx: WorkflowContext[Any]) -> None:
"""Execute the sub-workflow with raw input data.
@@ -246,8 +284,8 @@ class WorkflowExecutor(Executor):
input_data: The input data to send to the sub-workflow.
ctx: The workflow context from the parent.
"""
# Skip SubWorkflowResponse and SubWorkflowRequestInfo - they have specific handlers
if isinstance(input_data, (SubWorkflowResponse, SubWorkflowRequestInfo)):
# Skip RequestResponse - it has a specific handler
if isinstance(input_data, RequestResponse):
logger.debug(f"WorkflowExecutor {self.id} ignoring input of type {type(input_data)}")
return
@@ -318,15 +356,12 @@ class WorkflowExecutor(Executor):
execution_context.pending_requests[event.request_id] = event.data
# Map request to execution for response routing
self._request_to_execution[event.request_id] = execution_context.execution_id
# Wrap request with routing context and send to parent
# Set source_executor_id for response routing and send to parent
if not isinstance(event.data, RequestInfoMessage):
raise TypeError(f"Expected RequestInfoMessage, got {type(event.data)}")
wrapped_request = SubWorkflowRequestInfo(
request_id=event.request_id,
sub_workflow_id=self.id,
data=event.data,
)
await ctx.send_message(wrapped_request)
# Set the source_executor_id to this WorkflowExecutor's ID for response routing
event.data.source_executor_id = self.id
await ctx.send_message(event.data)
# Update expected response count for this execution
execution_context.expected_response_count = len(request_info_events)
@@ -375,7 +410,7 @@ class WorkflowExecutor(Executor):
@handler
async def handle_response(
self,
response: SubWorkflowResponse,
response: RequestResponse[RequestInfoMessage, Any],
ctx: WorkflowContext[Any],
) -> None:
"""Handle response from parent for a forwarded request.
python/packages/main/tests/workflow/test_checkpoint_decode.py
+8 -10
View File
@@ -14,11 +14,10 @@ class SampleRequest(RequestInfoMessage):
def test_decode_dataclass_with_nested_request() -> None:
original = RequestResponse[SampleRequest, str].handled("approve")
original = RequestResponse[SampleRequest, str].with_correlation(
original,
SampleRequest(request_id="abc", prompt="prompt"),
"abc",
original = RequestResponse[SampleRequest, str](
data="approve",
original_request=SampleRequest(request_id="abc", prompt="prompt"),
request_id="abc",
)
encoded = _encode_checkpoint_value(original)
@@ -32,11 +31,10 @@ def test_decode_dataclass_with_nested_request() -> None:
def test_is_instance_of_coerces_request_response_original_request_dict() -> None:
response = RequestResponse[SampleRequest, str].handled("approve")
response = RequestResponse[SampleRequest, str].with_correlation(
response,
SampleRequest(request_id="req-1", prompt="prompt"),
"req-1",
response = RequestResponse[SampleRequest, str](
data="approve",
original_request=SampleRequest(request_id="req-1", prompt="prompt"),
request_id="req-1",
)
# Simulate checkpoint decode fallback leaving a dict
python/packages/main/tests/workflow/test_request_info_executor_rehydrate.py
+4 -5
View File
@@ -171,11 +171,10 @@ def test_pending_requests_from_checkpoint_and_summary() -> None:
request = SimpleApproval(prompt="Review draft", draft="Draft text", iteration=3)
request.request_id = "req-42"
response = RequestResponse[SimpleApproval, str].handled("approve")
response = RequestResponse[SimpleApproval, str].with_correlation(
response,
request,
request.request_id,
response = RequestResponse[SimpleApproval, str](
data="approve",
original_request=request,
request_id=request.request_id,
)
encoded_response = _encode_checkpoint_value(response)
python/packages/main/tests/workflow/test_simple_sub_workflow.py
-108
View File
@@ -1,108 +0,0 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
from dataclasses import dataclass
from typing_extensions import Never
from agent_framework import (
Executor,
WorkflowBuilder,
WorkflowContext,
WorkflowExecutor,
handler,
)
@dataclass
class SimpleRequest:
"""Simple request for testing."""
text: str
@dataclass
class SimpleResponse:
"""Simple response for testing."""
result: str
class SimpleSubExecutor(Executor):
"""Simple executor for sub-workflow."""
def __init__(self):
super().__init__(id="simple_sub")
@handler
async def process(self, request: SimpleRequest, ctx: WorkflowContext[Never, SimpleResponse]) -> None:
"""Process a simple request."""
# Just echo back with prefix and complete
response = SimpleResponse(result=f"processed: {request.text}")
await ctx.yield_output(response)
class SimpleParent(Executor):
"""Simple parent executor."""
result: SimpleResponse | None = None
def __init__(self):
super().__init__(id="simple_parent")
@handler
async def start(self, text: str, ctx: WorkflowContext[SimpleRequest]) -> None:
"""Start the process."""
request = SimpleRequest(text=text)
await ctx.send_message(request, target_id="sub_workflow")
@handler
async def collect(self, response: SimpleResponse, ctx: WorkflowContext) -> None:
"""Collect the result."""
self.result = response
async def test_simple_sub_workflow():
"""Test the simplest possible sub-workflow."""
# Create sub-workflow with dummy executor to satisfy validation
sub_executor = SimpleSubExecutor()
class DummyExecutor(Executor):
def __init__(self):
super().__init__(id="dummy")
@handler
async def process(self, message: object, ctx: WorkflowContext) -> None:
pass # Do nothing
dummy = DummyExecutor()
sub_workflow = (
WorkflowBuilder()
.set_start_executor(sub_executor)
.add_edge(sub_executor, dummy) # Add edge to satisfy validation
.build()
)
# Create parent workflow
parent = SimpleParent()
workflow_executor = WorkflowExecutor(sub_workflow, id="sub_workflow")
main_workflow = (
WorkflowBuilder()
.set_start_executor(parent)
.add_edge(parent, workflow_executor)
.add_edge(workflow_executor, parent)
.build()
)
# Run the workflow
await main_workflow.run("hello world")
# Check result
assert parent.result is not None
assert parent.result.result == "processed: hello world"
if __name__ == "__main__":
# Run the simple test
asyncio.run(test_simple_sub_workflow())
python/packages/main/tests/workflow/test_sub_workflow.py
+129 -198
View File
@@ -1,6 +1,5 @@
# Copyright (c) Microsoft. All rights reserved.
import asyncio
from dataclasses import dataclass
from typing import Any
@@ -12,11 +11,11 @@ from agent_framework import (
RequestInfoExecutor,
RequestInfoMessage,
RequestResponse,
Workflow,
WorkflowBuilder,
WorkflowContext,
WorkflowExecutor,
handler,
intercepts_request,
)
@@ -45,6 +44,61 @@ class ValidationResult:
reason: str
# Test helper functions
def create_email_validation_workflow() -> Workflow:
"""Create a standard email validation workflow."""
email_validator = EmailValidator()
email_request_info = RequestInfoExecutor(id="email_request_info")
return (
WorkflowBuilder()
.set_start_executor(email_validator)
.add_edge(email_validator, email_request_info)
.add_edge(email_request_info, email_validator)
.build()
)
class BasicParent(Executor):
"""Basic parent executor for simple sub-workflow tests."""
result: ValidationResult | None = Field(default=None)
cache: dict[str, bool] = Field(default_factory=dict)
def __init__(self, cache: dict[str, bool] | None = None, **kwargs: Any):
if cache is not None:
kwargs["cache"] = cache
super().__init__(id="basic_parent", **kwargs)
@handler
async def start(self, email: str, ctx: WorkflowContext[EmailValidationRequest]) -> None:
request = EmailValidationRequest(email=email)
await ctx.send_message(request, target_id="email_workflow")
@handler
async def handle_domain_request(
self,
request: DomainCheckRequest,
ctx: WorkflowContext[RequestResponse[DomainCheckRequest, Any] | DomainCheckRequest],
) -> None:
"""Handle requests from sub-workflows with optional caching."""
domain_request = request
if domain_request.domain in self.cache:
# Return cached result
response = RequestResponse(
data=self.cache[domain_request.domain], original_request=request, request_id=request.request_id
)
await ctx.send_message(response, target_id=request.source_executor_id)
else:
# Not in cache, forward to external
await ctx.send_message(request)
@handler
async def collect(self, result: ValidationResult, ctx: WorkflowContext) -> None:
self.result = result
# Test executors
class EmailValidator(Executor):
"""Validates email addresses in a sub-workflow."""
@@ -54,7 +108,7 @@ class EmailValidator(Executor):
@handler
async def validate_request(
self, request: EmailValidationRequest, ctx: WorkflowContext[RequestInfoMessage, ValidationResult]
self, request: EmailValidationRequest, ctx: WorkflowContext[DomainCheckRequest, ValidationResult]
) -> None:
"""Validate an email address."""
# Extract domain and check if it's approved
@@ -101,17 +155,23 @@ class ParentOrchestrator(Executor):
request = EmailValidationRequest(email=email)
await ctx.send_message(request, target_id="email_workflow")
@intercepts_request
async def check_domain(
self, request: DomainCheckRequest, ctx: WorkflowContext[Any]
) -> RequestResponse[DomainCheckRequest, bool]:
"""Intercept domain check requests from sub-workflows."""
# Check if we know this domain
if request.domain in self.approved_domains:
return RequestResponse[DomainCheckRequest, bool].handled(True)
@handler
async def handle_domain_request(
self,
request: DomainCheckRequest,
ctx: WorkflowContext[RequestResponse[DomainCheckRequest, Any] | DomainCheckRequest],
) -> None:
"""Handle requests from sub-workflows."""
domain_request = request
# We don't know this domain, forward to external
return RequestResponse[DomainCheckRequest, bool].forward()
# Check if we know this domain
if domain_request.domain in self.approved_domains:
# Send response back to sub-workflow
response = RequestResponse(data=True, original_request=request, request_id=request.request_id)
await ctx.send_message(response, target_id=request.source_executor_id)
else:
# We don't know this domain, forward to external
await ctx.send_message(request)
@handler
async def collect_result(self, result: ValidationResult, ctx: WorkflowContext) -> None:
@@ -122,34 +182,10 @@ class ParentOrchestrator(Executor):
async def test_basic_sub_workflow() -> None:
"""Test basic sub-workflow execution without interception."""
# Create sub-workflow
email_validator = EmailValidator()
email_request_info = RequestInfoExecutor(id="email_request_info")
validation_workflow = (
WorkflowBuilder()
.set_start_executor(email_validator)
.add_edge(email_validator, email_request_info)
.add_edge(email_request_info, email_validator)
.build()
)
validation_workflow = create_email_validation_workflow()
# Create parent workflow without interception
class SimpleParent(Executor):
result: ValidationResult | None = Field(default=None)
def __init__(self, **kwargs: Any):
super().__init__(id="simple_parent", **kwargs)
@handler
async def start(self, email: str, ctx: WorkflowContext[EmailValidationRequest]) -> None:
request = EmailValidationRequest(email=email)
await ctx.send_message(request, target_id="email_workflow")
@handler
async def collect(self, result: ValidationResult, ctx: WorkflowContext) -> None:
self.result = result
parent = SimpleParent()
parent = BasicParent()
workflow_executor = WorkflowExecutor(validation_workflow, "email_workflow")
main_request_info = RequestInfoExecutor(id="main_request_info")
@@ -159,7 +195,7 @@ async def test_basic_sub_workflow() -> None:
.add_edge(parent, workflow_executor)
.add_edge(workflow_executor, parent)
.add_edge(workflow_executor, main_request_info)
.add_edge(main_request_info, workflow_executor) # CRITICAL: For SubWorkflowResponse routing
.add_edge(main_request_info, workflow_executor) # CRITICAL: For RequestResponse routing
.build()
)
@@ -184,21 +220,12 @@ async def test_basic_sub_workflow() -> None:
async def test_sub_workflow_with_interception():
"""Test sub-workflow with parent interception of requests."""
"""Test sub-workflow with parent interception and conditional forwarding."""
# Create sub-workflow
email_validator = EmailValidator()
email_request_info = RequestInfoExecutor(id="email_request_info")
validation_workflow = create_email_validation_workflow()
validation_workflow = (
WorkflowBuilder()
.set_start_executor(email_validator)
.add_edge(email_validator, email_request_info)
.add_edge(email_request_info, email_validator)
.build()
)
# Create parent workflow with interception
parent = ParentOrchestrator(approved_domains={"example.com", "internal.org"})
# Create parent workflow with interception cache
parent = BasicParent(cache={"example.com": True, "internal.org": True})
workflow_executor = WorkflowExecutor(validation_workflow, "email_workflow")
parent_request_info = RequestInfoExecutor(id="request_info")
@@ -208,29 +235,23 @@ async def test_sub_workflow_with_interception():
.add_edge(parent, workflow_executor)
.add_edge(workflow_executor, parent)
.add_edge(parent, parent_request_info) # For forwarded requests
.add_edge(parent_request_info, workflow_executor) # For SubWorkflowResponse routing
.add_edge(parent_request_info, workflow_executor) # For RequestResponse routing
.build()
)
# Test 1: Email with known domain (intercepted)
result = await main_workflow.run(["user@example.com"])
# Should complete without external requests
# Test 1: Email with cached domain (intercepted)
result = await main_workflow.run("user@example.com")
request_events = result.get_request_info_events()
assert len(request_events) == 0 # No external requests, handled internally
assert len(request_events) == 0 # No external requests, handled from cache
assert parent.result is not None
assert parent.result.email == "user@example.com"
assert parent.result.is_valid is True
assert len(parent.results) == 1
assert parent.results[0].email == "user@example.com"
assert parent.results[0].is_valid is True
assert parent.results[0].reason == "Domain approved"
# Test 2: Email with unknown domain (forwarded)
parent.results.clear()
result = await main_workflow.run(["user@unknown.com"])
# Should have external request
# Test 2: Email with unknown domain (forwarded to external)
parent.result = None
result = await main_workflow.run("user@unknown.com")
request_events = result.get_request_info_events()
assert len(request_events) == 1
assert len(request_events) == 1 # Forwarded to external
assert isinstance(request_events[0].data, DomainCheckRequest)
assert request_events[0].data.domain == "unknown.com"
@@ -238,89 +259,18 @@ async def test_sub_workflow_with_interception():
await main_workflow.send_responses({
request_events[0].request_id: False # Domain not approved
})
assert parent.result is not None
assert parent.result.email == "user@unknown.com"
assert parent.result.is_valid is False
assert len(parent.results) == 1
assert parent.results[0].email == "user@unknown.com"
assert parent.results[0].is_valid is False
assert parent.results[0].reason == "Domain not approved"
async def test_conditional_forwarding() -> None:
"""Test conditional forwarding with RequestResponse.forward()."""
class ConditionalParent(Executor):
"""Parent that conditionally handles requests."""
cache: dict[str, bool] = Field(default_factory=lambda: {"cached.com": True})
result: ValidationResult | None = Field(default=None)
def __init__(self, **kwargs: Any):
super().__init__(id="conditional_parent", **kwargs)
@handler
async def start(self, email: str, ctx: WorkflowContext[EmailValidationRequest]) -> None:
request = EmailValidationRequest(email=email)
await ctx.send_message(request, target_id="email_workflow")
@intercepts_request
async def check_domain(
self, request: DomainCheckRequest, ctx: WorkflowContext[Any]
) -> RequestResponse[DomainCheckRequest, bool]:
"""Check cache first, then forward if not found."""
if request.domain in self.cache:
# Return cached result
return RequestResponse[DomainCheckRequest, bool].handled(self.cache[request.domain])
# Not in cache, forward to external
return RequestResponse[DomainCheckRequest, bool].forward()
@handler
async def collect(self, result: ValidationResult, ctx: WorkflowContext) -> None:
self.result = result
# Setup workflows
email_validator = EmailValidator()
request_info = RequestInfoExecutor(id="request_info")
validation_workflow = (
WorkflowBuilder()
.set_start_executor(email_validator)
.add_edge(email_validator, request_info)
.add_edge(request_info, email_validator)
.build()
)
parent = ConditionalParent()
workflow_executor = WorkflowExecutor(validation_workflow, "email_workflow")
parent_request_info = RequestInfoExecutor(id="request_info")
main_workflow = (
WorkflowBuilder()
.set_start_executor(parent)
.add_edge(parent, workflow_executor)
.add_edge(workflow_executor, parent)
.add_edge(parent, parent_request_info)
.add_edge(parent_request_info, workflow_executor) # For SubWorkflowResponse routing
.build()
)
# Test cached domain
result = await main_workflow.run("user@cached.com")
# Test 3: Another cached domain
parent.result = None
result = await main_workflow.run("user@internal.org")
request_events = result.get_request_info_events()
assert len(request_events) == 0 # Handled from cache
assert parent.result is not None
assert parent.result.is_valid is True
# Test uncached domain
parent.result = None
result = await main_workflow.run("user@new.com")
request_events = result.get_request_info_events()
assert len(request_events) == 1 # Forwarded to external
await main_workflow.send_responses({request_events[0].request_id: True})
assert parent.result is not None
assert parent.result.is_valid is True
async def test_workflow_scoped_interception() -> None:
"""Test interception scoped to specific sub-workflows."""
@@ -339,42 +289,41 @@ async def test_workflow_scoped_interception() -> None:
await ctx.send_message(EmailValidationRequest(email=data["email1"]), target_id="workflow_a")
await ctx.send_message(EmailValidationRequest(email=data["email2"]), target_id="workflow_b")
@intercepts_request(from_workflow="workflow_a")
async def check_domain_a(
self, request: DomainCheckRequest, ctx: WorkflowContext[Any]
) -> RequestResponse[DomainCheckRequest, bool]:
"""Strict rules for workflow A."""
if request.domain == "strict.com":
return RequestResponse[DomainCheckRequest, bool].handled(True)
return RequestResponse[DomainCheckRequest, bool].forward()
@handler
async def handle_domain_request(
self,
request: DomainCheckRequest,
ctx: WorkflowContext[RequestResponse[DomainCheckRequest, Any] | DomainCheckRequest],
) -> None:
domain_request = request
@intercepts_request(from_workflow="workflow_b")
async def check_domain_b(
self, request: DomainCheckRequest, ctx: WorkflowContext[Any]
) -> RequestResponse[DomainCheckRequest, bool]:
"""Lenient rules for workflow B."""
if request.domain.endswith(".com"):
return RequestResponse[DomainCheckRequest, bool].handled(True)
return RequestResponse[DomainCheckRequest, bool].forward()
if request.source_executor_id == "workflow_a":
# Strict rules for workflow A
if domain_request.domain == "strict.com":
response = RequestResponse(data=True, original_request=request, request_id=request.request_id)
await ctx.send_message(response, target_id=request.source_executor_id)
else:
# Forward to external
await ctx.send_message(request)
elif request.source_executor_id == "workflow_b":
# Lenient rules for workflow B
if domain_request.domain.endswith(".com"):
response = RequestResponse(data=True, original_request=request, request_id=request.request_id)
await ctx.send_message(response, target_id=request.source_executor_id)
else:
# Forward to external
await ctx.send_message(request)
else:
# Unknown source, forward to external
await ctx.send_message(request)
@handler
async def collect(self, result: ValidationResult, ctx: WorkflowContext) -> None:
self.results[result.email] = result
# Create two identical sub-workflows
def create_validation_workflow():
validator = EmailValidator()
request_info = RequestInfoExecutor(id="request_info")
return (
WorkflowBuilder()
.set_start_executor(validator)
.add_edge(validator, request_info)
.add_edge(request_info, validator)
.build()
)
workflow_a = create_validation_workflow()
workflow_b = create_validation_workflow()
workflow_a = create_email_validation_workflow()
workflow_b = create_email_validation_workflow()
parent = MultiWorkflowParent()
executor_a = WorkflowExecutor(workflow_a, "workflow_a")
@@ -389,8 +338,8 @@ async def test_workflow_scoped_interception() -> None:
.add_edge(executor_a, parent)
.add_edge(executor_b, parent)
.add_edge(parent, parent_request_info)
.add_edge(parent_request_info, executor_a) # For SubWorkflowResponse routing
.add_edge(parent_request_info, executor_b) # For SubWorkflowResponse routing
.add_edge(parent_request_info, executor_a) # For RequestResponse routing
.add_edge(parent_request_info, executor_b) # For RequestResponse routing
.build()
)
@@ -432,16 +381,7 @@ async def test_concurrent_sub_workflow_execution() -> None:
self.results.append(result)
# Create sub-workflow for email validation
email_validator = EmailValidator()
email_request_info = RequestInfoExecutor(id="email_request_info")
validation_workflow = (
WorkflowBuilder()
.set_start_executor(email_validator)
.add_edge(email_validator, email_request_info)
.add_edge(email_request_info, email_validator)
.build()
)
validation_workflow = create_email_validation_workflow()
# Create parent workflow
processor = ConcurrentProcessor()
@@ -454,7 +394,7 @@ async def test_concurrent_sub_workflow_execution() -> None:
.add_edge(processor, workflow_executor)
.add_edge(workflow_executor, processor)
.add_edge(workflow_executor, parent_request_info) # For external requests
.add_edge(parent_request_info, workflow_executor) # For SubWorkflowResponse routing
.add_edge(parent_request_info, workflow_executor) # For RequestResponse routing
.build()
)
@@ -497,12 +437,3 @@ async def test_concurrent_sub_workflow_execution() -> None:
# Verify that concurrent executions were properly isolated
# (This is implicitly tested by the fact that we got correct results for all emails)
if __name__ == "__main__":
# Run tests
asyncio.run(test_basic_sub_workflow())
asyncio.run(test_sub_workflow_with_interception())
asyncio.run(test_conditional_forwarding())
asyncio.run(test_workflow_scoped_interception())
asyncio.run(test_concurrent_sub_workflow_execution())
python/packages/main/tests/workflow/test_typing_utils.py
+2 -1
View File
@@ -95,7 +95,8 @@ def test_request_response_type() -> None:
"""Test RequestResponse generic type checking."""
request_instance = RequestResponse[RequestInfoMessage, str](
is_handled=False,
data="approve",
request_id="req-1",
original_request=RequestInfoMessage(),
)
python/samples/getting_started/workflow/README.md
+2 -2
View File
@@ -49,8 +49,8 @@ Once comfortable with these, explore the rest of the samples below.
| Sample | File | Concepts |
|---|---|---|
| Sub-Workflow (Basics) | [composition/sub_workflow_basics.py](./composition/sub_workflow_basics.py) | Wrap a workflow as an executor and orchestrate sub-workflows |
| Sub-Workflow: Request Interception | [composition/sub_workflow_request_interception.py](./composition/sub_workflow_request_interception.py) | Intercept and forward requests with decorators and request handling |
| Sub-Workflow: Parallel Requests | [composition/sub_workflow_parallel_requests.py](./composition/sub_workflow_parallel_requests.py) | Multi-type interception and external forwarding patterns |
| Sub-Workflow: Request Interception | [composition/sub_workflow_request_interception.py](./composition/sub_workflow_request_interception.py) | Intercept and forward sub-workflow requests using @handler for RequestInfoMessage subclasses |
| Sub-Workflow: Parallel Requests | [composition/sub_workflow_parallel_requests.py](./composition/sub_workflow_parallel_requests.py) | Multiple specialized interceptors handling different request types from same sub-workflow |
### control-flow
| Sample | File | Concepts |
python/samples/getting_started/workflow/composition/sub_workflow_parallel_requests.py
+92 -91
View File
@@ -9,71 +9,60 @@ from typing_extensions import Never
from agent_framework import (
Executor,
RequestInfoExecutor,
RequestInfoMessage,
RequestResponse,
WorkflowBuilder,
WorkflowContext,
WorkflowExecutor,
handler,
)
# Import the new sub-workflow types directly from the implementation package
try:
from agent_framework import (
RequestInfoMessage,
RequestResponse,
WorkflowExecutor,
intercepts_request,
)
except ImportError:
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "packages", "workflow"))
from agent_framework import (
RequestInfoMessage,
RequestResponse,
WorkflowExecutor,
intercepts_request,
)
"""
Sample: Sub-workflow with parallel requests
Sample: Sub-workflow with parallel request handling by specialized interceptors
This sample demonstrates the PROPER pattern for request interception.
This sample demonstrates how different parent executors can handle different types of requests
from the same sub-workflow using regular @handler methods for RequestInfoMessage subclasses.
Prerequisites:
- No external services required (external handling simulated via `RequestInfoExecutor`).
Key principles:
1. Only ONE executor intercepts a given request type from a specific sub-workflow
2. Different executors can intercept DIFFERENT request types from the same sub-workflow
3. The same executor can intercept the same request type from DIFFERENT sub-workflows
This ensures:
- Deterministic behavior
- Clear responsibility boundaries
- Easier debugging and maintenance
Key architectural principles:
1. Specialized interceptors: Each parent executor handles only specific request types
2. Type-based routing: ResourceCache handles ResourceRequest, PolicyEngine handles PolicyCheckRequest
3. Automatic type filtering: Each interceptor only receives requests with matching types
4. Fallback forwarding: Unhandled requests are forwarded to external services
The example simulates a resource allocation system where:
- Sub-workflow requests resources (CPU, memory, etc.)
- A single Cache executor intercepts and handles resource requests
- The Cache can either satisfy from cache or forward to external
- Sub-workflow makes mixed requests for resources (CPU, memory) and policy checks
- ResourceCache executor intercepts ResourceRequest messages, serves from cache or forwards
- PolicyEngine executor intercepts PolicyCheckRequest messages, applies rules or forwards
- Each interceptor uses typed @handler methods for automatic filtering
Simple flow visualization:
Flow visualization:
Coordinator
|
| list[resource/policy requests]
| Mixed list[resource + policy requests]
v
[ Sub-workflow: WorkflowExecutor(ResourceRequester) ]
| |
| ResourceRequest | PolicyCheckRequest
v v
ResourceCache (@intercepts) PolicyEngine (@intercepts)
| handled/forward | handled/forward
v v
RequestInfo (external) <----- forwarded when not handled
| responses
|
| Emits different RequestInfoMessage types:
| - ResourceRequest
| - PolicyCheckRequest
v
Back to sub-workflow -> completion -> results collected
Parent workflow routes to specialized handlers:
| |
| ResourceCache.handle_resource_request | PolicyEngine.handle_policy_request
| (@handler ResourceRequest) | (@handler PolicyCheckRequest)
v v
Cache hit/miss decision Policy allow/deny decision
| |
| RequestResponse OR forward | RequestResponse OR forward
v v
Back to sub-workflow <----------> External RequestInfoExecutor
|
v
External responses route back
"""
@@ -201,9 +190,9 @@ class ResourceRequester(Executor):
return self._request_count == 0
# 3. Implement the Resource Cache - ONLY intercepts ResourceRequest
# 3. Implement the Resource Cache - Uses typed handler for ResourceRequest
class ResourceCache(Executor):
"""Interceptor that handles RESOURCE requests from cache."""
"""Interceptor that handles RESOURCE requests from cache using typed routing."""
# Use class attributes to avoid Pydantic assignment restrictions
cache: dict[str, int] = {"cpu": 10, "memory": 50, "disk": 100}
@@ -213,26 +202,32 @@ class ResourceCache(Executor):
super().__init__(id="resource_cache")
# Instance initialization only; state kept in class attributes as above
@intercepts_request
async def check_cache(
self, request: ResourceRequest, ctx: WorkflowContext
) -> RequestResponse[ResourceRequest, ResourceResponse]:
"""Intercept RESOURCE requests and check cache first."""
print(f"🏪 CACHE interceptor checking: {request.amount} {request.resource_type}")
@handler
async def handle_resource_request(
self, request: ResourceRequest, ctx: WorkflowContext[RequestResponse[ResourceRequest, Any] | ResourceRequest]
) -> None:
"""Handle RESOURCE requests from sub-workflows and check cache first."""
resource_request = request
print(f"🏪 CACHE interceptor checking: {resource_request.amount} {resource_request.resource_type}")
available = self.cache.get(request.resource_type, 0)
available = self.cache.get(resource_request.resource_type, 0)
if available >= request.amount:
if available >= resource_request.amount:
# We can satisfy from cache
self.cache[request.resource_type] -= request.amount
response = ResourceResponse(resource_type=request.resource_type, allocated=request.amount, source="cache")
print(f" ✅ Cache satisfied: {request.amount} {request.resource_type}")
self.results.append(response)
return RequestResponse[ResourceRequest, ResourceResponse].handled(response)
self.cache[resource_request.resource_type] -= resource_request.amount
response_data = ResourceResponse(
resource_type=resource_request.resource_type, allocated=resource_request.amount, source="cache"
)
print(f" ✅ Cache satisfied: {resource_request.amount} {resource_request.resource_type}")
self.results.append(response_data)
# Cache miss - forward to external
print(f" ❌ Cache miss: need {request.amount}, have {available} {request.resource_type}")
return RequestResponse[ResourceRequest, ResourceResponse].forward()
# Send response back to sub-workflow
response = RequestResponse(data=response_data, original_request=request, request_id=request.request_id)
await ctx.send_message(response, target_id=request.source_executor_id)
else:
# Cache miss - forward to external
print(f" ❌ Cache miss: need {resource_request.amount}, have {available} {resource_request.resource_type}")
await ctx.send_message(request)
@handler
async def collect_result(
@@ -247,9 +242,9 @@ class ResourceCache(Executor):
)
# 4. Implement the Policy Engine - ONLY intercepts PolicyCheckRequest (different type!)
# 4. Implement the Policy Engine - Uses typed handler for PolicyCheckRequest
class PolicyEngine(Executor):
"""Interceptor that handles POLICY requests."""
"""Interceptor that handles POLICY requests using typed routing."""
# Use class attributes for simple sample state
quota: dict[str, int] = {
@@ -263,28 +258,35 @@ class PolicyEngine(Executor):
super().__init__(id="policy_engine")
# Instance initialization only; state kept in class attributes as above
@intercepts_request
async def check_policy(
self, request: PolicyCheckRequest, ctx: WorkflowContext
) -> RequestResponse[PolicyCheckRequest, PolicyResponse]:
"""Intercept POLICY requests and apply rules."""
print(f"🛡️ POLICY interceptor checking: {request.amount} {request.resource_type}, policy={request.policy_type}")
@handler
async def handle_policy_request(
self, request: PolicyCheckRequest, ctx: WorkflowContext[RequestResponse[PolicyCheckRequest, Any] | PolicyCheckRequest]
) -> None:
"""Handle POLICY requests from sub-workflows and apply rules."""
policy_request = request
print(f"🛡️ POLICY interceptor checking: {policy_request.amount} {policy_request.resource_type}, policy={policy_request.policy_type}")
quota_limit = self.quota.get(request.resource_type, 0)
quota_limit = self.quota.get(policy_request.resource_type, 0)
if policy_request.policy_type == "quota":
if policy_request.amount <= quota_limit:
response_data = PolicyResponse(approved=True, reason=f"Within quota ({quota_limit})")
print(f" ✅ Policy approved: {policy_request.amount} <= {quota_limit}")
self.results.append(response_data)
# Send response back to sub-workflow
response = RequestResponse(data=response_data, original_request=request, request_id=request.request_id)
await ctx.send_message(response, target_id=request.source_executor_id)
return
if request.policy_type == "quota":
if request.amount <= quota_limit:
response = PolicyResponse(approved=True, reason=f"Within quota ({quota_limit})")
print(f" ✅ Policy approved: {request.amount} <= {quota_limit}")
self.results.append(response)
return RequestResponse[PolicyCheckRequest, PolicyResponse].handled(response)
# Exceeds quota - forward to external for review
print(f" ❌ Policy exceeds quota: {request.amount} > {quota_limit}, forwarding to external")
return RequestResponse[PolicyCheckRequest, PolicyResponse].forward()
print(f" ❌ Policy exceeds quota: {policy_request.amount} > {quota_limit}, forwarding to external")
await ctx.send_message(request)
return
# Unknown policy type - forward to external
print(f" ❓ Unknown policy type: {request.policy_type}, forwarding")
return RequestResponse[PolicyCheckRequest, PolicyResponse].forward()
print(f" ❓ Unknown policy type: {policy_request.policy_type}, forwarding")
await ctx.send_message(request)
@handler
async def collect_policy_result(
@@ -340,20 +342,19 @@ async def main() -> None:
# Create a simple coordinator that starts the process
coordinator = Coordinator()
# PROPER PATTERN: Each executor intercepts DIFFERENT request types
# TYPED ROUTING: Each executor handles specific typed RequestInfoMessage messages
main_workflow = (
WorkflowBuilder()
.set_start_executor(coordinator)
.add_edge(coordinator, workflow_executor) # Start sub-workflow
.add_edge(workflow_executor, coordinator) # Sub-workflow completion back to coordinator
.add_edge(workflow_executor, cache) # Cache intercepts ResourceRequest
.add_edge(cache, workflow_executor) # Cache handles ResourceRequest
.add_edge(workflow_executor, policy) # Policy handles PolicyCheckRequest
.add_edge(policy, workflow_executor) # Policy intercepts PolicyCheckRequest
.add_edge(cache, main_request_info) # Cache forwards to external
.add_edge(policy, main_request_info) # Policy forwards to external
.add_edge(main_request_info, workflow_executor) # External responses back
.add_edge(workflow_executor, main_request_info) # Sub-workflow forwards to main
.add_edge(workflow_executor, cache) # WorkflowExecutor sends ResourceRequest to cache
.add_edge(workflow_executor, policy) # WorkflowExecutor sends PolicyCheckRequest to policy
.add_edge(cache, workflow_executor) # Cache sends RequestResponse back
.add_edge(policy, workflow_executor) # Policy sends RequestResponse back
.add_edge(cache, main_request_info) # Cache forwards ResourceRequest to external
.add_edge(policy, main_request_info) # Policy forwards PolicyCheckRequest to external
.add_edge(main_request_info, workflow_executor) # External responses back to sub-workflow
.build()
)
python/samples/getting_started/workflow/composition/sub_workflow_request_interception.py
+29 -20
View File
@@ -12,7 +12,6 @@ from agent_framework import (
WorkflowContext,
WorkflowExecutor,
handler,
intercepts_request,
)
"""
@@ -20,8 +19,8 @@ Sample: Sub-Workflows with Request Interception
This sample shows how to:
1. Create workflows that execute other workflows as sub-workflows
2. Intercept requests from sub-workflows in parent workflows using @intercepts_request
3. Conditionally handle or forward requests using RequestResponse.handled() and RequestResponse.forward()
2. Intercept requests from sub-workflows using an executor with @handler for RequestInfoMessage subclasses
3. Conditionally handle or forward requests using RequestResponse messages
4. Handle external requests that are forwarded by the parent workflow
5. Proper request/response correlation for concurrent processing
@@ -35,9 +34,8 @@ The example simulates an email validation system where:
Key concepts demonstrated:
- WorkflowExecutor: Wraps a workflow to make it behave as an executor
- @intercepts_request: Decorator for parent workflows to handle sub-workflow requests
- RequestResponse: Enables conditional handling vs forwarding of requests
- Request correlation: Using request_id to match responses with original requests
- RequestInfoMessage handler: @handler method to intercept sub-workflow requests
- Request correlation: Using request_id and source_executor_id to match responses with original requests
- Concurrent processing: Multiple emails processed simultaneously without interference
- External request routing: RequestInfoExecutor handles forwarded external requests
- Sub-workflow isolation: Sub-workflows work normally without knowing they're nested
@@ -48,19 +46,19 @@ Prerequisites:
Simple flow visualization:
Parent Orchestrator (@intercepts_request)
Parent Orchestrator (handles DomainCheckRequest)
|
| EmailValidationRequest(email) x3 (concurrent)
v
[ Sub-workflow: WorkflowExecutor(EmailValidator) ]
|
| DomainCheckRequest(domain) with request_id correlation
| DomainCheckRequest(domain) with request_id and source_executor_id
v
Interception? yes -> handled locally with RequestResponse.handled(True)
Interception? yes -> handled locally with RequestResponse(data=True)
no -> forwarded to RequestInfoExecutor -> external service
|
v
Response routed back to sub-workflow using request_id
Response routed back to sub-workflow using source_executor_id
"""
@@ -92,7 +90,7 @@ class ValidationResult:
class EmailValidator(Executor):
"""Validates email addresses - doesn't know it's in a sub-workflow."""
def __init__(self):
def __init__(self) -> None:
"""Initialize the EmailValidator executor."""
super().__init__(id="email_validator")
# Use a dict to track multiple pending emails by request_id
@@ -180,17 +178,28 @@ class SmartEmailOrchestrator(Executor):
request = EmailValidationRequest(email=email)
await ctx.send_message(request, target_id="email_validator_workflow")
@intercepts_request
async def check_domain(
self, request: DomainCheckRequest, ctx: WorkflowContext
) -> RequestResponse[DomainCheckRequest, bool]:
"""Intercept domain check requests from sub-workflows."""
@handler
async def handle_domain_request(
self,
request: DomainCheckRequest,
ctx: WorkflowContext[RequestResponse[DomainCheckRequest, bool] | DomainCheckRequest]
) -> None:
"""Handle requests from sub-workflows."""
print(f"🔍 Parent intercepting domain check for: {request.domain}")
if request.domain in self.approved_domains:
print(f"✅ Domain '{request.domain}' is pre-approved locally!")
return RequestResponse[DomainCheckRequest, bool].handled(True)
print(f"❓ Domain '{request.domain}' unknown, forwarding to external service...")
return RequestResponse[DomainCheckRequest, bool].forward()
# Send response back to sub-workflow
response = RequestResponse(
data=True,
original_request=request,
request_id=request.request_id
)
await ctx.send_message(response, target_id=request.source_executor_id)
else:
print(f"❓ Domain '{request.domain}' unknown, forwarding to external service...")
# Forward to external handler
await ctx.send_message(request)
@handler
async def collect_result(self, result: ValidationResult, ctx: WorkflowContext) -> None:
@@ -233,7 +242,7 @@ async def run_example() -> None:
WorkflowBuilder()
.set_start_executor(orchestrator)
.add_edge(orchestrator, workflow_executor)
.add_edge(workflow_executor, orchestrator)
.add_edge(workflow_executor, orchestrator) # For ValidationResult collection and request interception
# Add edges for external request handling
.add_edge(orchestrator, main_request_info)
.add_edge(main_request_info, workflow_executor) # Route external responses to sub-workflow