m365-agents-py

npx skills add https://github.com/microsoft/skills --skill m365-agents-py

Kit de développement Microsoft 365 Agents (Python)

Créez des agents d'entreprise pour Microsoft 365, Teams et Copilot Studio à l'aide du Kit de développement Microsoft Agents avec hébergement aiohttp, routage AgentApplication, réponses en streaming et authentification basée sur MSAL.

Avant l'implémentation

  • Utilisez le MCP microsoft-docs pour vérifier les dernières signatures d'API pour AgentApplication, start_agent_process et les options d'authentification.
  • Confirmez les versions des packages sur PyPI pour les packages microsoft-agents-* que vous envisagez d'utiliser.

Avis important - Changements d'importation

⚠️ Changement cassant : Les mises à jour récentes ont changé la structure d'importation Python de microsoft.agents à microsoft_agents (utilisant des traits de soulignement à la place des points).

Installation

pip install microsoft-agents-hosting-core
pip install microsoft-agents-hosting-aiohttp
pip install microsoft-agents-activity
pip install microsoft-agents-authentication-msal
pip install microsoft-agents-copilotstudio-client
pip install python-dotenv aiohttp

Variables d'environnement (.env)

CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTID=<client-id>
CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTSECRET=<client-secret>
CONNECTIONS__SERVICE_CONNECTION__SETTINGS__TENANTID=<tenant-id>

# Facultatif : gestionnaires OAuth pour la connexion automatique
AGENTAPPLICATION__USERAUTHORIZATION__HANDLERS__GRAPH__SETTINGS__AZUREBOTOAUTHCONNECTIONNAME=<connection-name>

# Facultatif : Azure OpenAI pour le streaming
AZURE_OPENAI_ENDPOINT=<endpoint>
AZURE_OPENAI_API_VERSION=<version>
AZURE_OPENAI_API_KEY=<key>

# Facultatif : client Copilot Studio
COPILOTSTUDIOAGENT__ENVIRONMENTID=<environment-id>
COPILOTSTUDIOAGENT__SCHEMANAME=<schema-name>
COPILOTSTUDIOAGENT__TENANTID=<tenant-id>
COPILOTSTUDIOAGENT__AGENTAPPID=<app-id>

Flux de travail principal : AgentApplication hébergée avec aiohttp

import logging
from os import environ

from dotenv import load_dotenv
from aiohttp.web import Request, Response, Application, run_app

from microsoft_agents.activity import load_configuration_from_env
from microsoft_agents.hosting.core import (
    Authorization,
    AgentApplication,
    TurnState,
    TurnContext,
    MemoryStorage,
)
from microsoft_agents.hosting.aiohttp import (
    CloudAdapter,
    start_agent_process,
    jwt_authorization_middleware,
)
from microsoft_agents.authentication.msal import MsalConnectionManager

# Activer la journalisation
ms_agents_logger = logging.getLogger("microsoft_agents")
ms_agents_logger.addHandler(logging.StreamHandler())
ms_agents_logger.setLevel(logging.INFO)

# Charger la configuration
load_dotenv()
agents_sdk_config = load_configuration_from_env(environ)

# Créer le stockage et le gestionnaire de connexions
STORAGE = MemoryStorage()
CONNECTION_MANAGER = MsalConnectionManager(**agents_sdk_config)
ADAPTER = CloudAdapter(connection_manager=CONNECTION_MANAGER)
AUTHORIZATION = Authorization(STORAGE, CONNECTION_MANAGER, **agents_sdk_config)

# Créer AgentApplication
AGENT_APP = AgentApplication[TurnState](
    storage=STORAGE, adapter=ADAPTER, authorization=AUTHORIZATION, **agents_sdk_config
)


@AGENT_APP.conversation_update("membersAdded")
async def on_members_added(context: TurnContext, _state: TurnState):
    await context.send_activity("Welcome to the agent!")


@AGENT_APP.activity("message")
async def on_message(context: TurnContext, _state: TurnState):
    await context.send_activity(f"You said: {context.activity.text}")


@AGENT_APP.error
async def on_error(context: TurnContext, error: Exception):
    await context.send_activity("The agent encountered an error.")


# Configuration du serveur
async def entry_point(req: Request) -> Response:
    agent: AgentApplication = req.app["agent_app"]
    adapter: CloudAdapter = req.app["adapter"]
    return await start_agent_process(req, agent, adapter)


APP = Application(middlewares=[jwt_authorization_middleware])
APP.router.add_post("/api/messages", entry_point)
APP["agent_configuration"] = CONNECTION_MANAGER.get_default_connection_configuration()
APP["agent_app"] = AGENT_APP
APP["adapter"] = AGENT_APP.adapter

if __name__ == "__main__":
    run_app(APP, host="localhost", port=environ.get("PORT", 3978))

Routage AgentApplication

import re
from microsoft_agents.hosting.core import (
    AgentApplication, TurnState, TurnContext, MessageFactory
)
from microsoft_agents.activity import ActivityTypes

AGENT_APP = AgentApplication[TurnState](
    storage=STORAGE, adapter=ADAPTER, authorization=AUTHORIZATION, **agents_sdk_config
)

# Gestionnaire de bienvenue
@AGENT_APP.conversation_update("membersAdded")
async def on_members_added(context: TurnContext, _state: TurnState):
    await context.send_activity("Welcome!")

# Gestionnaire de messages basé sur regex
@AGENT_APP.message(re.compile(r"^hello$", re.IGNORECASE))
async def on_hello(context: TurnContext, _state: TurnState):
    await context.send_activity("Hello!")

# Gestionnaire de messages avec chaîne simple
@AGENT_APP.message("/status")
async def on_status(context: TurnContext, _state: TurnState):
    await context.send_activity("Status: OK")

# Gestionnaire de messages protégé par authentification
@AGENT_APP.message("/me", auth_handlers=["GRAPH"])
async def on_profile(context: TurnContext, state: TurnState):
    token_response = await AGENT_APP.auth.get_token(context, "GRAPH")
    if token_response and token_response.token:
        # Utiliser le token pour appeler l'API Graph
        await context.send_activity("Profile retrieved")

# Gestionnaire d'activité invoke
@AGENT_APP.activity(ActivityTypes.invoke)
async def on_invoke(context: TurnContext, _state: TurnState):
    invoke_response = Activity(
        type=ActivityTypes.invoke_response, value={"status": 200}
    )
    await context.send_activity(invoke_response)

# Gestionnaire de messages par défaut
@AGENT_APP.activity("message")
async def on_message(context: TurnContext, _state: TurnState):
    await context.send_activity(f"Echo: {context.activity.text}")

# Gestionnaire d'erreurs
@AGENT_APP.error
async def on_error(context: TurnContext, error: Exception):
    await context.send_activity("An error occurred.")

Réponses en streaming avec Azure OpenAI

from openai import AsyncAzureOpenAI
from microsoft_agents.activity import SensitivityUsageInfo

CLIENT = AsyncAzureOpenAI(
    api_version=environ["AZURE_OPENAI_API_VERSION"],
    azure_endpoint=environ["AZURE_OPENAI_ENDPOINT"],
    api_key=environ["AZURE_OPENAI_API_KEY"]
)

@AGENT_APP.message("poem")
async def on_poem_message(context: TurnContext, _state: TurnState):
    # Configurer la réponse en streaming
    context.streaming_response.set_feedback_loop(True)
    context.streaming_response.set_generated_by_ai_label(True)
    context.streaming_response.set_sensitivity_label(
        SensitivityUsageInfo(
            type="https://schema.org/Message",
            schema_type="CreativeWork",
            name="Internal",
        )
    )
    context.streaming_response.queue_informative_update("Starting a poem...\n")

    # Streamer depuis Azure OpenAI
    streamed_response = await CLIENT.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "You are a creative assistant."},
            {"role": "user", "content": "Write a poem about Python."}
        ],
        stream=True,
    )

    try:
        async for chunk in streamed_response:
            if chunk.choices and chunk.choices[0].delta.content:
                context.streaming_response.queue_text_chunk(
                    chunk.choices[0].delta.content
                )
    finally:
        await context.streaming_response.end_stream()

OAuth / Connexion automatique

@AGENT_APP.message("/logout")
async def logout(context: TurnContext, state: TurnState):
    await AGENT_APP.auth.sign_out(context, "GRAPH")
    await context.send_activity(MessageFactory.text("You have been logged out."))


@AGENT_APP.message("/me", auth_handlers=["GRAPH"])
async def profile_request(context: TurnContext, state: TurnState):
    user_token_response = await AGENT_APP.auth.get_token(context, "GRAPH")
    if user_token_response and user_token_response.token:
        # Utiliser le token pour appeler Microsoft Graph
        async with aiohttp.ClientSession() as session:
            headers = {
                "Authorization": f"Bearer {user_token_response.token}",
                "Content-Type": "application/json",
            }
            async with session.get(
                "https://graph.microsoft.com/v1.0/me", headers=headers
            ) as response:
                if response.status == 200:
                    user_info = await response.json()
                    await context.send_activity(f"Hello, {user_info['displayName']}!")

Client Copilot Studio (Directement au moteur)

import asyncio
from msal import PublicClientApplication
from microsoft_agents.activity import ActivityTypes, load_configuration_from_env
from microsoft_agents.copilotstudio.client import (
    ConnectionSettings,
    CopilotClient,
)

# Cache de tokens (fichier local pour les flux interactifs)
class LocalTokenCache:
    # Voir les exemples pour l'implémentation complète
    pass

def acquire_token(settings, app_client_id, tenant_id):
    pca = PublicClientApplication(
        client_id=app_client_id,
        authority=f"https://login.microsoftonline.com/{tenant_id}",
    )

    token_request = {"scopes": ["https://api.powerplatform.com/.default"]}
    accounts = pca.get_accounts()

    if accounts:
        response = pca.acquire_token_silent(token_request["scopes"], account=accounts[0])
        return response.get("access_token")
    else:
        response = pca.acquire_token_interactive(**token_request)
        return response.get("access_token")


async def main():
    settings = ConnectionSettings(
        environment_id=environ.get("COPILOTSTUDIOAGENT__ENVIRONMENTID"),
        agent_identifier=environ.get("COPILOTSTUDIOAGENT__SCHEMANAME"),
    )

    token = acquire_token(
        settings,
        app_client_id=environ.get("COPILOTSTUDIOAGENT__AGENTAPPID"),
        tenant_id=environ.get("COPILOTSTUDIOAGENT__TENANTID"),
    )

    copilot_client = CopilotClient(settings, token)

    # Démarrer la conversation
    act = copilot_client.start_conversation(True)
    async for action in act:
        if action.text:
            print(action.text)

    # Poser une question
    replies = copilot_client.ask_question("Hello!", action.conversation.id)
    async for reply in replies:
        if reply.type == ActivityTypes.message:
            print(reply.text)


asyncio.run(main())

Bonnes pratiques

  1. Utilisez le préfixe d'importation microsoft_agents (traits de soulignement, pas des points).
  2. Utilisez MemoryStorage uniquement pour le développement ; utilisez BlobStorage ou CosmosDB en production.
  3. Utilisez toujours load_configuration_from_env(environ) pour charger la configuration du SDK.
  4. Incluez jwt_authorization_middleware dans les middlewares de l'application aiohttp.
  5. Utilisez MsalConnectionManager pour l'authentification basée sur MSAL.
  6. Appelez end_stream() dans les blocs finally lors de l'utilisation de réponses en streaming.
  7. Utilisez le paramètre auth_handlers sur les décorateurs de messages pour les routes protégées par OAuth.
  8. Gardez les secrets dans les variables d'environnement, pas dans le code source.

Liens de référence

Ressource URL
Kit de développement Microsoft 365 Agents https://learn.microsoft.com/en-us/microsoft-365/agents-sdk/
Exemples GitHub (Python) https://github.com/microsoft/Agents-for-python
Packages PyPI https://pypi.org/search/?q=microsoft-agents
Intégrer avec Copilot Studio https://learn.microsoft.com/en-us/microsoft-365/agents-sdk/integrate-with-mcs

Skills similaires

update-avm-modules-in-bicep
github / awesome-copilot

31 949
entra-agent-user
github / awesome-copilot

31 949
azure-devops-cli
github / awesome-copilot

31 949
continual-learning
microsoft / skills

Implémenter un apprentissage continu entre sessions pour agents IA codeurs.

2 191
azure-upgrade
microsoft / skills

Évaluer et automatiser la mise à niveau de workloads Azure vers des plans ou SKU cibles.

2 191