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: Adjust workflows TypeVars from prefix to suffix naming convention (#3661)

Browse Source 
* Adjust workflows TypeVars from prefix to suffix naming convention

* Adjust shared state import

* Fix MCP tool kwargs serialization bug
This commit is contained in:
Evan Mattson
2026-02-04 20:07:23 +09:00
committed by GitHub
Unverified
parent 838a7fd61d
commit 2c2800aad4
8 changed files with 47 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/packages/core/agent_framework/_workflows/__init__.py
-2
View File
@@ -102,7 +102,6 @@ from ._runner_context import (
RunnerContext,
)
from ._sequential import SequentialBuilder
from ._shared_state import SharedState
from ._validation import (
EdgeDuplicationError,
GraphConnectivityError,
@@ -179,7 +178,6 @@ __all__ = [
"Runner",
"RunnerContext",
"SequentialBuilder",
"SharedState",
"SingleEdgeGroup",
"StandardMagenticManager",
"SubWorkflowRequestMessage",
python/packages/core/agent_framework/_workflows/_base_group_chat_orchestrator.py
+7 -7
View File
@@ -57,7 +57,7 @@ class GroupChatResponseMessage:
TerminationCondition: TypeAlias = Callable[[list[ChatMessage]], bool | Awaitable[bool]]
GroupChatWorkflowContext_T_Out: TypeAlias = AgentExecutorRequest | GroupChatRequestMessage | GroupChatParticipantMessage
GroupChatWorkflowContextOutT: TypeAlias = AgentExecutorRequest | GroupChatRequestMessage | GroupChatParticipantMessage
# region Group chat events
@@ -201,7 +201,7 @@ class BaseGroupChatOrchestrator(Executor, ABC):
async def handle_str(
self,
task: str,
ctx: WorkflowContext[GroupChatWorkflowContext_T_Out, list[ChatMessage]],
ctx: WorkflowContext[GroupChatWorkflowContextOutT, list[ChatMessage]],
) -> None:
"""Handler for string input as workflow entry point.
@@ -220,7 +220,7 @@ class BaseGroupChatOrchestrator(Executor, ABC):
async def handle_message(
self,
task: ChatMessage,
ctx: WorkflowContext[GroupChatWorkflowContext_T_Out, list[ChatMessage]],
ctx: WorkflowContext[GroupChatWorkflowContextOutT, list[ChatMessage]],
) -> None:
"""Handler for single ChatMessage input as workflow entry point.
@@ -239,7 +239,7 @@ class BaseGroupChatOrchestrator(Executor, ABC):
async def handle_messages(
self,
task: list[ChatMessage],
ctx: WorkflowContext[GroupChatWorkflowContext_T_Out, list[ChatMessage]],
ctx: WorkflowContext[GroupChatWorkflowContextOutT, list[ChatMessage]],
) -> None:
"""Handler for list of ChatMessages as workflow entry point.
@@ -262,7 +262,7 @@ class BaseGroupChatOrchestrator(Executor, ABC):
async def handle_participant_response(
self,
response: AgentExecutorResponse | GroupChatResponseMessage,
ctx: WorkflowContext[GroupChatWorkflowContext_T_Out, list[ChatMessage]],
ctx: WorkflowContext[GroupChatWorkflowContextOutT, list[ChatMessage]],
) -> None:
"""Handler for participant responses.
@@ -288,7 +288,7 @@ class BaseGroupChatOrchestrator(Executor, ABC):
async def _handle_messages(
self,
messages: list[ChatMessage],
ctx: WorkflowContext[GroupChatWorkflowContext_T_Out, list[ChatMessage]],
ctx: WorkflowContext[GroupChatWorkflowContextOutT, list[ChatMessage]],
) -> None:
"""Handle task messages from users as workflow entry point.
@@ -303,7 +303,7 @@ class BaseGroupChatOrchestrator(Executor, ABC):
async def _handle_response(
self,
response: AgentExecutorResponse | GroupChatResponseMessage,
ctx: WorkflowContext[GroupChatWorkflowContext_T_Out, list[ChatMessage]],
ctx: WorkflowContext[GroupChatWorkflowContextOutT, list[ChatMessage]],
) -> None:
"""Handle a participant response.
python/packages/core/agent_framework/_workflows/_edge.py
+4 -3
View File
@@ -17,6 +17,9 @@ logger = logging.getLogger(__name__)
# Conditions receive the message data and return bool (sync or async).
EdgeCondition: TypeAlias = Callable[[Any], bool | Awaitable[bool]]
# TypeVar for EdgeGroup subclasses used in class methods
EdgeGroupT = TypeVar("EdgeGroupT", bound="EdgeGroup")
def _extract_function_name(func: Callable[..., Any]) -> str:
"""Map a Python callable to a concise, human-focused identifier.
@@ -308,8 +311,6 @@ class EdgeGroup(DictConvertible):
from builtins import type as builtin_type
_T_EdgeGroup = TypeVar("_T_EdgeGroup", bound="EdgeGroup")
_TYPE_REGISTRY: ClassVar[dict[str, builtin_type["EdgeGroup"]]] = {}
def __init__(
@@ -392,7 +393,7 @@ class EdgeGroup(DictConvertible):
}
@classmethod
def register(cls, subclass: builtin_type[_T_EdgeGroup]) -> builtin_type[_T_EdgeGroup]:
def register(cls, subclass: builtin_type[EdgeGroupT]) -> builtin_type[EdgeGroupT]:
"""Register a subclass so deserialisation can recover the right type.
Registration is typically performed via the decorator syntax applied to
python/packages/core/agent_framework/_workflows/_executor.py
+4 -4
View File
@@ -116,8 +116,8 @@ class Executor(RequestInfoMixin, DictConvertible):
async def log_message(self, msg: str, ctx: WorkflowContext) -> None:
print(f"Received: {msg}") # Only logging, no outputs
### WorkflowContext[T_Out]
Enables sending messages of type T_Out via `ctx.send_message()`:
### WorkflowContext[OutT]
Enables sending messages of type OutT via `ctx.send_message()`:
.. code-block:: python
@@ -126,8 +126,8 @@ class Executor(RequestInfoMixin, DictConvertible):
async def handler(self, msg: str, ctx: WorkflowContext[int]) -> None:
await ctx.send_message(42) # Can send int messages
### WorkflowContext[T_Out, T_W_Out]
Enables both sending messages (T_Out) and yielding workflow outputs (T_W_Out):
### WorkflowContext[OutT, W_OutT]
Enables both sending messages (OutT) and yielding workflow outputs (W_OutT):
.. code-block:: python
python/packages/core/agent_framework/_workflows/_function_executor.py
+4 -4
View File
@@ -64,11 +64,11 @@ class FunctionExecutor(Executor):
output: Optional explicit output type(s) that can be sent via ``ctx.send_message()``.
Supports union types (e.g., ``str | int``) and string forward references.
When provided, takes precedence over introspection from the ``WorkflowContext``
first generic parameter (T_Out).
first generic parameter (OutT).
workflow_output: Optional explicit output type(s) that can be yielded via
``ctx.yield_output()``. Supports union types (e.g., ``str | int``) and string
forward references. When provided, takes precedence over introspection from the
``WorkflowContext`` second generic parameter (T_W_Out).
``WorkflowContext`` second generic parameter (W_OutT).
Raises:
ValueError: If func is a staticmethod or classmethod (use @handler on instance methods instead)
@@ -262,11 +262,11 @@ def executor(
output: Optional explicit output type(s) that can be sent via ``ctx.send_message()``.
Supports union types (e.g., ``str | int``) and string forward references.
When provided, takes precedence over introspection from the ``WorkflowContext``
first generic parameter (T_Out).
first generic parameter (OutT).
workflow_output: Optional explicit output type(s) that can be yielded via
``ctx.yield_output()``. Supports union types (e.g., ``str | int``) and string
forward references. When provided, takes precedence over introspection from the
``WorkflowContext`` second generic parameter (T_W_Out).
``WorkflowContext`` second generic parameter (W_OutT).
Returns:
A FunctionExecutor instance that can be wired into a Workflow.
python/packages/core/agent_framework/_workflows/_group_chat.py
+5 -5
View File
@@ -39,7 +39,7 @@ from ._base_group_chat_orchestrator import (
GroupChatParticipantMessage,
GroupChatRequestMessage,
GroupChatResponseMessage,
GroupChatWorkflowContext_T_Out,
GroupChatWorkflowContextOutT,
ParticipantRegistry,
TerminationCondition,
)
@@ -163,7 +163,7 @@ class GroupChatOrchestrator(BaseGroupChatOrchestrator):
async def _handle_messages(
self,
messages: list[ChatMessage],
ctx: WorkflowContext[GroupChatWorkflowContext_T_Out, list[ChatMessage]],
ctx: WorkflowContext[GroupChatWorkflowContextOutT, list[ChatMessage]],
) -> None:
"""Initialize orchestrator state and start the conversation loop."""
self._append_messages(messages)
@@ -189,7 +189,7 @@ class GroupChatOrchestrator(BaseGroupChatOrchestrator):
async def _handle_response(
self,
response: AgentExecutorResponse | GroupChatResponseMessage,
ctx: WorkflowContext[GroupChatWorkflowContext_T_Out, list[ChatMessage]],
ctx: WorkflowContext[GroupChatWorkflowContextOutT, list[ChatMessage]],
) -> None:
"""Handle a participant response."""
messages = self._process_participant_response(response)
@@ -324,7 +324,7 @@ class AgentBasedGroupChatOrchestrator(BaseGroupChatOrchestrator):
async def _handle_messages(
self,
messages: list[ChatMessage],
ctx: WorkflowContext[GroupChatWorkflowContext_T_Out, list[ChatMessage]],
ctx: WorkflowContext[GroupChatWorkflowContextOutT, list[ChatMessage]],
) -> None:
"""Initialize orchestrator state and start the conversation loop."""
self._append_messages(messages)
@@ -356,7 +356,7 @@ class AgentBasedGroupChatOrchestrator(BaseGroupChatOrchestrator):
async def _handle_response(
self,
response: AgentExecutorResponse | GroupChatResponseMessage,
ctx: WorkflowContext[GroupChatWorkflowContext_T_Out, list[ChatMessage]],
ctx: WorkflowContext[GroupChatWorkflowContextOutT, list[ChatMessage]],
) -> None:
"""Handle a participant response."""
messages = self._process_participant_response(response)
python/packages/core/agent_framework/_workflows/_magentic.py
+8 -8
View File
@@ -26,7 +26,7 @@ from ._base_group_chat_orchestrator import (
GroupChatParticipantMessage,
GroupChatRequestMessage,
GroupChatResponseMessage,
GroupChatWorkflowContext_T_Out,
GroupChatWorkflowContextOutT,
ParticipantRegistry,
)
from ._checkpoint import CheckpointStorage
@@ -904,7 +904,7 @@ class MagenticOrchestrator(BaseGroupChatOrchestrator):
async def _handle_messages(
self,
messages: list[ChatMessage],
ctx: WorkflowContext[GroupChatWorkflowContext_T_Out, list[ChatMessage]],
ctx: WorkflowContext[GroupChatWorkflowContextOutT, list[ChatMessage]],
) -> None:
"""Handle the initial task messages to start the workflow."""
if self._terminated:
@@ -955,7 +955,7 @@ class MagenticOrchestrator(BaseGroupChatOrchestrator):
async def _handle_response(
self,
response: AgentExecutorResponse | GroupChatResponseMessage,
ctx: WorkflowContext[GroupChatWorkflowContext_T_Out, list[ChatMessage]],
ctx: WorkflowContext[GroupChatWorkflowContextOutT, list[ChatMessage]],
) -> None:
"""Handle a response message from a participant."""
if self._magentic_context is None or self._task_ledger is None:
@@ -981,7 +981,7 @@ class MagenticOrchestrator(BaseGroupChatOrchestrator):
self,
original_request: MagenticPlanReviewRequest,
response: MagenticPlanReviewResponse,
ctx: WorkflowContext[GroupChatWorkflowContext_T_Out, list[ChatMessage]],
ctx: WorkflowContext[GroupChatWorkflowContextOutT, list[ChatMessage]],
) -> None:
"""Handle the human response to the plan review request.
@@ -1039,7 +1039,7 @@ class MagenticOrchestrator(BaseGroupChatOrchestrator):
async def _run_inner_loop(
self,
ctx: WorkflowContext[GroupChatWorkflowContext_T_Out, list[ChatMessage]],
ctx: WorkflowContext[GroupChatWorkflowContextOutT, list[ChatMessage]],
) -> None:
"""Run the inner orchestration loop. Coordination phase. Serialized with a lock."""
if self._magentic_context is None or self._task_ledger is None:
@@ -1049,7 +1049,7 @@ class MagenticOrchestrator(BaseGroupChatOrchestrator):
async def _run_inner_loop_helper(
self,
ctx: WorkflowContext[GroupChatWorkflowContext_T_Out, list[ChatMessage]],
ctx: WorkflowContext[GroupChatWorkflowContextOutT, list[ChatMessage]],
) -> None:
"""Run inner loop with exclusive access."""
# Narrow optional context for the remainder of this method
@@ -1135,7 +1135,7 @@ class MagenticOrchestrator(BaseGroupChatOrchestrator):
async def _reset_and_replan(
self,
ctx: WorkflowContext[GroupChatWorkflowContext_T_Out, list[ChatMessage]],
ctx: WorkflowContext[GroupChatWorkflowContextOutT, list[ChatMessage]],
) -> None:
"""Reset context and replan."""
if self._magentic_context is None:
@@ -1170,7 +1170,7 @@ class MagenticOrchestrator(BaseGroupChatOrchestrator):
async def _run_outer_loop(
self,
ctx: WorkflowContext[GroupChatWorkflowContext_T_Out, list[ChatMessage]],
ctx: WorkflowContext[GroupChatWorkflowContextOutT, list[ChatMessage]],
) -> None:
"""Run the outer orchestration loop - planning phase."""
if self._magentic_context is None:
python/packages/core/agent_framework/_workflows/_workflow_context.py
+15 -15
View File
@@ -31,8 +31,8 @@ from ._shared_state import SharedState
if TYPE_CHECKING:
from ._executor import Executor
T_Out = TypeVar("T_Out", default=Never)
T_W_Out = TypeVar("T_W_Out", default=Never)
OutT = TypeVar("OutT", default=Never)
W_OutT = TypeVar("W_OutT", default=Never)
logger = logging.getLogger(__name__)
@@ -67,7 +67,7 @@ def infer_output_types_from_ctx_annotation(
if origin is None:
return [], []
# Expecting WorkflowContext[T_Out, T_W_Out]
# Expecting WorkflowContext[OutT, W_OutT]
if origin is not WorkflowContext:
return [], []
@@ -75,7 +75,7 @@ def infer_output_types_from_ctx_annotation(
if not args:
return [], []
# WorkflowContext[T_Out] -> message_types from T_Out, no workflow output types
# WorkflowContext[OutT] -> message_types from OutT, no workflow output types
if len(args) == 1:
t = args[0]
t_origin = get_origin(t)
@@ -90,10 +90,10 @@ def infer_output_types_from_ctx_annotation(
return [], []
return [t], []
# WorkflowContext[T_Out, T_W_Out] -> message_types from T_Out, workflow_output_types from T_W_Out
# WorkflowContext[OutT, W_OutT] -> message_types from OutT, workflow_output_types from W_OutT
t_out, t_w_out = args[:2] # Take first two args in case there are more
# Process T_Out for message_types
# Process OutT for message_types
message_types: list[type[Any] | UnionType] = []
t_out_origin = get_origin(t_out)
if t_out is Any:
@@ -104,7 +104,7 @@ def infer_output_types_from_ctx_annotation(
else:
message_types = [t_out]
# Process T_W_Out for workflow_output_types
# Process W_OutT for workflow_output_types
workflow_output_types: list[type[Any] | UnionType] = []
t_w_out_origin = get_origin(t_w_out)
if t_w_out is Any:
@@ -176,7 +176,7 @@ def validate_workflow_context_annotation(
return isinstance(x, type) or get_origin(x) is not None or x is Never
for i, type_arg in enumerate(type_args):
param_description = "T_Out" if i == 0 else "T_W_Out"
param_description = "OutT" if i == 0 else "W_OutT"
# Allow Any explicitly
if type_arg is Any:
@@ -216,7 +216,7 @@ _FRAMEWORK_LIFECYCLE_EVENT_TYPES: tuple[type[WorkflowEvent], ...] = cast(
)
class WorkflowContext(Generic[T_Out, T_W_Out]):
class WorkflowContext(Generic[OutT, W_OutT]):
"""Execution context that enables executors to interact with workflows and other executors.
## Overview
@@ -235,8 +235,8 @@ class WorkflowContext(Generic[T_Out, T_W_Out]):
async def log_handler(message: str, ctx: WorkflowContext) -> None:
print(f"Received: {message}") # Only side effects
### WorkflowContext[T_Out]
Enables sending messages of type T_Out to other executors:
### WorkflowContext[OutT]
Enables sending messages of type OutT to other executors:
.. code-block:: python
@@ -244,8 +244,8 @@ class WorkflowContext(Generic[T_Out, T_W_Out]):
result = len(message)
await ctx.send_message(result) # Send int to downstream executors
### WorkflowContext[T_Out, T_W_Out]
Enables both sending messages (T_Out) and yielding workflow outputs (T_W_Out):
### WorkflowContext[OutT, W_OutT]
Enables both sending messages (OutT) and yielding workflow outputs (W_OutT):
.. code-block:: python
@@ -317,7 +317,7 @@ class WorkflowContext(Generic[T_Out, T_W_Out]):
"""
return self._request_id
async def send_message(self, message: T_Out, target_id: str | None = None) -> None:
async def send_message(self, message: OutT, target_id: str | None = None) -> None:
"""Send a message to the workflow context.
Args:
@@ -349,7 +349,7 @@ class WorkflowContext(Generic[T_Out, T_W_Out]):
await self._runner_context.send_message(msg)
async def yield_output(self, output: T_W_Out) -> None:
async def yield_output(self, output: W_OutT) -> None:
"""Set the output of the workflow.
Args: