通知智能体

通过使用“通知”模块，您可以构建能够响应来自 Microsoft 365 应用程序的事件和通知的智能体。 借助通知支持，当用户通过电子邮件、文档评论或其他协作场景与智能体交互时，智能体可以接收并处理警报。

通知工作流

请按照以下工作流为您的 AI 智能体应用程序启用通知：

  1. 安装通知软件包。

  2. 导入通知组件

    • 导入通知类和处理程序。
    • 导入活动类型和通道标识符。

  3. 注册通知处理程序

    • 使用通知处理程序方法注册路由。
    • 为特定通知类型（如电子邮件、Word、Excel 或 PowerPoint）配置处理程序。

  4. 在智能体代码中处理通知

    • 智能体会接收来自 Microsoft 365 应用程序的通知。
    • 处理传入的通知并做出适当响应。

通知类型

Agent 365 SDK 支持以下通知类型：

通知类型 描述 子频道 ID
电子邮件 智能体收到一封提及或收件人为其的电子邮件 email
Word 智能体在 Word 文档的评论中被提及 word
Excel Excel 文档中的注释中提到智能体 excel
PowerPoint 智能体在 PowerPoint 文档的评论中被提及 powerpoint
生命周期事件 智能体生命周期通知（用户身份创建、工作负载入职、用户删除） N/A

智能体生命周期事件

智能体生命周期事件使您的智能体能够响应与智能体用户身份管理相关的特定系统事件。 该 SDK 目前支持三种生命周期事件：

事件类型 事件 ID 描述
用户身份创建 agenticUserIdentityCreated 当创建智能体用户身份时触发
工作负载入职更新 agenticUserWorkloadOnboardingUpdated 当智能体用户的负载接入状态更新时触发
用户已删除 agenticUserDeleted 当智能体用户身份被删除时触发

通过使用这些事件，智能体可以针对用户生命周期变化执行初始化任务、清理操作或状态管理。

通知有效负载引用

当您的智能体收到通知时，有效载荷中包含特定于该通知类型的结构化数据。 了解这些有效载荷有助于您提取所需信息，从而有效处理通知。

电子邮件通知有效载荷

当用户向您的智能体发送电子邮件或在电子邮件中提及您的智能体时，您的智能体会收到结构如下所示的电子邮件通知：

{
  "id": "aaaaaaaa-0000-1111-2222-bbbbbbbbbbbb",
  "timestamp": "2026-02-06T17:45:20.740Z",
  "channelId": "agents",
  "serviceUrl": "http://localhost:56150/_connector",
  "recipient": {
    "id": "AgentName@contoso.onmicrosoft.com",
    "name": "My Agent",
    "agenticUserId": "<agentic-user-id>",
    "agenticAppId": "<agentic-app-id>",
    "tenantId": "<tenant-id>",
    "role": "agenticUser"
  },
  "conversation": {
    "id": "<conversation-id>",
    "conversationType": "personal",
    "tenantId": "<tenant-id>"
  },
  "from": {
    "id": "sender@contoso.onmicrosoft.com",
    "name": "Sender Name",
    "role": "user"
  },
  "type": "message",
  "channelData": {
    "tenant": {
      "id": "<tenant-id>"
    }
  },
  "locale": "en-US",
  "name": "emailNotification",
  "entities": [
    {
      "type": "clientInfo",
      "locale": "en-US",
      "timezone": null
    },
    {
      "id": "email",
      "type": "productInfo"
    },
    {
      "type": "emailNotification",
      "id": "<email-id>",
      "conversationId": "<conversation-id>",
      "htmlBody": "<body dir=\"ltr\">\n<div class=\"elementToProof\">Your email message content here</div>\n</body>"
    }
  ]
}

文档注释通知有效负载（Word、Excel、PowerPoint）

当用户在 Word、Excel 或 PowerPoint 文档的评论中提及您的智能体时，您的智能体会收到 WPX（Word、PowerPoint、Excel）评论通知：

{
  "id": "bbbbbbbb-1111-2222-3333-cccccccccccc",
  "timestamp": "2026-02-06T17:46:02.248Z",
  "channelId": "agents",
  "serviceUrl": "http://localhost:56150/_connector",
  "recipient": {
    "id": "AgentName@contoso.onmicrosoft.com",
    "name": "My Agent",
    "agenticUserId": "<agentic-user-id>",
    "agenticAppId": "<agentic-app-id>",
    "tenantId": "<tenant-id>",
    "role": "agenticUser"
  },
  "conversation": {
    "id": "<conversation-id>",
    "conversationType": "personal",
    "tenantId": "<tenant-id>",
    "topic": "<document-topic>"
  },
  "from": {
    "id": "sender@contoso.onmicrosoft.com",
    "name": "Sender Name",
    "role": "user"
  },
  "type": "message",
  "channelData": {
    "tenant": {
      "id": "<tenant-id>"
    },
    "productContext": "Word"
  },
  "locale": "en-US",
  "textFormat": "plain",
  "text": "<at>My Agent</at> - Please review this section\n",
  "attachments": [
    {
      "contentUrl": "<document-url>",
      "name": "<document-name>",
      "content": {
        "uniqueId": "<document-unique-id>",
        "fileType": "docx"
      },
      "contentType": "application/vnd.microsoft.teams.file.download.info"
    }
  ],
  "entities": [
    {
      "type": "clientInfo",
      "locale": "en-US",
      "timezone": null
    },
    {
      "mentioned": {
        "id": "AgentName@contoso.onmicrosoft.com",
        "name": "@My Agent"
      },
      "text": "<at>My Agent</at>",
      "type": "mention"
    },
    {
      "id": "Word",
      "type": "productInfo"
    },
    {
      "parentCommentId": "<parent-comment-id>",
      "commentId": "<comment-id>",
      "documentId": "<document-id>",
      "type": "wpxcomment"
    }
  ]
}

向您的智能体添加通知

请按照以下步骤在现有智能体中启用通知处理：

导入通知组件

将以下导入语句添加到您的智能体文件中：

from microsoft_agents_a365 import AgentApplication
from microsoft_agents_a365.notifications import (
    AgentNotification,
    AgentNotificationActivity,
    NotificationTypes
)
from microsoft_agents.activity import ChannelId
from microsoft_agents.hosting.core import Authorization, TurnContext
  • AgentApplication：用于构建 Agent365 应用程序的基类。 它提供了路由活动、管理状态和处理请求的核心功能。
  • AgentNotification：用于通过装饰器方法注册通知处理程序的类。 它提供了 on_agent_notification()、on_email()、on_word() 及其他便捷装饰器。
  • AgentNotificationActivity：包含已解析通知数据的封装器，其中包含 email_notification 和 wpx_comment_notification 等类型化属性，这些属性包含通知特有的元数据，例如 ID、对话详细信息和文档引用。
  • NotificationTypes：支持的通知类型的枚举，例如 EMAIL_NOTIFICATION、WPX_COMMENT。
  • ChannelId：用于指定通知通道，例如 ChannelId(channel="agents", sub_channel="*")。
  • Authorization：用于处理通知的授权上下文。
  • TurnContext：来自 Agents SDK 的当前对话轮次上下文。

在智能体中注册通知处理程序

在智能体的初始化中添加通知处理程序：

class YourAgent(AgentApplication):
    def __init__(self, app):
        # Create notification handler
        agent_notification = AgentNotification(app)
        
        # Register handler for all notifications
        @agent_notification.on_agent_notification(
            ChannelId(channel="agents", sub_channel="*")
        )
        async def handle_all_notifications(context, state, notification):
            # Route based on notification type
            if notification.notification_type == NotificationTypes.EMAIL_NOTIFICATION:
                await self.handle_email_notification(context, state, notification)
            elif notification.notification_type == NotificationTypes.WPX_COMMENT:
                await self.handle_comment_notification(context, state, notification)
            else:
                await context.send_activity('Notification type not yet implemented.')

实现特定的通知处理程序

为每种通知类型添加处理程序方法：

class YourAgent(AgentApplication):
    # ... __init__ from above ...
    
    async def handle_email_notification(self, context, state, notification):
        """Handle email notifications"""
        email = notification.email_notification
        
        if not email:
            await context.send_activity('No email data found')
            return
        
        # Process the email
        await context.send_activity(
            f'Received email notification. Email ID: {email.id}'
        )
        
        # Your email processing logic here
    
    async def handle_comment_notification(self, context, state, notification):
        """Handle document comment notifications"""
        comment = notification.wpx_comment_notification
        
        if not comment:
            await context.send_activity('No comment data found')
            return
        
        # Process the comment
        await context.send_activity(
            f'Received comment notification. Document ID: {comment.document_id}'
        )
        
        # Your comment processing logic here

识别发送者

每个通知活动都包含 Activity.From。 A365 平台会将发送者的基本身份信息填充到此属性中，因此您无需进行任何 API 调用或令牌获取。 在任何通知处理程序中均可访问该属性：

async def handle_email_notification(self, context, state, notification):
    from_prop = context.activity.from_property
    logger.info(
        "Notification from — DisplayName: '%s', UserId: '%s', AadObjectId: '%s'",
        getattr(from_prop, "name", None) or "(unknown)",
        getattr(from_prop, "id", None) or "(unknown)",
        getattr(from_prop, "aad_object_id", None) or "(none)",
    )
    display_name = getattr(from_prop, "name", None) or "unknown"
    # Use display_name in your response or LLM prompt

Activity.from_property 是 ChannelAccount 类 的实例，具有以下属性：

属性 Description
name 显示名称
id 频道用户 ID
aad_object_id Entra 对象 ID

重要提示

显示名称由用户自行控制。 在将其注入 LLM 系统提示词之前，请对其进行清理（去除控制字符、强制执行最大长度），以防止提示词注入攻击。

提示

当您的智能体具有相应权限时，请使用 aadObjectId 配合 Microsoft 图形 API 检索扩展个人资料数据（职位、经理、部门）。

专用通知处理程序

在设置 基本通知路由 之后，请使用专用处理程序方法进行更精细的控制。 通过使用这些方法，您可以：

  • 为同一通知类型注册多个处理程序。
  • 使用排序设置处理程序的优先级。
  • 为每个处理程序配置自动身份验证。

备注

对于大多数用例，通用处理程序模式已足够。 当您需要高级路由或为同一通知类型配置多个处理程序时，请使用这些专用处理程序。

适用于所有通知的专用处理程序

注册更多可处理所有通知类型的处理程序：

from microsoft_agents_a365.notifications import (
    AgentNotification,
    NotificationTypes
)
from microsoft_agents.activity import ChannelId

# Create notification handler
agent_notification = AgentNotification(app)

# Register handler for all notifications
@agent_notification.on_agent_notification(
    ChannelId(channel="agents", sub_channel="*")
)
async def handle_all_notifications(context, state, notification):
    if notification.notification_type == NotificationTypes.EMAIL_NOTIFICATION:
        if notification.email_notification:
            await context.send_activity(f"Received email: {notification.email_notification.id}")
    elif notification.notification_type == NotificationTypes.WPX_COMMENT:
        if notification.wpx_comment_notification:
            await context.send_activity(f"Received comment: {notification.wpx_comment_notification.comment_id}")

适用于电子邮件通知的专用处理程序

专门为电子邮件通知注册更多处理程序：

from microsoft_agents_a365.notifications import AgentNotification
from microsoft_agents.activity import ChannelId, AgentSubChannel

# Create notification handler
agent_notification = AgentNotification(app)

# Use the convenience method for email notifications
@agent_notification.on_email()
async def handle_email(context, state, notification):
    email = notification.email_notification
    
    if not email:
        await context.send_activity('No email found')
        return
    
    # Process the email
    email_id = email.id
    conversation_id = email.conversation_id
    
    # Send response
    await context.send_activity('Thank you for your email!')

适用于文档评论的专用处理程序

为 Word、Excel 和 PowerPoint 注释通知注册更多处理程序：

from microsoft_agents_a365.notifications import AgentNotification

# Create notification handler
agent_notification = AgentNotification(app)

# Use convenience methods for document notifications
@agent_notification.on_word()
async def handle_word(context, state, notification):
    comment = notification.wpx_comment_notification
    
    if comment:
        document_id = comment.document_id
        comment_id = comment.comment_id
        await context.send_activity(f'Processing Word comment: {comment_id}')

@agent_notification.on_excel()
async def handle_excel(context, state, notification):
    comment = notification.wpx_comment_notification
    
    if comment:
        await context.send_activity('Processing Excel comment')

@agent_notification.on_powerpoint()
async def handle_powerpoint(context, state, notification):
    comment = notification.wpx_comment_notification
    
    if comment:
        await context.send_activity('Processing PowerPoint comment')

生命周期事件专用处理程序

为智能体生命周期事件（例如用户身份创建、工作负载接入和用户删除）注册更多处理程序：

from microsoft_agents_a365.notifications import AgentNotification

# Create notification handler
agent_notification = AgentNotification(app)

# Handle all lifecycle events
@agent_notification.on_agent_lifecycle_notification("*")
async def handle_lifecycle(context, state, notification):
    lifecycle_notification = notification.agent_lifecycle_notification
    if lifecycle_notification:
        event_type = lifecycle_notification.lifecycle_event_type
        
        if event_type == "agenticUserIdentityCreated":
            await context.send_activity('User identity created')
        elif event_type == "agenticUserWorkloadOnboardingUpdated":
            await context.send_activity('Workload onboarding completed')
        elif event_type == "agenticUserDeleted":
            await context.send_activity('User identity deleted')

高级配置

本节介绍了用于微调通知处理程序的高级配置选项。 通过这些配置，您可以控制处理程序的执行顺序、管理身份验证要求，并针对复杂场景优化通知处理。

处理程序优先级和排名

当使用多个专用处理程序时，请使用排名值指定优先级顺序。 排名值越低，优先级越高：

from microsoft_agents_a365.notifications import AgentNotification
from microsoft_agents.activity import ChannelId, AgentSubChannel

# Create notification handler
agent_notification = AgentNotification(app)

# Higher priority handler (processed first)
@agent_notification.on_email(rank=100)
async def high_priority_email(context, state, notification):
    # Handle with high priority
    pass

# Lower priority handler (processed after higher priority)
@agent_notification.on_email(rank=200)
async def low_priority_email(context, state, notification):
    # Handle with lower priority
    pass

身份验证处理程序

为需要身份验证的通知配置自动登录处理程序：

from microsoft_agents_a365.notifications import AgentNotification
from microsoft_agents.activity import ChannelId, AgentSubChannel

# Create notification handler
agent_notification = AgentNotification(app)

# Handler with automatic authentication
@agent_notification.on_email(auto_sign_in_handlers=['agentic'])
async def authenticated_email(context, state, notification):
    # Authentication is handled automatically
    pass

代码示例

有关所有受支持框架中通知处理的完整工作示例，请参阅 Agent 365 示例。

使用通知测试您的智能体

实现通知处理程序后，请测试您的智能体，以确保其能正确接收和处理不同类型的通知。 请按照 测试指南 设置环境，然后重点关注 使用通知活动进行测试 部分，通过智能体身份验证来验证您的通知。

监控通知处理

添加可观测性功能以监控智能体的通知处理情况。 跟踪通知处理情况、响应时间和错误率，以了解智能体的性能。 了解有关实施追踪和监控的更多信息。

  • Last updated on