Microsoft Agent Framework : créer des workflows multi-agents en Python

Microsoft a passé 2025 à mener deux récits parallèles autour des agents : Semantic Kernel, le SDK de niveau entreprise doté de télémétrie, de middleware et de typage strict, et AutoGen, le projet de recherche célèbre pour ses abstractions multi-agents légères. Fin 2025, les deux équipes ont fusionné leurs efforts dans une bibliothèque unique — Microsoft Agent Framework — et début 2026 elle a atteint sa version 1.0 pour .NET et Python.

Ce tutoriel parcourt le SDK Python de bout en bout. Vous allez créer un système de tri des tickets de support : un ensemble d'agents qui classent un ticket entrant, rédigent une réponse à l'aide d'outils réels, acheminent les cas difficiles vers un humain, et font passer le tout par un contrôle qualité avant l'envoi. En chemin, vous découvrirez les deux API de workflow fournies par le framework et quand utiliser chacune.

Note sur l'API 2026. La version 1.0 a simplifié les types fondamentaux : ChatAgent est devenu Agent, ChatMessage est devenu Message, run_stream() a été intégré dans run(..., stream=True), et le décorateur @ai_function est devenu @tool. Ce tutoriel utilise partout les nouveaux noms. Si vous lisez d'anciens articles ou dépôts d'exemples, attendez-vous aux noms plus longs Chat*.

Prérequis

Avant de commencer, assurez-vous d'avoir :

• Python 3.10+ installé (python --version)
• Une clé API OpenAI (ou un point de terminaison Azure OpenAI / Microsoft Foundry)
• Une familiarité de base avec async/await en Python
• Un éditeur de code tel que VS Code

Nous utiliserons OpenAI directement car c'est le chemin le plus rapide vers un agent fonctionnel. Tout dans ce tutoriel se transpose proprement vers Azure OpenAI et Microsoft Foundry en remplaçant la classe cliente.

Ce que vous allez créer

Une chaîne de support multi-étapes composée de petits agents à usage unique :

1. Un agent de tri classe le ticket en catégorie et niveau d'urgence.
2. Un agent de support rédige une réponse, en appelant des outils pour consulter l'état de commande et la politique de remboursement.
3. Une porte d'intervention humaine se met en pause pour validation lorsque l'urgence est élevée.
4. Un agent qualité relit le brouillon final pour le ton et l'exactitude avant l'envoi.

Vous implémenterez cela d'abord avec de simples appels d'agents, puis avec l'API de workflow fonctionnelle (@workflow / @step), et enfin avec l'API de workflow graphe (WorkflowBuilder + exécuteurs + arêtes) afin de comprendre les compromis.

Étape 1 : configuration du projet

Créez un dossier de projet et un environnement virtuel, puis installez le framework.

mkdir support-agents && cd support-agents
python -m venv .venv
source .venv/bin/activate   # sous Windows : .venv\Scripts\activate
 
pip install agent-framework python-dotenv

Le paquet parapluie agent-framework intègre par défaut la prise en charge d'OpenAI et d'Azure OpenAI. Pour une installation plus légère, utilisez plutôt pip install agent-framework-core.

Créez un fichier .env pour vos identifiants. Le framework ne charge pas automatiquement les fichiers .env, nous appellerons donc load_dotenv() explicitement.

# .env
OPENAI_API_KEY=sk-...
OPENAI_CHAT_MODEL=gpt-4o-mini

Astuce : le framework lit OPENAI_CHAT_MODEL (et OPENAI_MODEL) depuis l'environnement. Notez la standardisation de 2026 : le paramètre est désormais model, jamais model_id.

Étape 2 : votre premier agent

Créez hello.py. Un agent est la combinaison d'un client de chat (la connexion au modèle) et d'instructions (le prompt système).

# hello.py
import asyncio
from dotenv import load_dotenv
 
from agent_framework import Agent
from agent_framework.openai import OpenAIChatClient
 
load_dotenv()
 
async def main() -> None:
    agent = Agent(
        client=OpenAIChatClient(),
        name="HelloAgent",
        instructions="You are a friendly support assistant. Keep answers brief.",
    )
 
    result = await agent.run("A customer asks: where is my order?")
    print(result)
 
if __name__ == "__main__":
    asyncio.run(main())

Exécutez-le :

python hello.py

L'appel agent.run() renvoie un objet de réponse dont la représentation textuelle est le texte du modèle. Pour recevoir les jetons au fur et à mesure de leur génération, passez stream=True :

print("Agent: ", end="", flush=True)
async for chunk in agent.run("Give me a one-sentence apology template.", stream=True):
    if chunk.text:
        print(chunk.text, end="", flush=True)
print()

La même classe Agent fonctionne avec n'importe quel fournisseur. Pour cibler Azure OpenAI, remplacez l'import par AzureOpenAIChatClient de agent_framework.azure ; pour cibler Microsoft Foundry, utilisez FoundryChatClient. Le code de l'agent ci-dessus ne change pas.

Étape 3 : donnez des outils à l'agent

Les agents deviennent utiles lorsqu'ils peuvent agir. Dans Agent Framework, un outil n'est qu'une fonction Python avec une docstring et des paramètres typés — le framework génère le schéma JSON et gère la boucle d'appel pour vous.

Créez tools.py :

# tools.py
from typing import Annotated
from pydantic import Field
 
def get_order_status(
    order_id: Annotated[str, Field(description="The customer's order ID, e.g. 'A-1042'")]
) -> str:
    """Look up the shipping status of an order by its ID."""
    # En production, ceci interrogerait votre base de commandes ou une API.
    fake_db = {
        "A-1042": "Shipped on 2026-06-14, arriving 2026-06-19.",
        "A-1099": "Processing — not yet shipped.",
    }
    return fake_db.get(order_id, "No order found with that ID.")
 
def get_refund_policy(
    region: Annotated[str, Field(description="Customer region: 'EU', 'US', or 'MENA'")]
) -> str:
    """Return the refund window and conditions for a region."""
    policies = {
        "EU": "14-day no-questions-asked returns under EU consumer law.",
        "US": "30-day returns with receipt.",
        "MENA": "14-day returns for unused items in original packaging.",
    }
    return policies.get(region, "Standard 14-day return policy applies.")

Enregistrez maintenant les outils en les passant à l'agent. C'est le modèle qui décide quand les appeler.

# support_agent.py
import asyncio
from dotenv import load_dotenv
 
from agent_framework import Agent
from agent_framework.openai import OpenAIChatClient
 
from tools import get_order_status, get_refund_policy
 
load_dotenv()
 
async def main() -> None:
    agent = Agent(
        client=OpenAIChatClient(),
        name="SupportAgent",
        instructions=(
            "You are a support agent. Use the tools to look up real order "
            "and policy data before answering. Never invent order details."
        ),
        tools=[get_order_status, get_refund_policy],
    )
 
    result = await agent.run(
        "Hi, where is order A-1042, and can I return it? I'm in the EU."
    )
    print(result)
 
if __name__ == "__main__":
    asyncio.run(main())

L'agent appellera get_order_status("A-1042") et get_refund_policy("EU"), puis combinera les deux résultats en une seule réponse fondée sur des données réelles. Vous n'avez écrit aucun code de gestion d'appel — le framework exécute la boucle d'outils automatiquement.

Variante avec décorateur. Pour les outils nécessitant une configuration ou des métadonnées plus riches, décorez la fonction avec @tool (importé depuis agent_framework). Une fonction simple passée dans la liste tools est traitée comme un outil implicitement ; le décorateur est donc facultatif dans les cas simples.

Étape 4 : mémoire multi-tours avec les sessions

Chaque appel agent.run() est sans état par défaut. Pour tenir une conversation, créez une session (autrefois appelée « thread ») et passez-la à chaque tour.

# session_demo.py
import asyncio
from dotenv import load_dotenv
 
from agent_framework import Agent
from agent_framework.openai import OpenAIChatClient
from tools import get_order_status
 
load_dotenv()
 
async def main() -> None:
    agent = Agent(
        client=OpenAIChatClient(),
        name="SupportAgent",
        instructions="You are a concise support agent.",
        tools=[get_order_status],
    )
 
    session = agent.create_session()
 
    r1 = await agent.run("Where is order A-1042?", session=session)
    print("Turn 1:", r1)
 
    # La relance n'a pas d'ID de commande — la session porte le contexte.
    r2 = await agent.run("And will it arrive before the weekend?", session=session)
    print("Turn 2:", r2)
 
if __name__ == "__main__":
    asyncio.run(main())

Deux renommages à retenir de la version 2026 : get_new_thread() est devenu create_session(), et le paramètre thread= est devenu session=. Si aucun fournisseur d'historique n'est configuré, le framework injecte automatiquement un InMemoryHistoryProvider, de sorte que les conversations fonctionnent d'emblée.

Étape 5 : orchestrer plusieurs agents

Les vrais systèmes utilisent plusieurs spécialistes plutôt qu'un agent qui fait tout. L'orchestration la plus simple est séquentielle : la sortie de l'agent A alimente l'agent B. Vous pouvez le faire à la main — les agents sont des objets Python composables.

# pipeline_manual.py
import asyncio
from dotenv import load_dotenv
 
from agent_framework import Agent
from agent_framework.openai import OpenAIChatClient
from tools import get_order_status, get_refund_policy
 
load_dotenv()
 
def make_agent(name: str, instructions: str, tools=None) -> Agent:
    return Agent(
        client=OpenAIChatClient(),
        name=name,
        instructions=instructions,
        tools=tools or [],
    )
 
async def main() -> None:
    triage = make_agent(
        "Triage",
        "Classify the ticket. Reply with exactly: 'CATEGORY: <billing|shipping|other> | "
        "URGENCY: <low|high>'. Nothing else.",
    )
    support = make_agent(
        "Support",
        "Draft a warm, accurate reply. Use tools for real data.",
        tools=[get_order_status, get_refund_policy],
    )
    qa = make_agent(
        "QA",
        "You review support drafts. If the tone is professional and the facts are "
        "grounded, reply with the draft unchanged. Otherwise rewrite it.",
    )
 
    ticket = "URGENT: order A-1099 still not here and I leave the country tomorrow! EU customer."
 
    label = await triage.run(ticket)
    print("Triage:", label)
 
    draft = await support.run(f"Ticket: {ticket}\nClassification: {label}")
    print("Draft:", draft)
 
    final = await qa.run(f"Review this draft:\n{draft}")
    print("Final:", final)
 
if __name__ == "__main__":
    asyncio.run(main())

Cela fonctionne, mais le flux de contrôle, la gestion des erreurs et l'observabilité reposent entièrement sur vous. C'est exactement ce que résolvent les API de workflow.

Étape 6 : l'API de workflow fonctionnelle

L'API fonctionnelle vous permet d'exprimer un workflow comme une simple fonction async décorée par @workflow, où chaque étape est un @step. Vous conservez le flux de contrôle natif de Python — if/else, boucles, asyncio.gather — tout en gagnant des événements par étape, le streaming, les points de reprise et la prise en charge de l'intervention humaine.

# workflow_functional.py
import asyncio
from dotenv import load_dotenv
 
from agent_framework import Agent, workflow, step
from agent_framework.openai import OpenAIChatClient
from tools import get_order_status, get_refund_policy
 
load_dotenv()
 
client = OpenAIChatClient()
 
triage = Agent(client=client, name="Triage",
               instructions="Reply 'URGENCY: high' or 'URGENCY: low' and one category word.")
support = Agent(client=client, name="Support",
               instructions="Draft an accurate, friendly reply. Use tools for real data.",
               tools=[get_order_status, get_refund_policy])
qa = Agent(client=client, name="QA",
           instructions="Approve or rewrite the draft for tone and factual grounding.")
 
@step
async def classify(ticket: str) -> dict:
    label = str(await triage.run(ticket))
    return {"ticket": ticket, "label": label, "high": "high" in label.lower()}
 
@step
async def draft_reply(state: dict) -> dict:
    draft = str(await support.run(f"Ticket: {state['ticket']}\nLabel: {state['label']}"))
    state["draft"] = draft
    return state
 
@step
async def review(state: dict) -> str:
    return str(await qa.run(f"Review and finalize:\n{state['draft']}"))
 
@workflow
async def support_pipeline(ticket: str, ctx) -> str:
    state = await classify(ticket)
 
    # Branchement Python natif : escalade des tickets urgents vers un humain.
    if state["high"]:
        decision = await ctx.request_info(
            f"High-urgency ticket needs approval before auto-reply:\n{state['label']}"
        )
        if str(decision).strip().lower().startswith("no"):
            return "Escalated to a human agent. No automated reply sent."
 
    state = await draft_reply(state)
    return await review(state)
 
async def main() -> None:
    result = await support_pipeline.run(
        "URGENT: order A-1099 not delivered, leaving tomorrow! EU."
    )
    print(result)
 
if __name__ == "__main__":
    asyncio.run(main())

Quelques points à noter :

• ctx.request_info(...) est la primitive d'intervention humaine. Lors d'une exécution interactive, elle met le workflow en pause et fait remonter un événement de requête auquel votre application répond ; dans les exécutions automatisées, vous branchez un répondeur.
• Chaque @step produit son propre événement, vous obtenez donc une observabilité fine gratuitement.
• Comme c'est du Python ordinaire, vous pouvez utiliser asyncio.gather pour exécuter des étapes indépendantes en parallèle — par exemple, classer et récupérer l'historique du compte en même temps.

L'API fonctionnelle est le point de départ recommandé. Passez à l'API graphe lorsque vous avez besoin d'un routage de messages strict et validé par les types entre de nombreux exécuteurs.

Étape 7 : l'API de workflow graphe

L'API graphe modélise le workflow comme un graphe orienté d'exécuteurs reliés par des arêtes. Elle brille pour les topologies fixes, le parallélisme de type fan-out/fan-in et les points de reprise à la frontière des super-étapes. Vous la construisez avec WorkflowBuilder.

# workflow_graph.py
import asyncio
from dotenv import load_dotenv
 
from agent_framework import (
    Agent, WorkflowBuilder, executor, WorkflowContext,
)
from agent_framework.openai import OpenAIChatClient
from tools import get_order_status, get_refund_policy
 
load_dotenv()
 
client = OpenAIChatClient()
 
support = Agent(client=client, name="Support",
                instructions="Draft an accurate, friendly reply. Use tools for real data.",
                tools=[get_order_status, get_refund_policy])
qa = Agent(client=client, name="QA",
           instructions="Finalize the draft for tone and accuracy.")
 
@executor(id="classify")
async def classify(ticket: str, ctx: WorkflowContext[str]) -> None:
    label = str(await Agent(
        client=client, name="Triage",
        instructions="Reply with one category word and 'high' or 'low' urgency.",
    ).run(ticket))
    # Conserve le ticket d'origine dans l'état partagé pour les exécuteurs suivants.
    ctx.set_state("ticket", ticket)
    await ctx.send_message(label)
 
@executor(id="draft")
async def draft(label: str, ctx: WorkflowContext[str]) -> None:
    ticket = ctx.get_state("ticket")
    reply = str(await support.run(f"Ticket: {ticket}\nLabel: {label}"))
    await ctx.send_message(reply)
 
@executor(id="review")
async def review(reply: str, ctx: WorkflowContext) -> None:
    final = str(await qa.run(f"Finalize:\n{reply}"))
    await ctx.yield_output(final)
 
async def main() -> None:
    workflow = (
        WorkflowBuilder()
        .set_start_executor(classify)
        .add_edge(classify, draft)
        .add_edge(draft, review)
        .build()
    )
 
    async for event in workflow.run("Where is order A-1042? EU customer.", stream=True):
        if event.type == "output":
            print("FINAL:", event.data)
 
if __name__ == "__main__":
    asyncio.run(main())

Concepts clés de l'API graphe :

• Un exécuteur est une unité de travail — un agent ou une logique personnalisée — déclaré avec le décorateur @executor et un gestionnaire qui reçoit un WorkflowContext.
• ctx.send_message(...) transmet un message typé le long des arêtes sortantes ; ctx.yield_output(...) émet un résultat final de workflow.
• ctx.set_state(...) / ctx.get_state(...) lisent et écrivent l'état partagé. Dans la version 2026, ils sont devenus synchrones (sans await), et shared_state a été renommé state.
• add_edge(a, b) relie les exécuteurs. Vous pouvez attacher une fonction de condition à une arête pour un routage basé sur le contenu, et fan-out vers plusieurs exécuteurs pour des super-étapes parallèles.

Notez le modèle d'événements unifié : au lieu de nombreuses sous-classes d'événements, vous vérifiez event.type — "output", "request_info", etc. Pour l'intervention humaine dans l'API graphe, ajoutez un nœud RequestInfoExecutor.

Étape 8 : observabilité

Parce qu'Agent Framework hérite de la lignée entreprise de Semantic Kernel, il émet des traces et des métriques OpenTelemetry d'emblée. Chaque exécution d'agent et chaque exécuteur de workflow devient un span, vous pouvez donc voir la consommation de jetons, les appels d'outils et la latence dans n'importe quel backend compatible OTel.

from agent_framework.observability import setup_observability
 
# À appeler une fois au démarrage. Lit OTEL_EXPORTER_OTLP_ENDPOINT depuis l'environnement.
setup_observability()

Pointez OTEL_EXPORTER_OTLP_ENDPOINT vers un collecteur local (ou Aspire Dashboard / Jaeger) et vous obtenez une trace complète de quel agent a appelé quel outil, avec le chronométrage — inestimable lorsqu'une exécution multi-agents se comporte mal en production.

Tester votre implémentation

Vous ne voulez pas solliciter un modèle payant dans les tests unitaires. Testez directement les parties déterministes — vos outils — et vérifiez la structure du workflow.

# test_tools.py
from tools import get_order_status, get_refund_policy
 
def test_known_order():
    assert "Shipped" in get_order_status("A-1042")
 
def test_unknown_order():
    assert "No order found" in get_order_status("ZZZ")
 
def test_region_policy():
    assert "14-day" in get_refund_policy("EU")
    assert "30-day" in get_refund_policy("US")

Exécutez avec pytest. Pour les tests au niveau de l'agent, le framework fournit des utilitaires de client de test pour simuler les réponses du modèle et vérifier que l'outil attendu a bien été invoqué, sans appel réseau.

Dépannage

ImportError: cannot import name 'ChatAgent' — Vous êtes sur l'API 1.0+. Utilisez Agent, pas ChatAgent. Idem pour ChatMessage (devenu Message).

L'agent n'appelle jamais mon outil — Assurez-vous que la fonction a une docstring et des paramètres typés et Annotated. Le framework construit le schéma de l'outil à partir de ceux-ci ; une description manquante ne donne au modèle rien à quoi se rattacher. Vérifiez aussi que vos instructions demandent explicitement à l'agent d'utiliser les outils.

Les valeurs .env sont ignorées — Agent Framework ne charge pas .env automatiquement. Appelez load_dotenv() avant de construire tout client, ou exportez les variables dans votre shell.

AttributeError sur shared_state / await ctx.get_shared_state — En 2026, ce sont ctx.state, ctx.get_state(...) et ctx.set_state(...), et ils sont synchrones. Supprimez le await.

Erreurs d'authentification avec Azure — DefaultAzureCredential est pratique en local mais sonde de nombreuses sources. En production, préférez un identifiant spécifique comme ManagedIdentityCredential pour éviter la latence et les bascules inattendues.

Étapes suivantes

• Ajoutez un RequestInfoExecutor au workflow graphe pour que les tickets très urgents se mettent en pause pour un humain, à l'image de l'exemple fonctionnel.
• Remplacez OpenAIChatClient par AzureOpenAIChatClient et déployez derrière un point de terminaison Foundry pour la gouvernance d'entreprise.
• Explorez les schémas d'orchestration intégrés concurrent, hand-off et magentic pour un comportement multi-agents plus dynamique.
• Encapsulez un workflow entier en tant qu'agent avec .as_agent() et imbriquez-le dans un système plus vaste.

Si vous construisez plutôt des agents en TypeScript, nos guides sur l'OpenAI Agents SDK et Mastra couvrent les mêmes schémas dans cet écosystème. Pour les alternatives orientées Python, consultez nos tutoriels Pydantic AI et Agno.

Conclusion

Microsoft Agent Framework vous offre une bibliothèque cohérente unique qui passe d'un « hello agent » de trois lignes à un workflow multi-agents validé par les types, observable et avec intervention humaine. Le modèle mental est simple : un agent est un modèle plus des instructions plus des outils ; un workflow est la manière dont vous composez les agents avec un flux de contrôle explicite. Commencez par de simples appels d'agents, passez à l'API fonctionnelle @workflow lorsque vous avez besoin d'orchestration, et adoptez le WorkflowBuilder graphe lorsque vous avez besoin d'un routage strict à grande échelle. Parce qu'il porte la télémétrie de Semantic Kernel et l'ergonomie d'AutoGen, c'est l'un des rares frameworks d'agents conçus pour la production dès le premier jour — un choix par défaut solide pour les équipes de la région MENA qui standardisent sur l'infrastructure Microsoft et Azure.