Exploring Multi-Agent Workflows with Microsoft Agent Framework

As organizations race to automate decision-making, content generation, analysis, and execution at scale, the future is shifting from isolated AI agents to multi-agent workflows where specialized agents collaborate, delegate, challenge, and refine each other's work. By allowing multiple AI agents to work together, organizations can tackle complex business processes more efficiently, improve accuracy, and scale operations in ways that would be difficult for a single agent to achieve alone.

In this blog, we will explore 5 types of multi-agent workflows offered by the Microsoft Agent Framework to design, manage, and scale complex multi-agent workflows.

Prerequisites for the Tutorial:

1. Azure Subscription
2. Microsoft Foundry resource deployed in a resource group. Deploy any model (such as gpt-4.1-mini) in the Foundry project.

1. Concurrent Orchestration

In this orchestration, the same input is sent to multiple agents simultaneously and consolidated. Each agent handles tasks independently and the results are combined. All agents work at the same time.

Used when:

1. You need different approaches/perspectives for a problem
2. Group decision making-based scenarios
3. Voting-based scenarios

Use Case: Ticket Assessment on Various Criteria

Consider a customer-support use case in which a new ticket must be assessed quickly and routed correctly. With concurrent orchestration, the same incoming ticket is sent to three specialized agents at the same time.

import os
import asyncio
from typing import cast
from agent_framework import Message
from agent_framework.foundry import FoundryChatClient
from azure.identity import DefaultAzureCredential
from agent_framework.orchestrations import ConcurrentBuilder
from dotenv import load_dotenv
load_dotenv()

async def main():
    credential = DefaultAzureCredential()

    chat_client = FoundryChatClient(
        credential=credential,
        project_endpoint=os.getenv("AZURE_AI_PROJECT_ENDPOINT"),
        model=os.getenv("AZURE_AI_MODEL_DEPLOYMENT_NAME")
    )

    sentiment_agent = chat_client.as_agent(
        name="Sentiment Agent",
        instructions="You are a helpful assistant that analyzes the sentiment of a support ticket."
    )

    category_agent = chat_client.as_agent(
        name="Category Agent",
        instructions="You are a helpful assistant that categorizes a support ticket into categories such as Billing, Technical, Refund, or Account."
    )

    priority_agent = chat_client.as_agent(
        name="Priority Agent",
        instructions="You are a helpful assistant that determines the priority of a support ticket as High, Medium, or Low."
    )

    workflow = ConcurrentBuilder(
        participants = [sentiment_agent, category_agent, priority_agent]
    ).build()

    result = await workflow.run("I was charged twice and I'm furious — refund me now!")
    outputs = result.get_outputs()

    i = 1
    for response in outputs:
        for msg in cast(list[Message], response.messages):
            name = msg.author_name or ("assistant" if msg.role == "assistant" else "user")
            print(f"{'-' * 60}\n{i:02d} [{name}]\n{msg.text}")
            i += 1

if __name__ == "__main__":
    asyncio.run(main())

2. Sequential Orchestration

Used when the output of one agent is consumed by subsequent agents one after another. This pattern is ideal for workflows where each step depends on the previous one.

Used When:

1. There is a multi-step process, where each step relies on the output of the previous one.
2. Situations that benefit from iterative refinement, such as drafting, reviewing, and improving content.
3. Each stage produces an output and the next output builds upon that output.

Use Case: Automated Support Ticket Triage

Consider a customer-support operation that receives large volumes of unstructured tickets and must route each one accurately. In this sequential workflow, a Summarizer Agent first condenses the raw ticket into one or two sentences that capture the customer’s core intent. Its output is then passed to a Classifier Agent, which assigns exactly one category—Billing, Technical, Refund, or Urgent.

import os
import asyncio
from typing import cast
from agent_framework import Message
from agent_framework.foundry import FoundryChatClient
from azure.identity import DefaultAzureCredential
from agent_framework.orchestrations import SequentialBuilder
from dotenv import load_dotenv
load_dotenv()

async def main():
    credential = DefaultAzureCredential()

    chat_client = FoundryChatClient(
        credential=credential,
        project_endpoint=os.getenv("AZURE_AI_PROJECT_ENDPOINT"),
        model=os.getenv("AZURE_AI_MODEL_DEPLOYMENT_NAME")
    )

    summarizer_agent = chat_client.as_agent(
        name="Summarizer Agent",
        description="Summarizes a support ticket into 1-2 sentences of core intent.",
        instructions="You are a helpful assistant that summarizes support tickets into concise summaries."
    )

    classifier_agent = chat_client.as_agent(
        name="Classifier Agent",
        description="Classifies a ticket summary into: Billing, Technical, Refund, or Urgent.",
        instructions="You are a helpful assistant that classifies a support ticket summary strictly into one of the following categories: Billing, Technical, Refund, or Urgent."
    )

    workflow = SequentialBuilder(
    participants=[summarizer_agent, classifier_agent],
    output_from = "all"
    ).build()

    ticket = "I was charged twice for my subscription this month and need a refund ASAP."
    result = await workflow.run(ticket)
    outputs = result.get_outputs()
    
    i = 1
    for response in outputs:
        for msg in cast(list[Message], response.messages):
            name = msg.author_name or ("assistant" if msg.role == "assistant" else "user")
            print(f"{'-' * 60}\n{i:02d} [{name}]\n{msg.text}")
            i += 1


if __name__ == "__main__":
    asyncio.run(main())

3. Group Chat Orchestration

Manages a collaborative conversation between multiple agents, optionally involving a human in the process (Human-in-the-Loop). There is a central chat manager that decides which agent responds next and when to request for human input.

Used when:

1. Scenarios that require debates or group brainstorming.

Use Case: Cross-Functional Feature Proposal Review

A product team uses specialist agents to review a feature proposal in one shared discussion. Product Agent assesses value, Engineering Agent feasibility, Design Agent usability, and Security Agent compliance. A Manager Agent guides the debate and concludes with a recommendation to proceed, revise, or reject.

import os
import asyncio
from typing import cast
from agent_framework import AgentResponseUpdate, Message
from agent_framework.foundry import FoundryChatClient
from azure.identity import DefaultAzureCredential
from agent_framework.orchestrations import GroupChatBuilder
from dotenv import load_dotenv
load_dotenv()

async def main():
    credential = DefaultAzureCredential()

    chat_client = FoundryChatClient(
        credential=credential,
        project_endpoint=os.getenv("AZURE_AI_PROJECT_ENDPOINT"),
        model=os.getenv("AZURE_AI_MODEL_DEPLOYMENT_NAME")
    )

    product_agent = chat_client.as_agent(
        name="Product Agent",
        instructions="You are a helpful assistant that argues for user value and business priority.",
        default_options={"store": False} # turn off server-side storage, so the client keeps history in the local session and re-sends the full conversation every turn.
    )

    engineering_agent = chat_client.as_agent(
        name="Engineering Agent",
        instructions="You are a helpful assistant that raises engineering feasibility, effort, and technical risks.",
        default_options={"store": False}
    )

    design_agent = chat_client.as_agent(
        name="Design Agent",
        instructions="You are a helpful assistant that focuses on UX and usability concerns.",
        default_options={"store": False}
    )

    security_agent = chat_client.as_agent(
        name="Security Agent",
        instructions="You are a helpful assistant that flags compliance and data-protection issues.",
        default_options={"store": False}
    )

    manager_agent = chat_client.as_agent(
        name="Manager Agent",
        instructions=(
            "You moderate a design-review discussion. Each turn, choose the SINGLE next "
            "participant to speak from: Product Agent, Engineering Agent, Design Agent, "
            "Security Agent. Never select the same participant twice in a row. Once every "
            "perspective has been heard and a clear decision is reached, terminate the "
            "conversation with a short recommendation."
        ),
        default_options={"store": False}  
    )

    workflow = GroupChatBuilder(
        participants=[product_agent, engineering_agent, design_agent, security_agent],
        orchestrator_agent=manager_agent,
        max_rounds=5,
        intermediate_output_from="all",
    ).build()

    stream = await workflow.run("Proposal: add biometric login to the mobile app. Should we build it next quarter?", 
                                stream=True)

    last_executor = None
    async for event in stream:
        if event.type in ("intermediate", "output") and isinstance(event.data, AgentResponseUpdate):
            executor = event.data.author_name
            if executor != last_executor:
                print(f"\n{'-' * 60}\n{executor}:\n")
                last_executor = executor
            print(event.data.text, end="", flush=True)

    result = await stream.get_final_response()
    print(f"\n{'=' * 60}\nFinished ({len(result.get_outputs())} output messages).")

if __name__ == "__main__":
    asyncio.run(main())

4. Handoff Orchestration

This orchestration lets agents assign a task to other agents based on their expertise.

Used when:

1. Multiple agents are involved, but the order of execution is unknown/non-deterministic.

Use Case: Dynamic Customer Support Routing

A Triage Agent greets the customer, identifies the issue, and hands the conversation to the right specialist: a Refund Agent for refunds and returns, or an Order Status Agent for shipping updates. If the customer’s need changes, the specialist hands the conversation back to triage for seamless rerouting.

import os
import asyncio
from typing import cast
from agent_framework import AgentResponseUpdate, Message
from agent_framework.foundry import FoundryChatClient
from azure.identity import DefaultAzureCredential
from agent_framework.orchestrations import HandoffBuilder
from dotenv import load_dotenv
load_dotenv()

async def main():
    credential = DefaultAzureCredential()

    chat_client = FoundryChatClient(
        credential=credential,
        project_endpoint=os.getenv("AZURE_AI_PROJECT_ENDPOINT"),
        model=os.getenv("AZURE_AI_MODEL_DEPLOYMENT_NAME"),
    )

    triage_agent = chat_client.as_agent(
        name="triage_agent",
        instructions=(
                "Greet the customer and briefly acknowledge their issue in one sentence, "
                "then hand off to the right specialist."
        ),        
        default_options={"store": False},
        require_per_service_call_history_persistence=True
    )

    refund_agent = chat_client.as_agent(
        name="refund_agent",
        instructions="You process refunds and returns. Ask for the order number and resolve the request.",
        default_options={"store": False},
        require_per_service_call_history_persistence=True
    )

    order_status_agent = chat_client.as_agent(
        name="order_status_agent",
        description="Answers questions about order status and shipping.",
        instructions="You answer order status and shipping questions.",
        default_options={"store": False},
        require_per_service_call_history_persistence=True
    )

    workflow = (
        HandoffBuilder(participants=[triage_agent, refund_agent, order_status_agent])
        .with_start_agent(triage_agent)
        .add_handoff(triage_agent, [refund_agent, order_status_agent])  # triage can route to either
        .add_handoff(refund_agent, [triage_agent])                     # specialists can hand back
        .add_handoff(order_status_agent, [triage_agent])
        .build()
    )

    # Interactive loop: run once, then answer each request_info until it ends.
    user_message: str | None = "Hi, I was charged twice for order #12345 and want a refund."
    responses: dict | None = None
    last_executor = None

    while True:
        if responses is not None:
            stream = workflow.run(responses=responses, stream=True)
        else:
            stream = workflow.run(user_message, stream=True)

        pending_request_id = None
        async for event in stream:
            if event.type in ("intermediate", "output") and isinstance(event.data, AgentResponseUpdate):
                executor = event.data.author_name
                if executor != last_executor:
                    print(f"\n{'-' * 60}\n{executor}:\n")
                    last_executor = executor
                print(event.data.text, end="", flush=True)
            elif event.type == "request_info":
                pending_request_id = event.request_id  # workflow is waiting for the user

        if pending_request_id is None:
            break  # no input requested -> conversation finished

        user_text = input("\n\nYou: ")
        if not user_text.strip():
            break  # empty answer terminates the handoff workflow
        responses = {pending_request_id: [Message(role="user", contents=[user_text])]}

if __name__ == "__main__":
    asyncio.run(main())

5. Magentic Orchestration

This orchestration allows for dynamic collaboration of multiple agents. Used when the exact workflow is not known upfront, and for complex open-ended problems.

There is a task ledger which keeps track of the tasks that need to be done. The progress ledger keeps track of the tasks completed and what all was learnt.

The manager agent does the following:

1. Maintains the overall goal
2. Creates and updates the task ledger
3. Chooses which agent should work next
4. Tracks progress in the progress ledger
5. Replans when stuck
6. Synthesizes the final answer

Use Case: Autonomous Blog Drafting and Review

Given a topic, a Magentic manager plans and drives an iterative draft-and-review cycle: it directs a writer agent to produce the post, routes the draft to an editor agent for clarity and length feedback, loops back for revisions when needed, and finalizes the post once it meets the quality bar — all without a human specifying the step order.

import os
import asyncio
from typing import cast
from agent_framework import AgentResponseUpdate
from agent_framework.foundry import FoundryChatClient
from azure.identity import DefaultAzureCredential
from agent_framework.orchestrations import MagenticBuilder
from dotenv import load_dotenv
load_dotenv()

async def main():
    credential = DefaultAzureCredential()
    
    chat_client = FoundryChatClient(
        credential=credential,
        project_endpoint=os.getenv("AZURE_AI_PROJECT_ENDPOINT"),
        model=os.getenv("AZURE_AI_MODEL_DEPLOYMENT_NAME")
    )

    writer_agent = chat_client.as_agent(
        name="Writer Agent",
        instructions="You write clear, engaging blog posts on the requested topic.",
        default_options={"store": False},
    )

    editor_agent = chat_client.as_agent(
        name="Editor Agent",
        instructions="You review drafts for clarity and length, and suggest concise improvements.",
        default_options={"store": False},
    )

    manager_agent = chat_client.as_agent(
        name="Manager Agent",
        instructions="You coordinate the writer and editor to produce a polished final blog post.",
        default_options={"store": False},
    )

    workflow = MagenticBuilder(
        participants=[writer_agent, editor_agent],
        manager_agent=manager_agent,
        max_stall_count=2,
        max_round_count=10,
        intermediate_output_from="all").build()

    stream = workflow.run(
        "Write a 300-word blog post explaining why sleep matters for productivity.",
        stream=True,
    )

    last_executor = None
    async for event in stream:
        if event.type in ("intermediate", "output") and isinstance(event.data, AgentResponseUpdate):
            executor = event.data.author_name
            if executor != last_executor:
                print(f"\n{'-' * 60}\n{executor}:\n")
                last_executor = executor
            print(event.data.text, end="", flush=True)

    result = await stream.get_final_response()
    print(f"\n{'=' * 60}\nFinished ({len(result.get_outputs())} output messages).")

if __name__ == "__main__":
    asyncio.run(main())

Difference Between Group Chat, Handoff, and Magentic Orchestration

Next Steps:

1. Try it yourself - Clone the GitHub Repo to get started!

2. Build upon the Magentic Orchestration use case - "Write a well-researched 800-word article on whether biometric login improves security. Verify claims, add real statistics, and include a counter-argument section." Add participants agents such as researcher_agent, fact_checker_agent, writer_agent, and editor_agent

Related Resources:

1. Reference Used in the Blog (Including images): Introduction - Training | Microsoft Learn
2. More on Magentic Orchestration: Use Magentic Orchestration - Training | Microsoft Learn