From 6a491fb3896151bceddebcbd3a395c967d8f03c6 Mon Sep 17 00:00:00 2001 From: droideronline Date: Mon, 31 Aug 2026 16:43:07 +0530 Subject: [PATCH 1/5] Python: support mixed workflow invocation kwargs --- .../_workflows/_agent_executor.py | 26 ++++++++++-------- .../agent_framework/_workflows/_workflow.py | 17 +++++++++--- .../_workflows/_workflow_executor.py | 15 +++++------ .../tests/workflow/test_agent_executor.py | 15 +++++------ .../tests/workflow/test_workflow_kwargs.py | 27 +++++++++++++++++++ 5 files changed, 69 insertions(+), 31 deletions(-) diff --git a/python/packages/core/agent_framework/_workflows/_agent_executor.py b/python/packages/core/agent_framework/_workflows/_agent_executor.py index b7787736fa5..4c07816a158 100644 --- a/python/packages/core/agent_framework/_workflows/_agent_executor.py +++ b/python/packages/core/agent_framework/_workflows/_agent_executor.py @@ -581,22 +581,26 @@ def _resolve_executor_kwargs(self, resolved: dict[str, Any] | None) -> dict[str, """ if not isinstance(resolved, dict): return None - # Use explicit key-presence checks so that an empty per-executor dict is - # honoured (e.g. to clear kwargs) instead of falling through to global. - if self.id in resolved: - executor_kwargs = resolved[self.id] - elif GLOBAL_KWARGS_KEY in resolved: - executor_kwargs = resolved[GLOBAL_KWARGS_KEY] - else: + global_kwargs = resolved.get(GLOBAL_KWARGS_KEY) + executor_kwargs = resolved.get(self.id) + if global_kwargs is None and executor_kwargs is None: return None - if not isinstance(executor_kwargs, dict): + if global_kwargs is not None and not isinstance(global_kwargs, dict): logger.warning( - "Executor %s expected a dict for its kwargs, but got %s. Ignoring.", + "Executor %s expected a dict for global kwargs, but got %s. Ignoring.", self.id, - type(executor_kwargs), # type: ignore + type(global_kwargs), ) + return None + if executor_kwargs is not None and not isinstance(executor_kwargs, dict): + logger.warning( + "Executor %s expected a dict for its kwargs, but got %s. Ignoring.", + self.id, + type(executor_kwargs), + ) return None - return executor_kwargs # type: ignore + # Specific values override global values for the same function argument. + return {**(global_kwargs or {}), **(executor_kwargs or {})} diff --git a/python/packages/core/agent_framework/_workflows/_workflow.py b/python/packages/core/agent_framework/_workflows/_workflow.py index 04f3f9aa875..e0f5e703f59 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow.py +++ b/python/packages/core/agent_framework/_workflows/_workflow.py @@ -740,7 +740,8 @@ def run( include_status_events: Whether to include status events (non-streaming only). function_invocation_kwargs: Keyword arguments forwarded to tool invocations in subagents. Either a mapping for agent name or agent executor id to kwargs, - or a flat mapping of kwargs for all tool invocations. + or a flat mapping of kwargs for all tool invocations. To combine global and + executor-specific kwargs, use the ``"__global__"`` key for the global mapping. client_kwargs: Keyword arguments forwarded to chat client calls in subagents. Either a mapping for agent name or agent executor id to kwargs, or a flat mapping of kwargs for all chat client calls. @@ -1065,7 +1066,8 @@ def _resolve_invocation_kwargs( Detects whether the provided kwargs dict uses per-executor targeting by checking if any top-level key matches a known executor ID in the workflow. If at least one key matches, all entries are treated as per-executor. Otherwise the dict is treated - as global kwargs that apply to every executor. + as global kwargs that apply to every executor. The ``"__global__"`` key can be used + explicitly to combine global kwargs with per-executor overrides. Args: kwargs: The raw invocation kwargs from the caller. @@ -1074,8 +1076,17 @@ def _resolve_invocation_kwargs( Returns: A dict with either: - ``{"__global__": }`` for global kwargs, or - - The original dict unchanged for per-executor kwargs. + - A mapping containing ``"__global__"`` and per-executor kwargs. """ + if GLOBAL_KWARGS_KEY in kwargs: + global_kwargs = kwargs[GLOBAL_KWARGS_KEY] + if not isinstance(global_kwargs, Mapping): + raise ValueError(f"{GLOBAL_KWARGS_KEY} must contain a mapping of global kwargs.") + resolved = dict(kwargs) + resolved[GLOBAL_KWARGS_KEY] = dict(global_kwargs) + logger.info("Explicit global %s provided; applying it with any per-executor overrides.", param_name) + return resolved + executor_ids = set(self.executors.keys()) matched_ids = kwargs.keys() & executor_ids if matched_ids: diff --git a/python/packages/core/agent_framework/_workflows/_workflow_executor.py b/python/packages/core/agent_framework/_workflows/_workflow_executor.py index 1a8f988d19b..901aaf2cf28 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow_executor.py +++ b/python/packages/core/agent_framework/_workflows/_workflow_executor.py @@ -10,7 +10,7 @@ if TYPE_CHECKING: from ._workflow import Workflow -from ._const import GLOBAL_KWARGS_KEY, WORKFLOW_RUN_KWARGS_KEY +from ._const import WORKFLOW_RUN_KWARGS_KEY from ._events import ( WorkflowEvent, WorkflowRunState, @@ -375,21 +375,18 @@ async def process_workflow(self, input_data: object, ctx: WorkflowContext[Any, A # Get kwargs from parent workflow's State to propagate to subworkflow parent_kwargs: dict[str, Any] = ctx.get_state(WORKFLOW_RUN_KWARGS_KEY, {}) - # Extract invocation kwargs recognised by Workflow.run() - # The state stores resolved format (with __global__ wrapper for global kwargs). - # Unwrap __global__ before passing to the subworkflow so it gets re-resolved - # against the subworkflow's own executor IDs. + # Extract invocation kwargs recognised by Workflow.run(). The state stores + # the resolved format, which can include a global mapping and executor overrides. + # Pass it through so the subworkflow resolves it against its own executor IDs. fi_kwargs: dict[str, Any] | None = None ci_kwargs: dict[str, Any] | None = None for key in ("function_invocation_kwargs", "client_kwargs"): resolved = parent_kwargs.get(key) if isinstance(resolved, dict): - # Unwrap global sentinel; pass per-executor dicts as-is - unwrapped: dict[str, Any] = resolved.get(GLOBAL_KWARGS_KEY, resolved) # type: ignore if key == "function_invocation_kwargs": - fi_kwargs = unwrapped # type: ignore + fi_kwargs = resolved else: - ci_kwargs = unwrapped # type: ignore + ci_kwargs = resolved # Run the sub-workflow and collect all events, passing parent kwargs result = await self.workflow.run( diff --git a/python/packages/core/tests/workflow/test_agent_executor.py b/python/packages/core/tests/workflow/test_agent_executor.py index ccb1e9425bf..9e124db1d9c 100644 --- a/python/packages/core/tests/workflow/test_agent_executor.py +++ b/python/packages/core/tests/workflow/test_agent_executor.py @@ -622,15 +622,15 @@ async def test_resolve_executor_kwargs_returns_none_for_none_input() -> None: assert result is None -async def test_resolve_executor_kwargs_prefers_executor_id_over_global() -> None: - """_resolve_executor_kwargs prefers executor-specific entry over __global__.""" +async def test_resolve_executor_kwargs_merges_executor_id_over_global() -> None: + """_resolve_executor_kwargs merges executor-specific entries over __global__.""" agent = _CountingAgent(id="a", name="A") executor = AgentExecutor(agent, id="exec_a") # Dict has both a per-executor entry and a global entry resolved = {"exec_a": {"specific": True}, GLOBAL_KWARGS_KEY: {"global": True}} result = executor._resolve_executor_kwargs(resolved) # pyright: ignore[reportPrivateUsage] - assert result == {"specific": True} + assert result == {"global": True, "specific": True} async def test_prepare_agent_run_args_extracts_function_invocation_kwargs() -> None: @@ -689,16 +689,15 @@ async def test_prepare_agent_run_args_per_executor_no_match() -> None: assert fi_kwargs is None -async def test_resolve_executor_kwargs_empty_per_executor_does_not_fallback_to_global() -> None: - """An explicit empty per-executor dict should not fall through to global kwargs.""" +async def test_resolve_executor_kwargs_empty_per_executor_keeps_global_kwargs() -> None: + """An explicit empty per-executor dict keeps the global kwargs.""" agent = _CountingAgent(id="a", name="A") executor = AgentExecutor(agent, id="exec_a") - # Per-executor entry for exec_a is empty, but global has values. - # The empty dict should be honoured (no fallback to global). + # Per-executor entry for exec_a is empty, so only global values apply. resolved = {"exec_a": {}, GLOBAL_KWARGS_KEY: {"global_key": "global_val"}} # type: ignore[var-annotated] result = executor._resolve_executor_kwargs(resolved) # pyright: ignore[reportPrivateUsage] - assert result == {} + assert result == {"global_key": "global_val"} # region Tool approval emission diff --git a/python/packages/core/tests/workflow/test_workflow_kwargs.py b/python/packages/core/tests/workflow/test_workflow_kwargs.py index 93c6c93d580..4f2ebcd7eb5 100644 --- a/python/packages/core/tests/workflow/test_workflow_kwargs.py +++ b/python/packages/core/tests/workflow/test_workflow_kwargs.py @@ -870,6 +870,33 @@ async def test_per_executor_function_invocation_kwargs_routes_to_correct_agent() assert agent2.captured_kwargs[0].get("function_invocation_kwargs") == {"tool_param": "value_for_agent2"} +async def test_global_and_per_executor_function_invocation_kwargs_are_merged() -> None: + """Global function kwargs are merged with executor-specific overrides.""" + agent1 = _KwargsCapturingAgent(name="agent1") + agent2 = _KwargsCapturingAgent(name="agent2") + workflow = SequentialBuilder(participants=[agent1, agent2]).build() + + fi_kwargs = { + "__global__": {"shared": "value", "overridden": "global"}, + "agent1": {"overridden": "agent1"}, + "agent2": {"agent_only": True}, + } + + async for event in workflow.run("test", stream=True, function_invocation_kwargs=fi_kwargs): + if event.type == "status" and event.state == WorkflowRunState.IDLE: + break + + assert agent1.captured_kwargs[0].get("function_invocation_kwargs") == { + "shared": "value", + "overridden": "agent1", + } + assert agent2.captured_kwargs[0].get("function_invocation_kwargs") == { + "shared": "value", + "overridden": "global", + "agent_only": True, + } + + async def test_per_executor_kwargs_unmatched_agent_gets_none() -> None: """An agent not targeted in per-executor kwargs should receive None for that kwarg.""" agent1 = _KwargsCapturingAgent(name="agent1") From cda37e485e72a4bc3ed8cc0781b2adaeaef900c0 Mon Sep 17 00:00:00 2001 From: droideronline Date: Tue, 1 Sep 2026 11:54:39 +0530 Subject: [PATCH 2/5] Python: preserve workflow kwargs compatibility --- .../packages/core/agent_framework/__init__.py | 3 +- .../core/agent_framework/__init__.pyi | 3 +- .../core/agent_framework/_workflows/_agent.py | 44 +++++++--- .../core/agent_framework/_workflows/_const.py | 4 + .../agent_framework/_workflows/_workflow.py | 80 +++++++++++++------ .../_workflows/_workflow_executor.py | 25 +++--- .../tests/workflow/test_workflow_kwargs.py | 71 ++++++++++++++-- 7 files changed, 176 insertions(+), 54 deletions(-) diff --git a/python/packages/core/agent_framework/__init__.py b/python/packages/core/agent_framework/__init__.py index 77c56fa7559..1a06e91b728 100644 --- a/python/packages/core/agent_framework/__init__.py +++ b/python/packages/core/agent_framework/__init__.py @@ -343,7 +343,7 @@ "validate_workflow_graph", ), "._workflows._viz": ("WorkflowViz",), - "._workflows._workflow": ("Workflow", "WorkflowRunResult"), + "._workflows._workflow": ("Workflow", "WorkflowInvocationKwargs", "WorkflowRunResult"), "._workflows._workflow_builder": ("WorkflowBuilder",), "._workflows._workflow_context": ("WorkflowContext",), "._workflows._workflow_executor": ( @@ -593,6 +593,7 @@ "WorkflowEventType", "WorkflowException", "WorkflowExecutor", + "WorkflowInvocationKwargs", "WorkflowMessage", "WorkflowRunResult", "WorkflowRunState", diff --git a/python/packages/core/agent_framework/__init__.pyi b/python/packages/core/agent_framework/__init__.pyi index fa8f6a75ae6..820908b2fba 100644 --- a/python/packages/core/agent_framework/__init__.pyi +++ b/python/packages/core/agent_framework/__init__.pyi @@ -306,7 +306,7 @@ from ._workflows._validation import ( validate_workflow_graph, ) from ._workflows._viz import WorkflowViz -from ._workflows._workflow import Workflow, WorkflowRunResult +from ._workflows._workflow import Workflow, WorkflowInvocationKwargs, WorkflowRunResult from ._workflows._workflow_builder import WorkflowBuilder from ._workflows._workflow_context import WorkflowContext from ._workflows._workflow_executor import SubWorkflowRequestMessage, SubWorkflowResponseMessage, WorkflowExecutor @@ -559,6 +559,7 @@ __all__ = [ "WorkflowExecutor", "WorkflowMessage", "WorkflowRunResult", + "WorkflowInvocationKwargs", "WorkflowRunState", "WorkflowRunnerException", "WorkflowValidationError", diff --git a/python/packages/core/agent_framework/_workflows/_agent.py b/python/packages/core/agent_framework/_workflows/_agent.py index 353a1a2efbc..cab8df615ef 100644 --- a/python/packages/core/agent_framework/_workflows/_agent.py +++ b/python/packages/core/agent_framework/_workflows/_agent.py @@ -44,7 +44,7 @@ from typing_extensions import TypedDict # pragma: no cover if TYPE_CHECKING: - from ._workflow import Workflow + from ._workflow import Workflow, WorkflowInvocationKwargs logger = logging.getLogger(__name__) @@ -155,8 +155,11 @@ def run( session: AgentSession | None = None, checkpoint_id: str | None = None, checkpoint_storage: CheckpointStorage | None = None, - function_invocation_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, - client_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, + function_invocation_kwargs: WorkflowInvocationKwargs + | Mapping[str, Mapping[str, Any]] + | Mapping[str, Any] + | None = None, + client_kwargs: WorkflowInvocationKwargs | Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, ) -> ResponseStream[AgentResponseUpdate, AgentResponse]: ... @overload @@ -168,8 +171,11 @@ async def run( session: AgentSession | None = None, checkpoint_id: str | None = None, checkpoint_storage: CheckpointStorage | None = None, - function_invocation_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, - client_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, + function_invocation_kwargs: WorkflowInvocationKwargs + | Mapping[str, Mapping[str, Any]] + | Mapping[str, Any] + | None = None, + client_kwargs: WorkflowInvocationKwargs | Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, ) -> AgentResponse: ... def run( @@ -180,8 +186,11 @@ def run( session: AgentSession | None = None, checkpoint_id: str | None = None, checkpoint_storage: CheckpointStorage | None = None, - function_invocation_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, - client_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, + function_invocation_kwargs: WorkflowInvocationKwargs + | Mapping[str, Mapping[str, Any]] + | Mapping[str, Any] + | None = None, + client_kwargs: WorkflowInvocationKwargs | Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, ) -> ResponseStream[AgentResponseUpdate, AgentResponse] | Awaitable[AgentResponse]: """Get a response from the workflow agent. @@ -246,8 +255,11 @@ async def _run_impl( session: AgentSession | None, checkpoint_id: str | None = None, checkpoint_storage: CheckpointStorage | None = None, - function_invocation_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, - client_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, + function_invocation_kwargs: WorkflowInvocationKwargs + | Mapping[str, Mapping[str, Any]] + | Mapping[str, Any] + | None = None, + client_kwargs: WorkflowInvocationKwargs | Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, ) -> AgentResponse: """Internal implementation of non-streaming execution. @@ -326,8 +338,11 @@ async def _run_stream_impl( session: AgentSession | None, checkpoint_id: str | None = None, checkpoint_storage: CheckpointStorage | None = None, - function_invocation_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, - client_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, + function_invocation_kwargs: WorkflowInvocationKwargs + | Mapping[str, Mapping[str, Any]] + | Mapping[str, Any] + | None = None, + client_kwargs: WorkflowInvocationKwargs | Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, ) -> AsyncIterable[AgentResponseUpdate]: """Internal implementation of streaming execution. @@ -405,8 +420,11 @@ async def _run_core( checkpoint_id: str | None, checkpoint_storage: CheckpointStorage | None, streaming: bool, - function_invocation_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, - client_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, + function_invocation_kwargs: WorkflowInvocationKwargs + | Mapping[str, Mapping[str, Any]] + | Mapping[str, Any] + | None = None, + client_kwargs: WorkflowInvocationKwargs | Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, ) -> AsyncIterable[WorkflowEvent]: """Core implementation that yields workflow events for both streaming and non-streaming modes. diff --git a/python/packages/core/agent_framework/_workflows/_const.py b/python/packages/core/agent_framework/_workflows/_const.py index e83025bbdc7..a84881196a5 100644 --- a/python/packages/core/agent_framework/_workflows/_const.py +++ b/python/packages/core/agent_framework/_workflows/_const.py @@ -14,6 +14,10 @@ # to pass kwargs from workflow.run() through to agent.run() and @tool functions. WORKFLOW_RUN_KWARGS_KEY = "_workflow_run_kwargs" +# State keys used to preserve caller-provided kwargs for nested workflow routing. +RAW_FUNCTION_INVOCATION_KWARGS_KEY = "_raw_function_invocation_kwargs" +RAW_CLIENT_KWARGS_KEY = "_raw_client_kwargs" + # Sentinel key used in resolved invocation kwargs dicts to denote global kwargs # that apply to all executors (as opposed to per-executor keyed entries). GLOBAL_KWARGS_KEY = "__global__" diff --git a/python/packages/core/agent_framework/_workflows/_workflow.py b/python/packages/core/agent_framework/_workflows/_workflow.py index e0f5e703f59..3178a976b49 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow.py +++ b/python/packages/core/agent_framework/_workflows/_workflow.py @@ -21,7 +21,14 @@ from ..exceptions import WorkflowException from ..observability import OtelAttr, capture_exception, create_workflow_span from ._checkpoint import CheckpointStorage -from ._const import DEFAULT_MAX_ITERATIONS, GLOBAL_KWARGS_KEY, INTERNAL_SOURCE_ID, WORKFLOW_RUN_KWARGS_KEY +from ._const import ( + DEFAULT_MAX_ITERATIONS, + GLOBAL_KWARGS_KEY, + INTERNAL_SOURCE_ID, + RAW_CLIENT_KWARGS_KEY, + RAW_FUNCTION_INVOCATION_KWARGS_KEY, + WORKFLOW_RUN_KWARGS_KEY, +) from ._edge import ( EdgeGroup, FanOutEdgeGroup, @@ -205,6 +212,18 @@ def classify(self, executor_id: str) -> Literal["output", "intermediate"] | None return None +@dataclass(frozen=True) +class WorkflowInvocationKwargs: + """Explicit global and executor-specific kwargs for a workflow run. + + Use this wrapper when shared kwargs should be combined with executor-specific + overrides. Plain mappings retain their existing global or per-executor behavior. + """ + + global_kwargs: Mapping[str, Any] = field(default_factory=dict) + executor_kwargs: Mapping[str, Mapping[str, Any]] = field(default_factory=dict) + + class Workflow(DictConvertible): """A graph-based execution engine that orchestrates connected executors. @@ -480,8 +499,11 @@ async def _run_workflow_with_tracing( initial_executor_fn: Callable[[], Awaitable[None]] | None = None, is_continuation: bool = False, streaming: bool = False, - function_invocation_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, - client_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, + function_invocation_kwargs: WorkflowInvocationKwargs + | Mapping[str, Mapping[str, Any]] + | Mapping[str, Any] + | None = None, + client_kwargs: WorkflowInvocationKwargs | Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, ) -> AsyncIterable[WorkflowEvent]: """Private method to run workflow with proper tracing. @@ -556,10 +578,12 @@ async def _run_workflow_with_tracing( combined_kwargs["function_invocation_kwargs"] = self._resolve_invocation_kwargs( function_invocation_kwargs, "function_invocation_kwargs" ) + combined_kwargs[RAW_FUNCTION_INVOCATION_KWARGS_KEY] = function_invocation_kwargs if client_kwargs is not None: combined_kwargs["client_kwargs"] = self._resolve_invocation_kwargs( client_kwargs, "client_kwargs" ) + combined_kwargs[RAW_CLIENT_KWARGS_KEY] = client_kwargs self._runner.state.set(WORKFLOW_RUN_KWARGS_KEY, combined_kwargs) elif not is_continuation: self._runner.state.set(WORKFLOW_RUN_KWARGS_KEY, {}) @@ -688,8 +712,8 @@ def run( responses: Mapping[str, Any] | None = None, checkpoint_id: str | None = None, checkpoint_storage: CheckpointStorage | None = None, - function_invocation_kwargs: Mapping[str, Any] | None = None, - client_kwargs: Mapping[str, Any] | None = None, + function_invocation_kwargs: WorkflowInvocationKwargs | Mapping[str, Any] | None = None, + client_kwargs: WorkflowInvocationKwargs | Mapping[str, Any] | None = None, ) -> ResponseStream[WorkflowEvent, WorkflowRunResult]: ... @overload @@ -702,8 +726,8 @@ def run( checkpoint_id: str | None = None, checkpoint_storage: CheckpointStorage | None = None, include_status_events: bool = False, - function_invocation_kwargs: Mapping[str, Any] | None = None, - client_kwargs: Mapping[str, Any] | None = None, + function_invocation_kwargs: WorkflowInvocationKwargs | Mapping[str, Any] | None = None, + client_kwargs: WorkflowInvocationKwargs | Mapping[str, Any] | None = None, ) -> Awaitable[WorkflowRunResult]: ... def run( @@ -715,8 +739,11 @@ def run( checkpoint_id: str | None = None, checkpoint_storage: CheckpointStorage | None = None, include_status_events: bool = False, - function_invocation_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, - client_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, + function_invocation_kwargs: WorkflowInvocationKwargs + | Mapping[str, Mapping[str, Any]] + | Mapping[str, Any] + | None = None, + client_kwargs: WorkflowInvocationKwargs | Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, ) -> ResponseStream[WorkflowEvent, WorkflowRunResult] | Awaitable[WorkflowRunResult]: """Run the workflow, optionally streaming events. @@ -740,11 +767,14 @@ def run( include_status_events: Whether to include status events (non-streaming only). function_invocation_kwargs: Keyword arguments forwarded to tool invocations in subagents. Either a mapping for agent name or agent executor id to kwargs, - or a flat mapping of kwargs for all tool invocations. To combine global and - executor-specific kwargs, use the ``"__global__"`` key for the global mapping. + a flat mapping of kwargs for all tool invocations, or a + ``WorkflowInvocationKwargs`` instance to combine global and executor-specific + kwargs. client_kwargs: Keyword arguments forwarded to chat client calls in subagents. Either a mapping for agent name or agent executor id to kwargs, - or a flat mapping of kwargs for all chat client calls. + a flat mapping of kwargs for all chat client calls, or a + ``WorkflowInvocationKwargs`` instance to combine global and executor-specific + kwargs. Returns: When stream=True: A ResponseStream[WorkflowEvent, WorkflowRunResult] for @@ -803,8 +833,11 @@ async def _run_core( checkpoint_id: str | None = None, checkpoint_storage: CheckpointStorage | None = None, streaming: bool = False, - function_invocation_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, - client_kwargs: Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, + function_invocation_kwargs: WorkflowInvocationKwargs + | Mapping[str, Mapping[str, Any]] + | Mapping[str, Any] + | None = None, + client_kwargs: WorkflowInvocationKwargs | Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, ) -> AsyncIterable[WorkflowEvent]: """Single core execution path for both streaming and non-streaming modes. @@ -1058,7 +1091,7 @@ def _get_executor_by_id(self, executor_id: str) -> Executor: def _resolve_invocation_kwargs( self, - kwargs: Mapping[str, Any], + kwargs: WorkflowInvocationKwargs | Mapping[str, Any], param_name: str, ) -> dict[str, Any]: """Resolve invocation kwargs into a normalized per-executor or global format. @@ -1074,17 +1107,14 @@ def _resolve_invocation_kwargs( param_name: The parameter name (for logging), e.g. ``"function_invocation_kwargs"``. Returns: - A dict with either: - - ``{"__global__": }`` for global kwargs, or - - A mapping containing ``"__global__"`` and per-executor kwargs. + A dict containing normalized global or per-executor mappings. """ - if GLOBAL_KWARGS_KEY in kwargs: - global_kwargs = kwargs[GLOBAL_KWARGS_KEY] - if not isinstance(global_kwargs, Mapping): - raise ValueError(f"{GLOBAL_KWARGS_KEY} must contain a mapping of global kwargs.") - resolved = dict(kwargs) - resolved[GLOBAL_KWARGS_KEY] = dict(global_kwargs) - logger.info("Explicit global %s provided; applying it with any per-executor overrides.", param_name) + if isinstance(kwargs, WorkflowInvocationKwargs): + resolved = {GLOBAL_KWARGS_KEY: dict(kwargs.global_kwargs)} + resolved.update({ + executor_id: dict(executor_kwargs) for executor_id, executor_kwargs in kwargs.executor_kwargs.items() + }) + logger.info("Explicit global %s provided with executor-specific overrides.", param_name) return resolved executor_ids = set(self.executors.keys()) diff --git a/python/packages/core/agent_framework/_workflows/_workflow_executor.py b/python/packages/core/agent_framework/_workflows/_workflow_executor.py index 901aaf2cf28..97611024558 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow_executor.py +++ b/python/packages/core/agent_framework/_workflows/_workflow_executor.py @@ -4,13 +4,18 @@ import logging import sys import types +from collections.abc import Mapping from dataclasses import dataclass from typing import TYPE_CHECKING, Any if TYPE_CHECKING: from ._workflow import Workflow -from ._const import WORKFLOW_RUN_KWARGS_KEY +from ._const import ( + RAW_CLIENT_KWARGS_KEY, + RAW_FUNCTION_INVOCATION_KWARGS_KEY, + WORKFLOW_RUN_KWARGS_KEY, +) from ._events import ( WorkflowEvent, WorkflowRunState, @@ -20,7 +25,7 @@ from ._request_info_mixin import response_handler from ._runner_context import WorkflowMessage from ._typing_utils import is_instance_of -from ._workflow import WorkflowRunResult +from ._workflow import WorkflowInvocationKwargs, WorkflowRunResult from ._workflow_context import WorkflowContext if sys.version_info >= (3, 12): @@ -375,14 +380,16 @@ async def process_workflow(self, input_data: object, ctx: WorkflowContext[Any, A # Get kwargs from parent workflow's State to propagate to subworkflow parent_kwargs: dict[str, Any] = ctx.get_state(WORKFLOW_RUN_KWARGS_KEY, {}) - # Extract invocation kwargs recognised by Workflow.run(). The state stores - # the resolved format, which can include a global mapping and executor overrides. - # Pass it through so the subworkflow resolves it against its own executor IDs. - fi_kwargs: dict[str, Any] | None = None - ci_kwargs: dict[str, Any] | None = None + # Use the caller's raw kwargs so legacy per-executor mappings are resolved + # against the child workflow's executor IDs rather than the parent's. + fi_kwargs: WorkflowInvocationKwargs | Mapping[str, Any] | None = None + ci_kwargs: WorkflowInvocationKwargs | Mapping[str, Any] | None = None for key in ("function_invocation_kwargs", "client_kwargs"): - resolved = parent_kwargs.get(key) - if isinstance(resolved, dict): + raw_key = ( + RAW_FUNCTION_INVOCATION_KWARGS_KEY if key == "function_invocation_kwargs" else RAW_CLIENT_KWARGS_KEY + ) + resolved = parent_kwargs.get(raw_key, parent_kwargs.get(key)) + if isinstance(resolved, dict) or resolved is not None: if key == "function_invocation_kwargs": fi_kwargs = resolved else: diff --git a/python/packages/core/tests/workflow/test_workflow_kwargs.py b/python/packages/core/tests/workflow/test_workflow_kwargs.py index 4f2ebcd7eb5..652f9977457 100644 --- a/python/packages/core/tests/workflow/test_workflow_kwargs.py +++ b/python/packages/core/tests/workflow/test_workflow_kwargs.py @@ -16,6 +16,7 @@ Content, Message, ResponseStream, + WorkflowInvocationKwargs, WorkflowRunState, ) from agent_framework._workflows._const import WORKFLOW_RUN_KWARGS_KEY @@ -789,6 +790,35 @@ async def test_nested_subworkflow_kwargs_propagation() -> None: ) +async def test_mixed_kwargs_route_through_subworkflow() -> None: + """Mixed kwargs preserve global values and child executor-specific routing.""" + from agent_framework._workflows._workflow_executor import WorkflowExecutor + + inner_agent1 = _KwargsCapturingAgent(name="inner_agent1") + inner_agent2 = _KwargsCapturingAgent(name="inner_agent2") + inner_workflow = SequentialBuilder(participants=[inner_agent1, inner_agent2]).build() + subworkflow_executor = WorkflowExecutor(workflow=inner_workflow, id="subworkflow") + outer_workflow = SequentialBuilder(participants=[subworkflow_executor]).build() + + fi_kwargs = WorkflowInvocationKwargs( + global_kwargs={"shared": "value", "overridden": "global"}, + executor_kwargs={"inner_agent2": {"overridden": "inner_agent2"}}, + ) + + async for event in outer_workflow.run("test", stream=True, function_invocation_kwargs=fi_kwargs): + if event.type == "status" and event.state == WorkflowRunState.IDLE: + break + + assert inner_agent1.captured_kwargs[0].get("function_invocation_kwargs") == { + "shared": "value", + "overridden": "global", + } + assert inner_agent2.captured_kwargs[0].get("function_invocation_kwargs") == { + "shared": "value", + "overridden": "inner_agent2", + } + + # endregion @@ -876,11 +906,13 @@ async def test_global_and_per_executor_function_invocation_kwargs_are_merged() - agent2 = _KwargsCapturingAgent(name="agent2") workflow = SequentialBuilder(participants=[agent1, agent2]).build() - fi_kwargs = { - "__global__": {"shared": "value", "overridden": "global"}, - "agent1": {"overridden": "agent1"}, - "agent2": {"agent_only": True}, - } + fi_kwargs = WorkflowInvocationKwargs( + global_kwargs={"shared": "value", "overridden": "global"}, + executor_kwargs={ + "agent1": {"overridden": "agent1"}, + "agent2": {"agent_only": True}, + }, + ) async for event in workflow.run("test", stream=True, function_invocation_kwargs=fi_kwargs): if event.type == "status" and event.state == WorkflowRunState.IDLE: @@ -967,6 +999,35 @@ async def test_per_executor_client_kwargs_routes_correctly() -> None: assert agent2.captured_kwargs[0].get("client_kwargs") == {"temperature": 0.9} +async def test_global_and_per_executor_client_kwargs_are_merged() -> None: + """Global client kwargs are merged with executor-specific overrides.""" + agent1 = _KwargsCapturingAgent(name="agent1") + agent2 = _KwargsCapturingAgent(name="agent2") + workflow = SequentialBuilder(participants=[agent1, agent2]).build() + + ci_kwargs = WorkflowInvocationKwargs( + global_kwargs={"shared": "value", "overridden": "global"}, + executor_kwargs={ + "agent1": {"overridden": "agent1"}, + "agent2": {"agent_only": True}, + }, + ) + + async for event in workflow.run("test", stream=True, client_kwargs=ci_kwargs): + if event.type == "status" and event.state == WorkflowRunState.IDLE: + break + + assert agent1.captured_kwargs[0].get("client_kwargs") == { + "shared": "value", + "overridden": "agent1", + } + assert agent2.captured_kwargs[0].get("client_kwargs") == { + "shared": "value", + "overridden": "global", + "agent_only": True, + } + + async def test_resolve_invocation_kwargs_logs_per_executor(caplog: "LogCaptureFixture") -> None: """Workflow._resolve_invocation_kwargs logs info when per-executor format is detected.""" import logging From 9921c5bb26eeb24c725ea67750377fa14d2e68f7 Mon Sep 17 00:00:00 2001 From: droideronline Date: Fri, 4 Sep 2026 13:18:44 +0530 Subject: [PATCH 3/5] Python: fix workflow kwargs type checking --- .../_workflows/_agent_executor.py | 8 +-- .../agent_framework/_workflows/_workflow.py | 6 +- .../_workflows/_workflow_executor.py | 13 +++-- .../tests/core/test_serializable_mixin.py | 55 +++++++++++++++++++ .../tests/workflow/test_agent_executor.py | 1 + 5 files changed, 72 insertions(+), 11 deletions(-) diff --git a/python/packages/core/agent_framework/_workflows/_agent_executor.py b/python/packages/core/agent_framework/_workflows/_agent_executor.py index 4c07816a158..f79ce394f53 100644 --- a/python/packages/core/agent_framework/_workflows/_agent_executor.py +++ b/python/packages/core/agent_framework/_workflows/_agent_executor.py @@ -581,8 +581,8 @@ def _resolve_executor_kwargs(self, resolved: dict[str, Any] | None) -> dict[str, """ if not isinstance(resolved, dict): return None - global_kwargs = resolved.get(GLOBAL_KWARGS_KEY) - executor_kwargs = resolved.get(self.id) + global_kwargs: Any = resolved.get(GLOBAL_KWARGS_KEY) + executor_kwargs: Any = resolved.get(self.id) if global_kwargs is None and executor_kwargs is None: return None @@ -590,7 +590,7 @@ def _resolve_executor_kwargs(self, resolved: dict[str, Any] | None) -> dict[str, logger.warning( "Executor %s expected a dict for global kwargs, but got %s. Ignoring.", self.id, - type(global_kwargs), + cast(type[Any], type(global_kwargs)), ) return None @@ -598,7 +598,7 @@ def _resolve_executor_kwargs(self, resolved: dict[str, Any] | None) -> dict[str, logger.warning( "Executor %s expected a dict for its kwargs, but got %s. Ignoring.", self.id, - type(executor_kwargs), + cast(type[Any], type(executor_kwargs)), ) return None diff --git a/python/packages/core/agent_framework/_workflows/_workflow.py b/python/packages/core/agent_framework/_workflows/_workflow.py index 3178a976b49..53ff3cc3298 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow.py +++ b/python/packages/core/agent_framework/_workflows/_workflow.py @@ -220,8 +220,10 @@ class WorkflowInvocationKwargs: overrides. Plain mappings retain their existing global or per-executor behavior. """ - global_kwargs: Mapping[str, Any] = field(default_factory=dict) - executor_kwargs: Mapping[str, Mapping[str, Any]] = field(default_factory=dict) + global_kwargs: Mapping[str, Any] = field(default_factory=lambda: dict[str, Any]()) + executor_kwargs: Mapping[str, Mapping[str, Any]] = field( + default_factory=lambda: dict[str, Mapping[str, Any]]() + ) class Workflow(DictConvertible): diff --git a/python/packages/core/agent_framework/_workflows/_workflow_executor.py b/python/packages/core/agent_framework/_workflows/_workflow_executor.py index 97611024558..29ba90881cc 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow_executor.py +++ b/python/packages/core/agent_framework/_workflows/_workflow_executor.py @@ -6,7 +6,7 @@ import types from collections.abc import Mapping from dataclasses import dataclass -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, cast if TYPE_CHECKING: from ._workflow import Workflow @@ -388,8 +388,11 @@ async def process_workflow(self, input_data: object, ctx: WorkflowContext[Any, A raw_key = ( RAW_FUNCTION_INVOCATION_KWARGS_KEY if key == "function_invocation_kwargs" else RAW_CLIENT_KWARGS_KEY ) - resolved = parent_kwargs.get(raw_key, parent_kwargs.get(key)) - if isinstance(resolved, dict) or resolved is not None: + resolved = cast( + WorkflowInvocationKwargs | Mapping[str, Any] | None, + parent_kwargs.get(raw_key, parent_kwargs.get(key)), + ) + if resolved is not None: if key == "function_invocation_kwargs": fi_kwargs = resolved else: @@ -398,8 +401,8 @@ async def process_workflow(self, input_data: object, ctx: WorkflowContext[Any, A # Run the sub-workflow and collect all events, passing parent kwargs result = await self.workflow.run( input_data, - function_invocation_kwargs=fi_kwargs, # type: ignore - client_kwargs=ci_kwargs, # type: ignore + function_invocation_kwargs=fi_kwargs, + client_kwargs=ci_kwargs, ) logger.debug(f"WorkflowExecutor {self.id} sub-workflow {self.workflow.id} completed with {len(result)} events") diff --git a/python/packages/core/tests/core/test_serializable_mixin.py b/python/packages/core/tests/core/test_serializable_mixin.py index 03853e83868..59c4238dfec 100644 --- a/python/packages/core/tests/core/test_serializable_mixin.py +++ b/python/packages/core/tests/core/test_serializable_mixin.py @@ -572,6 +572,61 @@ def __init__(self, items: list, opaque: Any = None, additional_properties: dict assert cloned.items is not obj.items assert cloned.items == ["a"] + def test_shallow_copy_preserves_pickle_omitted_fields(self): + """Shallow copies retain runtime fields that pickle omits.""" + + class TestClass(SerializationMixin): + def __init__(self, raw_representation: Any): + self.raw_representation = raw_representation + + raw = object() + cloned = copy.copy(TestClass(raw)) + + assert cloned.raw_representation is raw + + def test_pickle_restores_slot_fields(self): + """Pickle state should include fields declared in slots.""" + + class TestClass(SerializationMixin): + __slots__ = ("value",) + + def __init__(self, value: str): + self.value = value + + original = TestClass("value") + restored = TestClass.__new__(TestClass) + restored.__setstate__(original.__getstate__()) + + assert restored.value == "value" + + def test_pickle_restores_legacy_tuple_state(self): + """Pickle restoration should accept the legacy dict-and-slots tuple.""" + + class TestClass(SerializationMixin): + __slots__ = ("value",) + + def __init__(self): + self.value = "new" + + restored = TestClass.__new__(TestClass) + restored.__setstate__(({"other": "dict"}, {"value": "legacy"})) + + assert restored.value == "legacy" + + def test_pickle_omission_is_separate_from_shallow_copy_policy(self): + """Fields shallow-copied by default remain persistent unless explicitly omitted.""" + + class TestClass(SerializationMixin): + _PICKLE_OMIT_FIELDS = set() + + def __init__(self, raw_representation: Any): + self.raw_representation = raw_representation + + raw = {"provider": "value"} + state = TestClass(raw).__getstate__() + + assert state["raw_representation"] == raw + def test_dependency_dict_merge_does_not_mutate_input(self): """Test that dict dependency merging does not mutate the caller's input dictionary.""" diff --git a/python/packages/core/tests/workflow/test_agent_executor.py b/python/packages/core/tests/workflow/test_agent_executor.py index 9e124db1d9c..9d7e4978a06 100644 --- a/python/packages/core/tests/workflow/test_agent_executor.py +++ b/python/packages/core/tests/workflow/test_agent_executor.py @@ -1,5 +1,6 @@ # Copyright (c) Microsoft. All rights reserved. +import pickle from collections.abc import AsyncIterable, Awaitable from typing import Any, Literal, overload From 9e661c9026fd2d356d9dbb24b66a11e187712d5d Mon Sep 17 00:00:00 2001 From: droideronline Date: Tue, 8 Sep 2026 13:31:25 +0530 Subject: [PATCH 4/5] Fix workflow cancellation and response routing --- .../core/agent_framework/_workflows/_agent.py | 48 +++++++++++++++---- .../agent_framework/_workflows/_workflow.py | 11 ++--- 2 files changed, 44 insertions(+), 15 deletions(-) diff --git a/python/packages/core/agent_framework/_workflows/_agent.py b/python/packages/core/agent_framework/_workflows/_agent.py index 8936db68d29..8198ad297e7 100644 --- a/python/packages/core/agent_framework/_workflows/_agent.py +++ b/python/packages/core/agent_framework/_workflows/_agent.py @@ -152,7 +152,7 @@ def run( self, messages: AgentRunInputs | None = None, *, - stream: Literal[True], + stream: Literal[False] = ..., session: AgentSession | None = None, checkpoint_id: str | None = None, checkpoint_storage: CheckpointStorage | None = None, @@ -165,11 +165,11 @@ def run( ) -> ResponseStream[AgentResponseUpdate, AgentResponse]: ... @overload - async def run( + def run( self, messages: AgentRunInputs | None = None, *, - stream: Literal[False] = ..., + stream: Literal[True], session: AgentSession | None = None, checkpoint_id: str | None = None, checkpoint_storage: CheckpointStorage | None = None, @@ -179,7 +179,7 @@ async def run( | Mapping[str, Any] | None = None, client_kwargs: WorkflowInvocationKwargs | Mapping[str, Mapping[str, Any]] | Mapping[str, Any] | None = None, - ) -> AgentResponse: ... + ) -> ResponseStream[AgentResponseUpdate, AgentResponse]: ... def run( self, @@ -488,7 +488,8 @@ async def _run_core( # NOTE: It is possible that some pending requests are not fulfilled, # and we will let the workflow to handle this -- the agent does not # have an opinion on this. - function_responses = self._extract_function_responses(input_messages) + pending_requests = await self.workflow._runner_context.get_pending_request_info_events() # pyright: ignore[reportPrivateUsage] + function_responses = self._extract_function_responses(input_messages, pending_requests) if streaming: async for event in self.workflow.run( responses=function_responses, @@ -765,22 +766,51 @@ def _process_request_info_event( arguments=args, ) - def _extract_function_responses(self, input_messages: Sequence[Message]) -> dict[str, Any]: + def _extract_function_responses( + self, + input_messages: Sequence[Message], + pending_requests: Mapping[str, WorkflowEvent[Any]] | None = None, + ) -> dict[str, Any]: """Extract function responses from input messages. The responses are for pending requests that the workflow is waiting on, and will be passed to the workflow. The pending requests are processed to either `function_approval_request` or `function_call` content by `_process_request_info_event`. """ + pending_requests = pending_requests or {} function_responses: dict[str, Any] = {} for message in input_messages: for content in message.contents: if content.type == "function_approval_response": - request_id: str = content.id # type: ignore[assignment] + request_id = content.id + if request_id is None: + raise AgentInvalidResponseException("Function approval response is missing its request ID.") function_responses[request_id] = content elif content.type == "function_result": - response_data = content.result if hasattr(content, "result") else str(content) - function_responses[content.call_id] = response_data # type: ignore + request_id = content.call_id + if request_id is None: + raise AgentInvalidResponseException("Function result is missing its call ID.") + response_request_id = request_id + pending_request = pending_requests.get(response_request_id) + if pending_request is None: + matching_requests = [ + (pending_id, pending_event) + for pending_id, pending_event in pending_requests.items() + if isinstance(pending_event.data, Content) + and pending_event.data.type == "function_call" + and pending_event.data.call_id == request_id + ] + if len(matching_requests) == 1: + response_request_id, pending_request = matching_requests[0] + response_data = ( + content + if pending_request is not None + and pending_request.response_type is Content + and isinstance(pending_request.data, Content) + and pending_request.data.type == "function_call" + else content.result + ) + function_responses[response_request_id] = response_data else: raise AgentInvalidResponseException( "Unexpected content type while awaiting request info responses." diff --git a/python/packages/core/agent_framework/_workflows/_workflow.py b/python/packages/core/agent_framework/_workflows/_workflow.py index 2ab75d7330a..6e9f256b9bd 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow.py +++ b/python/packages/core/agent_framework/_workflows/_workflow.py @@ -544,7 +544,6 @@ async def _run_workflow_with_tracing( OtelAttr.WORKFLOW_RUN_SPAN, attributes, ) as span: - saw_request = False emitted_in_progress_pending = False try: # Add workflow started event (telemetry + surface state to consumers) @@ -609,9 +608,6 @@ async def _run_workflow_with_tracing( # All executor executions happen within workflow span async for event in self._runner.run_until_convergence(): - # Track request events for final status determination - if event.type == "request_info": - saw_request = True yield event if event.type == "request_info" and not emitted_in_progress_pending: @@ -620,8 +616,11 @@ async def _run_workflow_with_tracing( with _framework_event_origin(): pending_status = WorkflowEvent.status(self._status) yield pending_status - # Workflow runs until idle - emit final status based on whether requests are pending - if saw_request: + # Workflow runs until idle - emit final status based on whether requests are pending. + # Continuations such as cancellation may retain an existing sibling request without + # re-emitting its request_info event during this run. + pending_requests = await self._runner.context.get_pending_request_info_events() + if pending_requests: self._status = WorkflowRunState.IDLE_WITH_PENDING_REQUESTS with _framework_event_origin(): terminal_status = WorkflowEvent.status(self._status) From 486a39a09ae26474752fa1bf764a8d16c041cfdd Mon Sep 17 00:00:00 2001 From: droideronline Date: Tue, 8 Sep 2026 13:35:59 +0530 Subject: [PATCH 5/5] Fix nested workflow request cancellation --- .../agent_framework/_workflows/_workflow_executor.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/python/packages/core/agent_framework/_workflows/_workflow_executor.py b/python/packages/core/agent_framework/_workflows/_workflow_executor.py index 89dd21c68db..749283d2d74 100644 --- a/python/packages/core/agent_framework/_workflows/_workflow_executor.py +++ b/python/packages/core/agent_framework/_workflows/_workflow_executor.py @@ -463,6 +463,15 @@ async def handle_propagated_request_response( ctx=ctx, ) + @override + async def _cancel_pending_request(self, request_id: str, ctx: WorkflowContext[Any, Any]) -> None: + """Propagate cancellation into the wrapped workflow.""" + result = await self.workflow.cancel_pending_requests( + [request_id], + tools=ctx.get_runtime_tools(), + ) + await self._process_workflow_result(result, ctx) + @override async def on_checkpoint_save(self) -> dict[str, Any]: """Get the current state of the WorkflowExecutor for checkpointing purposes."""