* fix(a2a): reject empty invocations explicitly Key decisions: - Keep A2A continuation authority explicit; durable session task state only enriches diagnostics. - Raise AgentInvalidRequestException with participant and available task context instead of inventing input. - Leave AgentExecutor and Group Chat production contracts unchanged. Files changed: - packages/a2a/agent_framework_a2a/_agent.py - packages/a2a/tests/test_a2a_agent.py - packages/a2a/tests/test_a2a_group_chat.py Notes for next iteration: - No blockers. INPUT_REQUIRED pause/resume remains a separate task. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * fix(a2a): pause group chat for remote input Key decisions: - Translate A2A INPUT_REQUIRED task content into the existing Content user-input-request contract. - Use the remote task ID as stable request correlation for streamed and finalized responses. - Reuse AgentExecutor request handling so caller input resumes the same task without a workflow-specific A2A path. Files changed: - packages/a2a/agent_framework_a2a/_agent.py - packages/a2a/tests/test_a2a_agent.py - packages/a2a/tests/test_a2a_group_chat.py Notes for next iteration: - Checkpoint restoration of pending A2A input is now unblocked. - The local issue file could not be moved because repository issue files are restricted by content exclusion policy. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * fix(a2a): restore pending input from checkpoints Key decisions: - Keep normalized INPUT_REQUIRED content durable by excluding transport-only protobuf raw representations. - Restore through the existing AgentExecutor checkpoint and request-response path without a new schema or continuation API. - Cover file-backed restoration in streaming and non-streaming Group Chat runs, including unrelated-response rejection and exact task resumption. Files changed: - packages/a2a/agent_framework_a2a/_agent.py - packages/a2a/tests/test_a2a_group_chat.py Notes for next iteration: - The local issue file could not be moved because repository issue files are restricted by content exclusion policy. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * test(handoff): lock textless target context Key decisions: - Exercise the built Handoff workflow in streaming and non-streaming modes instead of bypassing routing, sessions, or termination. - Keep the slice test-only because current production already carries the initial task to a textless handoff target without synthetic user input. - Revisit the source to verify its handoff function call retains a matching result and user-turn termination sees only caller messages. Files changed: - packages/orchestrations/tests/test_handoff.py Notes for next iteration: - No production defect was reproduced. - The local issue file could not be moved because repository issue files are restricted by content exclusion policy. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * test(handoff): use resolved IDs in event assertions * fix(workflows): preserve A2A input request semantics * fix(workflows): preserve input request correlation * fix(a2a): deduplicate message-less input requests * fix(workflows): preserve specialized input requests * test(openai): use current web search model --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Agent Framework Orchestrations
Orchestration patterns for Microsoft Agent Framework. This package provides high-level builders for common multi-agent workflow patterns.
Installation
pip install agent-framework-orchestrations
Orchestration Patterns
SequentialBuilder
Chain agents/executors in sequence, passing conversation context along:
from agent_framework.orchestrations import SequentialBuilder
workflow = SequentialBuilder(participants=[agent1, agent2, agent3]).build()
# Preserve agent1 and agent2 as visible progress, while the default builder output remains Workflow Output.
workflow = SequentialBuilder(
participants=[agent1, agent2, agent3],
intermediate_output_from=[agent1, agent2],
).build()
ConcurrentBuilder
Fan-out to multiple agents in parallel, then aggregate results:
from agent_framework.orchestrations import ConcurrentBuilder
workflow = ConcurrentBuilder(participants=[agent1, agent2, agent3]).build()
HandoffBuilder
Decentralized agent routing where agents decide handoff targets:
from agent_framework.orchestrations import HandoffBuilder
workflow = (
HandoffBuilder()
.participants([triage, billing, support])
.with_start_agent(triage)
.build()
)
GroupChatBuilder
Orchestrator-directed multi-agent conversations:
from agent_framework.orchestrations import GroupChatBuilder
workflow = GroupChatBuilder(
participants=[agent1, agent2],
selection_func=my_selector,
intermediate_output_from=[agent1, agent2],
).build()
MagenticBuilder
Sophisticated multi-agent orchestration using the Magentic One pattern:
from agent_framework.orchestrations import MagenticBuilder
workflow = MagenticBuilder(
participants=[researcher, writer, reviewer],
manager_agent=manager_agent,
intermediate_output_from=[researcher, writer, reviewer],
).build()
Output Selection
Orchestration builders expose Workflow Output selection using participant names. The core rule is that output_from
is an allow-list for Workflow Output, not a routing rule for every other participant output. Unselected participant
payloads are hidden unless intermediate_output_from explicitly selects them as Intermediate Output.
output_fromdesignates participant emissions as Workflow Output (type='output'events).intermediate_output_fromdesignates participant emissions as Intermediate Output (type='intermediate'events).
If neither list is provided, each builder uses its documented default Workflow Output contract. Sequential emits the last participant; Concurrent, GroupChat, and Magentic emit their aggregator/orchestrator/manager output; Handoff emits participants.
| Selection | Workflow Output | Intermediate Output | Hidden payloads |
|---|---|---|---|
| Omit both selections | Builder default Workflow Output contract | None | Builder-specific non-output participant payloads |
output_from="all" |
Every output-capable participant | None | None |
output_from=[writer] |
Only writer |
None | All other participant payloads |
output_from=[writer], intermediate_output_from="all_other" |
Only writer |
Every output-capable participant not selected by output_from |
None |
intermediate_output_from="all_other" |
None, except builder-internal default output executors where applicable | Every output-capable participant | Builder-internal plumbing payloads |
output_from=[], intermediate_output_from="all_other" |
None, except builder-internal default output executors where applicable | Every output-capable participant | Builder-internal plumbing payloads |
output_from=[writer], intermediate_output_from=[researcher, reviewer] |
Only writer |
researcher and reviewer |
Any other participant payloads |
Invalid selections fail at construction or build time:
| Invalid selection | Why it fails |
|---|---|
output_from="all_other" |
"all_other" is only valid for intermediate_output_from |
intermediate_output_from="all" |
"all" is only valid for output_from |
| The same participant in both selections | One payload cannot be both Workflow Output and Intermediate Output |
| Duplicate participant selections | Duplicates are treated as configuration errors |
| Unknown participant selections | Typos and missing participants are rejected |
output_from=[], intermediate_output_from=[] |
Both explicit selections are empty |
When an orchestration is wrapped with workflow.as_agent(), Workflow Output becomes normal response text. Intermediate
Output becomes text_reasoning content so callers can inspect progress without changing .text behavior.
Documentation
For more information, see the Agent Framework documentation.