m365-agents-py

仓库: sickn33/antigravity-awesome-skills
安装量: 35
排名: #19527

安装

npx skills add https://github.com/sickn33/antigravity-awesome-skills --skill m365-agents-py
Microsoft 365 Agents SDK (Python)
Build enterprise agents for Microsoft 365, Teams, and Copilot Studio using the Microsoft Agents SDK with aiohttp hosting, AgentApplication routing, streaming responses, and MSAL-based authentication.
Before implementation
Use the microsoft-docs MCP to verify the latest API signatures for AgentApplication, start_agent_process, and authentication options.
Confirm package versions on PyPI for the microsoft-agents-* packages you plan to use.
Important Notice - Import Changes
⚠️ Breaking Change
Recent updates have changed the Python import structure from microsoft.agents to microsoft_agents (using underscores instead of dots). 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 Environment Variables (.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

Core Workflow: aiohttp-hosted AgentApplication 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

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

AGENT_APP

AgentApplicationTurnState @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 Routing import re from microsoft_agents . hosting . core import ( AgentApplication , TurnState , TurnContext , MessageFactory ) from microsoft_agents . activity import ActivityTypes AGENT_APP = AgentApplicationTurnState

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." ) Streaming Responses with 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 / Auto Sign-In @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 Client (Direct to Engine) 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" ) , ) copilot_client = CopilotClient ( settings , token )

Start conversation

act

copilot_client . start_conversation ( True ) async for action in act : if action . text : print ( action . text )

Ask 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 ( ) ) Best Practices Use microsoft_agents import prefix (underscores, not dots). Use MemoryStorage only for development; use BlobStorage or CosmosDB in production. Always use load_configuration_from_env(environ) to load SDK configuration. Include jwt_authorization_middleware in aiohttp Application middlewares. Use MsalConnectionManager for MSAL-based authentication. Call end_stream() in finally blocks when using streaming responses. Use auth_handlers parameter on message decorators for OAuth-protected routes. Keep secrets in environment variables, not in source code. Reference Files File Contents references/acceptance-criteria.md Import paths, hosting pipeline, streaming, OAuth, and Copilot Studio patterns Reference Links Resource URL Microsoft 365 Agents SDK https://learn.microsoft.com/en-us/microsoft-365/agents-sdk/ GitHub samples (Python) https://github.com/microsoft/Agents-for-python PyPI packages https://pypi.org/search/?q=microsoft-agents Integrate with Copilot Studio https://learn.microsoft.com/en-us/microsoft-365/agents-sdk/integrate-with-mcs When to Use This skill is applicable to execute the workflow or actions described in the overview.

← 返回排行榜