m365-agents-py

CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTID= 
CONNECTIONS__SERVICE_CONNECTION__SETTINGS__CLIENTSECRET= 
CONNECTIONS__SERVICE_CONNECTION__SETTINGS__TENANTID= 

# Optional: OAuth handlers for auto sign-in
AGENTAPPLICATION__USERAUTHORIZATION__HANDLERS__GRAPH__SETTINGS__AZUREBOTOAUTHCONNECTIONNAME= 

# Optional: Azure OpenAI for streaming (AAD auth via DefaultAzureCredential)
AZURE_OPENAI_ENDPOINT= 
AZURE_OPENAI_API_VERSION= 

# Optional: Copilot Studio client
COPILOTSTUDIOAGENT__ENVIRONMENTID= 
COPILOTSTUDIOAGENT__SCHEMANAME= 
COPILOTSTUDIOAGENT__TENANTID= 
COPILOTSTUDIOAGENT__AGENTAPPID= 

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 = 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.")

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

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
)

# 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.")

from openai import AsyncAzureOpenAI
from azure.identity import DefaultAzureCredential, get_bearer_token_provider
from microsoft_agents.activity import SensitivityUsageInfo

# AAD token provider (preferred over AZURE_OPENAI_API_KEY)
token_provider = get_bearer_token_provider(
 DefaultAzureCredential(),
 "https://cognitiveservices.azure.com/.default",
)

# Module-level singleton: client lives for the agent app lifetime.
CLIENT = AsyncAzureOpenAI(
 api_version=environ["AZURE_OPENAI_API_VERSION"],
 azure_endpoint=environ["AZURE_OPENAI_ENDPOINT"],
 azure_ad_token_provider=token_provider,
)

@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()

@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']}!")

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"),
 )

 # CopilotClient does not implement the context manager protocol.
 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())