Skip to main content 首页 创作者 beko2210 firstbrain m365-agents-py
m365-agents-py Microsoft 365 Agents SDK for Python. Build multichannel agents for Teams/M365/Copilot Studio with aiohttp hosting, AgentApplication routing, streaming responses, and MSAL-based auth.
跳到安装 Skills Marketplace 发现并探索由社区构建的 Agent Skills
用 Codex 或 Claude 帮你安装 复制这段 Prompt,粘贴到 Codex、Claude 或其他助手里,让它检查 Skill 页面并帮你完成安装。
直接命令不会经过审查 Prompt;运行前请先检查来源。
npx skills add https://github.com/BEKO2210/Firstbrain --skill m365-agents-py命令会保持在同一行。复制前请横向滚动并检查完整内容。
想先保存到本地?可下载 SkillsMP 当前能够提供的文件。
下载 Zip 下载中... name m365-agents-py description Microsoft 365 Agents SDK for Python. Build multichannel agents for Teams/M365/Copilot Studio with aiohttp hosting, AgentApplication routing, streaming responses, and MSAL-based auth. type skill created 2026-02-27T00:00:00.000Z domain productivity category developer-experience risk unknown source community tags ["skill","productivity","developer-experience","m365","agents"]
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>
AGENTAPPLICATION__USERAUTHORIZATION__HANDLERS__GRAPH__SETTINGS__AZUREBOTOAUTHCONNECTIONNAME=<connection-name>
AZURE_OPENAI_ENDPOINT=<endpoint>
AZURE_OPENAI_API_VERSION=<version>
AZURE_OPENAI_API_KEY=<key>
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,
)
microsoft_agents.hosting.aiohttp (
CloudAdapter,
start_agent_process,
jwt_authorization_middleware,
)
microsoft_agents.authentication.msal MsalConnectionManager
ms_agents_logger = logging.getLogger( )
ms_agents_logger.addHandler(logging.StreamHandler())
ms_agents_logger.setLevel(logging.INFO)
load_dotenv()
agents_sdk_config = load_configuration_from_env(environ)
STORAGE = MemoryStorage()
CONNECTION_MANAGER = MsalConnectionManager(**agents_sdk_config)
ADAPTER = CloudAdapter(connection_manager=CONNECTION_MANAGER)
AUTHORIZATION = Authorization(STORAGE, CONNECTION_MANAGER, **agents_sdk_config)
AGENT_APP = AgentApplicationTurnState
( ):
context.send_activity( )
( ):
context.send_activity( )
( ):
context.send_activity( )
( ) -> Response:
agent: AgentApplication = req.app[ ]
adapter: CloudAdapter = req.app[ ]
start_agent_process(req, agent, adapter)
APP = Application(middlewares=[jwt_authorization_middleware])
APP.router.add_post( , entry_point)
APP[ ] = CONNECTION_MANAGER.get_default_connection_configuration()
APP[ ] = AGENT_APP
APP[ ] = AGENT_APP.adapter
__name__ == :
run_app(APP, host= , port=environ.get( , ))
from
import
from
import
"microsoft_agents"
@AGENT_APP.conversation_update("membersAdded" )
async
def
on_members_added
context: TurnContext, _state: TurnState
await
"Welcome to the agent!"
@AGENT_APP.activity("message" )
async
def
on_message
context: TurnContext, _state: TurnState
await
f"You said: {context.activity.text} "
@AGENT_APP.error
async
def
on_error
context: TurnContext, error: Exception
await
"The agent encountered an error."
async
def
entry_point
req: Request
"agent_app"
"adapter"
return
await
"/api/messages"
"agent_configuration"
"agent_app"
"adapter"
if
"__main__"
"localhost"
"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
@AGENT_APP.conversation_update("membersAdded" )
async def on_members_added (context: TurnContext, _state: TurnState ):
await context.send_activity("Welcome!" )
@AGENT_APP.message(re.compile (r"^hello$" , re.IGNORECASE ) )
async def on_hello (context: TurnContext, _state: TurnState ):
await context.send_activity("Hello!" )
@AGENT_APP.message("/status" )
async def on_status (context: TurnContext, _state: TurnState ):
await context.send_activity("Status: OK" )
@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:
await context.send_activity("Profile retrieved" )
@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)
@AGENT_APP.activity("message" )
async def on_message (context: TurnContext, _state: TurnState ):
await context.send_activity(f"Echo: {context.activity.text} " )
@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 ):
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" )
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:
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,
)
class LocalTokenCache :
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)
act = copilot_client.start_conversation(True )
async for action in act:
if action.text:
print (action.text)
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
When to Use This skill is applicable to execute the workflow or actions described in the overview.
Connections
Domain: [[Produktivitaet & Werkzeuge]]
Kategorie: [[Developer Experience]]
Navigation: [[Skills Uebersicht]], [[Home]]