Microsoft 365 multi-agent workflow with Microsoft Agent Framework

Most production AI tasks are too complex for a single agent. You need specialists that collaborate, a destination expert that researches the best places to visit, a budget analyst that breaks down costs, an activities curator that finds the best experiences, all coordinated by an orchestration layer you can test locally, deploy to Microsoft Foundry, and surface to end users in Microsoft 365 (Teams or Copilot).

This article takes you through a self-contained runnable example inspired by the end-to-end samples in the Microsoft Agent Framework repository. Each section builds on the previous one:

You can stop at any layer. A WorkflowBuilder workflow works locally without Foundry; a hosted agent works without M365. The stack composes, you adopt what you need.

What we'll build

1. Build the logic with workflow and tools

Agent Framework provides WorkflowBuilder, a directed-graph orchestrator: connect agents with edges, and the framework handles fan-out (parallel branches), fan-in (wait for all branches), and sequential chains. Use add_edge() to build any topology: diamond, pipeline, or tree. Agents become truly useful when they can also call external functions via the @tool decorator so we'll wire both concepts together in a single module.

Diamond workflow with a tool-equipped Destination Expert

The diamond workflow sample (example/hosted_agents_in_workflow.py, inspired by the upstream agents_in_workflow) chains four agents in a diamond. The Destination Expert gets a @tool to ground its research in structured destination data:

import os
from typing import Annotated

from agent_framework import Agent, WorkflowBuilder, tool
from agent_framework.foundry import FoundryChatClient
from azure.identity import DefaultAzureCredential
from dotenv import load_dotenv
from pydantic import Field

load_dotenv()

PROJECT_ENDPOINT = os.getenv("AZURE_AI_PROJECT_ENDPOINT")
MODEL = os.getenv("AZURE_AI_MODEL_DEPLOYMENT_NAME", "gpt-4.1-mini")

DESTINATION_DATA = {
    "japan": {
        "best_season": "March–May (cherry blossom) or Oct–Nov (autumn foliage)",
        "avg_daily_cost": "$150–250/person",
        "top_attractions": ["Tokyo", "Kyoto temples", "Mount Fuji", "Osaka street food"],
        "visa_note": "Visa-free for 70+ countries (up to 90 days)",
    },
    # ... Italy, Mexico ...
}


@tool(approval_mode="never_require")
def get_destination_info(
    destination: Annotated[str, Field(description="The travel destination to look up.")],
) -> str:
    """Look up best travel season, costs, top attractions, and visa info for a destination."""
    key = destination.lower().strip()
    for entry_key in sorted(DESTINATION_DATA, key=len, reverse=True):
        entry_lower = entry_key.lower()
        if entry_lower in key or key in entry_lower:
            data = DESTINATION_DATA[entry_key]
            lines = [f"Destination info for '{destination}':"]
            lines.append(f"  Best season: {data['best_season']}")
            lines.append(f"  Avg daily cost: {data['avg_daily_cost']}")
            lines.append(f"  Top attractions: {', '.join(data['top_attractions'])}")
            lines.append(f"  Visa note: {data['visa_note']}")
            return "\n".join(lines)
    return f"No destination data available for '{destination}'. Try 'Japan', 'Italy', or 'Mexico'."


def main():
    credential = DefaultAzureCredential()

    def make_client():
        return FoundryChatClient(
            project_endpoint=PROJECT_ENDPOINT, model=MODEL, credential=credential,
        )

    # The Destination Expert gets the tool — it's the one that looks up factual data
    dest = Agent(client=make_client(), name="destination_expert", instructions=(
        "You're an expert travel researcher. "
        "Given a vacation request, provide concise insights on the best destinations, "
        "ideal travel seasons, and must-see attractions."
    ), tools=get_destination_info)

    budget = Agent(client=make_client(), name="budget_analyst", instructions=(
        "You're a meticulous travel budget analyst. "
        "Break down estimated costs for flights, accommodation, food, and activities. "
        "Suggest money-saving tips."
    ))
    activities = Agent(client=make_client(), name="activities_curator", instructions=(
        "You're an enthusiastic activities curator. "
        "Recommend must-do experiences, hidden gems, and local favorites. "
        "Include a mix of culture, food, nature, and adventure."
    ))
    coordinator = Agent(client=make_client(), name="trip_coordinator", instructions=(
        "You are a Trip Coordinator. You receive research from three specialists: "
        "a Destination Expert, a Budget Analyst, and an Activities Curator. "
        "Synthesize their inputs into one coherent, actionable travel plan. "
        "Resolve any conflicts (e.g. suggested activities that exceed the budget) "
        "and present a single day-by-day itinerary with costs."
    ))

    # Diamond: dest fans out to budget + activities, both fan in to coordinator
    workflow = (
        WorkflowBuilder(start_executor=dest)
        .add_edge(dest, budget)
        .add_edge(dest, activities)
        .add_edge(budget, coordinator)
        .add_edge(activities, coordinator)
        .build()
    )

Four things to notice:

1. WorkflowBuilder + add_edge() creates the diamond topology. The Destination Expert runs first, then Budget Analyst and Activities Curator run in parallel on its output (the framework detects they share the same source and schedules them concurrently). The Trip Coordinator waits for both to finish, then synthesizes a single coherent plan.
2. Type annotations drive the tool schema. Annotated[str, Field(description=...)] generates the JSON schema the model sees. Pydantic's Field lets you add descriptions, constraints, and defaults.
3. approval_mode controls human-in-the-loop. "never_require" is fine for read-only lookups like destination data. For tools with side effects imagine a book_trip tool that charges a credit card use "always_require" so the framework pauses execution and waits for human approval before the tool runs.
4. Only the Destination Expert has a tool. The other specialists don't need one. tools=get_destination_info accepts a single tool or a list.

A user asks "Plan a 10-day trip to Japan for a family of 4 on a $5,000 budget", the destination expert calls get_destination_info("Japan") to retrieve real data, the budget analyst and activities curator each work on those grounded insights in parallel, and the coordinator weaves everything into one actionable itinerary.

This single build_fan_out_fan_in_workflow() function is the brain. Everything that follows: hosting, deployment, M365 distribution, wraps this same function without changing it.

2. Hosting and deploying

The workflow from Section 1 becomes an HTTP endpoint through a two-line bridge:

from azure.ai.agentserver.agentframework import from_agent_framework

# Wrap any Agent (single, workflow, or tool-equipped) and start the server
from_agent_framework(agent).run()         # sync
await from_agent_framework(agent).run_async()  # async

from_agent_framework() wraps your Agent in an HTTP server that:

• Listens on http://localhost:8088/responses (POST)
• Accepts the same JSON schema as the OpenAI Responses API (input, stream, conversation, etc.)
• Returns streaming or non-streaming responses in the same wire format

Any client that speaks the Responses protocol: Foundry, azd ai agent invoke, or a simple curl, works out of the box.

Here's the workflow from Section 1, now hosted:

# (continued from the WorkflowBuilder example above)
from azure.ai.agentserver.agentframework import from_agent_framework

from_agent_framework(lambda: workflow).run(port=8088)

# Terminal 1 – start the hosted agent
pip install -r requirements.txt   # if not already done
python -m example.hosted_agents_in_workflow

# Terminal 2 – invoke it
curl -sS -H "Content-Type: application/json" -X POST http://localhost:8088/responses \
  -d '{"input":"Plan a 10-day trip to Japan for a family of 4 on a $5,000 budget.","stream":false}'

The response contains insights from all four agents: destination recommendations, cost breakdown, curated activities, and a synthesized itinerary from the coordinator, in one API call.

Deploying to Foundry

The repo includes a Dockerfile and an agent.yaml manifest, everything you need to deploy the workflow to Microsoft Foundry as a hosted agent.

Dockerfile packages the workflow for container hosting:

FROM python:3.12-slim
WORKDIR /app
COPY . user_agent/
WORKDIR /app/user_agent
RUN pip install --no-cache-dir -r requirements.txt \
    && pip install --no-cache-dir azure-ai-agentserver-core==1.0.0b17 \
    && pip install --no-cache-dir --no-deps azure-ai-agentserver-agentframework==1.0.0b17
EXPOSE 8088
CMD ["python", "example/hosted_agents_in_workflow.py"]

agent.yaml declares the agent's protocol, environment variables, and model resources:

kind: hosted
name: vacation-planner
protocols:
    - protocol: responses
      version: v1
environment_variables:
    - name: AZURE_AI_PROJECT_ENDPOINT
      value: ${AZURE_AI_PROJECT_ENDPOINT}
    - name: AZURE_AI_MODEL_DEPLOYMENT_NAME
      value: ${AZURE_AI_MODEL_DEPLOYMENT_NAME}

The kind: hosted marker tells Foundry this is a container-based agent. Deploy with the Azure Developer CLI:

azd auth login
azd ai agent init             # scaffolds Bicep + agent.yaml (skip if you already have them)
azd provision                  # creates Foundry project, ACR, App Insights, RBAC
azd deploy                     # builds container, pushes to ACR, registers hosted agent
azd ai agent invoke            # conversation with the deployed agent

Splitting provision and deploy makes each step independently retriable, if the container build fails, you can re-run azd deploy without reprovisioning. You can also combine both into azd up.

Once deployed, invoke the agent directly from the CLI:

azd ai agent invoke --new-session "Plan a 10-day trip to Japan for a family of 4 on a \$5,000 budget."

azd ai agent init generates production-ready IaC (infra/main.bicep, managed identity with correct RBAC, Foundry Project + model deployment). Customize the Bicep, commit it, and run the same workflow in GitHub Actions for staging/prod promotion. See the hosted agents documentation for a full walkthrough.

3. Add an interface and distributing to Microsoft 365

Once your workflow logic works as a hosted endpoint, you can expose it to end users in Teams, Copilot, or any Microsoft 365 surface. The example/m365_vacation_planner.py example (inspired by the upstream m365-agent sample) bridges Agent Framework with the Microsoft 365 Agents SDK, and the logic doesn't change at all. It's a single import away.

The M365 SDK wraps your existing workflow with three components:

from aiohttp import web
from microsoft_agents.hosting.aiohttp import CloudAdapter, start_agent_process
from microsoft_agents.hosting.aiohttp.jwt_authorization_middleware import (
    jwt_authorization_middleware,
)
from microsoft_agents.hosting.core import (
    AgentApplication,
    MemoryStorage,
    TurnContext,
    TurnState,
)
from microsoft_agents.hosting.core.authorization import AgentAuthConfiguration

# Import the brain from Section 1 — no duplication, one line
from example.hosted_agents_in_workflow import build_fan_out_fan_in_workflow

workflow = build_fan_out_fan_in_workflow()
storage = MemoryStorage()
adapter = CloudAdapter()
agent_app = AgentApplication[TurnState](storage=storage, adapter=adapter)


@agent_app.activity("message")
async def on_message(context: TurnContext, _: TurnState):
    user_message = context.activity.text or ""
    if not user_message.strip():
        return
    result = await workflow.run(user_message)
    outputs = result.get_outputs()
    reply = outputs[-1] if outputs else "No response generated."
    await context.send_activity(str(reply))


async def health(_: web.Request) -> web.Response:
    return web.json_response({"status": "ok"})

async def entry_point(req: web.Request) -> web.Response:
    return await start_agent_process(req, agent_app, adapter)

auth_config = AgentAuthConfiguration(anonymous_allowed=True)
app = web.Application(middlewares=[jwt_authorization_middleware])
app["agent_configuration"] = auth_config
app.add_routes([
    web.post("/api/messages", entry_point),
    web.get("/api/health", health),
])

if __name__ == "__main__":
    web.run_app(app, host="localhost", port=3978)

What each piece does:

• from example.hosted_agents_in_workflow import build_fan_out_fan_in_workflow: This is the payoff of progressive design. The workflow, tools, and all agent logic live in one module; the M365 file is a thin channel wrapper.
• AgentApplication[TurnState]: The M365 SDK's application host. It manages turn state, authentication, and activity routing.
• ​_app.activity("message"): Routes incoming user messages to your workflow. You call workflow.run(), extract outputs with result.get_outputs(), and send the final output back through context.send_activity().
• jwt_authorization_middleware + AgentAuthConfiguration: Handles authentication. With anonymous_allowed=True, unauthenticated requests get anonymous claims so the adapter can process them without Entra credentials (ideal for local dev).
• CloudAdapter + start_agent_process: Bridges aiohttp HTTP requests into the M365 SDK processing pipeline.

The key insight: your build_fan_out_fan_in_workflow() function is identical whether it's hosted via from_agent_framework().run() (Section 2) or wrapped in AgentApplication (here). The framework doesn't care about the distribution channel, and neither does your code.

Running locally (anonymous mode)

export USE_ANONYMOUS_MODE=True
export PORT=3978
export AZURE_AI_PROJECT_ENDPOINT="https://<your-resource>.services.ai.azure.com/api/projects/<project>"
export AZURE_AI_MODEL_DEPLOYMENT_NAME="gpt-4o"

az login
python -m example.m365_vacation_planner

Best practices

• Start with prompt agents for rapid iteration, move to hosted agents when you need custom Python tools, multi-agent workflows, or third-party libraries.
• Use DefaultAzureCredential / AzureCliCredential locally and Managed Identity in production. Never embed keys.
• Use azd ai agent monitor --session-id <id> --follow during development to get real-time traces from your hosted agent.
• Set approval_mode="always_require" on tools that have side effects. The framework pauses execution and waits for human approval.
• Build for linux/amd64 when deploying to Foundry from Apple Silicon or let azd handle it with cloud builds.

Get started locally

The example in this article lives here and can be run directly:

python -m venv .venv && source .venv/bin/activate
pip install -r requirements.txt

# Copy the env template and fill in your Foundry project endpoint
cp .env.example .env
# Edit .env with your AZURE_AI_PROJECT_ENDPOINT and AZURE_AI_MODEL_DEPLOYMENT_NAME

# The brain: diamond workflow + tools (localhost:8088)
python -m example.hosted_agents_in_workflow

# The interface: same brain, wrapped in M365 Agents SDK (localhost:3978)
python -m example.m365_vacation_planner

Further reading and upstream references:

• Agent Framework docs: aka.ms/agent-framework
• Hosted agents overview: learn.microsoft.com
• M365 Agents SDK: learn.microsoft.com
• Azure Developer CLI agent extension: azd extension install azure.ai.agents
• Upstream end-to-end samples (the originals that inspired the example in this article): github.com/microsoft/agent-framework/.../05-end-to-end

The upstream repo also contains samples for evaluation and red-teaming, ChatKit integration, AG-UI handoff workflows, Purview policy enforcement, and Neo4j GraphRAG, each worthy of its own deep-dive.