m365-agents-py

補足資料の入手可能性

このカタログのコピーには以下のインラインガイダンスが含まれていますが、次の補足パスは含まれていません。これらは外部で必要となるものであり、同梱ファイルではありません。

• <support-root>/references/acceptance-criteria.md

これらのパスに依存する手順を実行する前に、元の補足資料または既存のインストール環境を求めてください。このカタログには、検証済みのダウンロード元が記載されていません。各パスをそのインストール環境を基準に解決し、内容を確認したうえで、該当するライセンスと実行の前提条件を確認してください。必要なファイルが存在することを確認する前に、コマンドを実行したり、参照先の手順に従ったと主張したりしないでください。不足しているスクリプトを捏造したり、同じ名前の無関係なファイルを代替として扱ったりしないでください。ここに完全に記載されている独立した手順は、ユーザーが許可した範囲内で実行できます。依存する手順は利用できないものとして報告してください。

Microsoft 365 Agents SDK（Python）

aiohttp ホスティング、AgentApplication ルーティング、ストリーミング応答、MSAL ベースの認証を備えた Microsoft Agents SDK を使用して、Microsoft 365、Teams、Copilot Studio 向けのエンタープライズエージェントを構築します。

実装前

• microsoft-docs MCP を使用して、AgentApplication、start_agent_process、認証オプションの最新の API シグネチャを確認してください。
• 使用する予定の microsoft-agents-* パッケージの PyPI 上のバージョンを確認してください。

重要なお知らせ - インポートの変更

⚠️ 破壊的変更：最近の更新により、Python のインポート構造が microsoft.agents から microsoft_agents（ドットではなくアンダースコアを使用）に変更されました。

インストール

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

環境変数（.env）

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

# Optional: OAuth handlers for auto sign-in
AGENTAPPLICATION__USERAUTHORIZATION__HANDLERS__GRAPH__SETTINGS__AZUREBOTOAUTHCONNECTIONNAME=<connection-name>

# Optional: Azure OpenAI for streaming
AZURE_OPENAI_ENDPOINT=<endpoint>
AZURE_OPENAI_API_VERSION=<version>
AZURE_OPENAI_API_KEY=<key>

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

スニペットを使用するには、バージョンが固定された SDK と、各セクションに示されたセットアップ／インポートが必要です。重複する /me ハンドラーは選択肢であり、同時に登録するものではありません。合成メッセージと許可された ID を使用し、ミドルウェアによる検証を維持してください。また、ログやクラウドトレーシングを通じて、シークレット、非公開の会話ターン、トークンを公開しないでください。Azure OpenAI のモデルフィールドには、想定上の公開モデル ID ではなく、実際のデプロイ名を設定してください。

コアワークフロー：aiohttp ホスト型 AgentApplication

import logging
from os import environ

from dotenv import load_dotenv
import aiohttp
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

# Enable logging
ms_agents_logger = logging.getLogger("microsoft_agents")
ms_agents_logger.addHandler(logging.StreamHandler())
ms_agents_logger.setLevel(logging.INFO)

# Load configuration
load_dotenv()
agents_sdk_config = load_configuration_from_env(environ)

# Create storage and connection manager
STORAGE = MemoryStorage()
CONNECTION_MANAGER = MsalConnectionManager(**agents_sdk_config)
ADAPTER = CloudAdapter(connection_manager=CONNECTION_MANAGER)
AUTHORIZATION = Authorization(STORAGE, CONNECTION_MANAGER, **agents_sdk_config)

# Create AgentApplication
ApplicationType = AgentApplication[TurnState]
AGENT_APP = ApplicationType(
    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.")


# Server setup
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))

AgentApplication ルーティング

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

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

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

# Regex-based message handler
@AGENT_APP.message(re.compile(r"^hello$", re.IGNORECASE))
async def on_hello(context: TurnContext, _state: TurnState):
    await context.send_activity("Hello!")

# Simple string message handler
@AGENT_APP.message("/status")
async def on_status(context: TurnContext, _state: TurnState):
    await context.send_activity("Status: OK")

# Auth-protected message handler
@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:
        # Use token to call Graph API
        await context.send_activity("Profile retrieved")

# Invoke activity handler
@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)

# Fallback message handler
@AGENT_APP.activity("message")
async def on_message(context: TurnContext, _state: TurnState):
    await context.send_activity(f"Echo: {context.activity.text}")

# Error handler
@AGENT_APP.error
async def on_error(context: TurnContext, error: Exception):
    await context.send_activity("An error occurred.")

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):
    # Configure streaming response
    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")

    # Stream from 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／自動サインイン

@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:
        # Use token to call 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']}!")

Copilot Studio クライアント（エンジンに直接接続）

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

# Token cache (local file for interactive flows)
class LocalTokenCache:
    # See samples for full implementation
    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"),
    )
    
    if not token:
        raise RuntimeError("Token acquisition failed")
    copilot_client = CopilotClient(settings, token)
    
    # Start conversation
    conversation_id = None
    act = copilot_client.start_conversation(True)
    async for action in act:
        if action.conversation:
            conversation_id = action.conversation.id
        if action.text:
            print(action.text)
    
    # Ask question
    if not conversation_id:
        raise RuntimeError("No conversation ID returned")
    replies = copilot_client.ask_question("Hello!", conversation_id)
    async for reply in replies:
        if reply.type == ActivityTypes.message:
            print(reply.text)


asyncio.run(main())

ベストプラクティス

1. microsoft_agents のインポートプレフィックスを使用してください（ドットではなくアンダースコア）。
2. 開発時に限り MemoryStorage を使用してください。本番環境では BlobStorage または CosmosDB を使用してください。
3. SDK の設定を読み込むには、必ず load_configuration_from_env(environ) を使用してください。
4. aiohttp Application のミドルウェアに jwt_authorization_middleware を含めてください。
5. MSAL ベースの認証には MsalConnectionManager を使用してください。
6. ストリーミング応答を使用する場合は、finally ブロック内で end_stream() を呼び出してください。
7. OAuth で保護されたルートには、メッセージデコレーターの auth_handlers パラメーターを使用してください。
8. シークレットはソースコードではなく、環境変数に保存してください。

参照ファイル

ファイル	内容
references/acceptance-criteria.md	インポートパス、ホスティングパイプライン、ストリーミング、OAuth、Copilot Studio のパターン

参照リンク

リソース	URL
Microsoft 365 Agents SDK	https://learn.microsoft.com/en-us/microsoft-365/agents-sdk/
GitHub のサンプル（Python）	https://github.com/microsoft/Agents-for-python
PyPI パッケージ	https://pypi.org/search/?q=microsoft-agents
Copilot Studio との統合	https://learn.microsoft.com/en-us/microsoft-365/agents-sdk/integrate-with-mcs

使用する場面

このスキルは、概要に記載されたワークフローやアクションを実行する場合に適用できます。

制限事項

• タスクが上記の説明範囲に明確に一致する場合に限り、このスキルを使用してください。
• 出力を、環境固有の検証、テスト、専門家によるレビューの代わりとして扱わないでください。
• 必要な入力、権限、安全上の境界、または成功基準が不足している場合は、作業を中止して確認を求めてください。