Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
from .agent_notification import (
AgentHandler,
AgentNotification,
RouteHandler,
)

# Import all models from the models subpackage
Expand All @@ -29,6 +30,7 @@
# Main notification handler
"AgentNotification",
"AgentHandler",
"RouteHandler",
# Models and data classes
"AgentNotificationActivity",
"EmailReference",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
from __future__ import annotations

from collections.abc import Awaitable, Callable, Iterable
from typing import Any, TypeVar
from typing import Any, TypeAlias

from microsoft_agents.activity import ChannelId
from microsoft_agents.hosting.core import TurnContext
Expand All @@ -14,8 +14,12 @@
from .models.agent_notification_activity import AgentNotificationActivity, NotificationTypes
from .models.agent_subchannel import AgentSubChannel

TContext = TypeVar("TContext", bound=TurnContext)
TState = TypeVar("TState", bound=TurnState)
#: Type alias for the route handler function registered with the application.
#:
#: Route handlers are the inner async functions that the application framework calls
#: when a matching notification is received. They accept a :class:`~microsoft_agents.hosting.core.TurnContext`
#: and a :class:`~microsoft_agents.hosting.core.app.state.TurnState`.
RouteHandler: TypeAlias = Callable[[TurnContext, TurnState], Awaitable[None]]

#: Type alias for agent notification handler functions.
#:
Expand All @@ -26,7 +30,8 @@
#: Args:
#: context: The turn context for the current conversation turn.
#: state: The application state for the current turn.
#: notification: The typed notification activity with parsed entities.
#: notification: The typed :class:`~microsoft_agents_a365.notifications.AgentNotificationActivity`
#: with parsed entities.
#:
#: Example:
#:
Expand All @@ -40,7 +45,9 @@
#: email = notification.email
#: if email:
#: print(f"Processing email: {email.id}")
AgentHandler = Callable[[TContext, TState, AgentNotificationActivity], Awaitable[None]]
AgentHandler: TypeAlias = Callable[
[TurnContext, TurnState, AgentNotificationActivity], Awaitable[None]
]


class AgentNotification:
Expand Down Expand Up @@ -220,9 +227,7 @@ def decorator(handler: AgentHandler):

return decorator

def on_email(
self, **kwargs: Any
) -> Callable[[AgentHandler], Callable[[TurnContext, TurnState], Awaitable[None]]]:
def on_email(self, **kwargs: Any) -> Callable[[AgentHandler], RouteHandler]:
"""Register a handler for Outlook email notifications.

This is a convenience decorator that registers a handler for notifications
Expand Down Expand Up @@ -251,9 +256,7 @@ async def handle_email(context, state, notification):
ChannelId(channel="agents", sub_channel=AgentSubChannel.EMAIL), **kwargs
)

def on_word(
self, **kwargs: Any
) -> Callable[[AgentHandler], Callable[[TurnContext, TurnState], Awaitable[None]]]:
def on_word(self, **kwargs: Any) -> Callable[[AgentHandler], RouteHandler]:
"""Register a handler for Microsoft Word comment notifications.

This is a convenience decorator that registers a handler for notifications
Expand All @@ -278,9 +281,7 @@ async def handle_word_comment(context, state, notification):
ChannelId(channel="agents", sub_channel=AgentSubChannel.WORD), **kwargs
)

def on_excel(
self, **kwargs: Any
) -> Callable[[AgentHandler], Callable[[TurnContext, TurnState], Awaitable[None]]]:
def on_excel(self, **kwargs: Any) -> Callable[[AgentHandler], RouteHandler]:
"""Register a handler for Microsoft Excel comment notifications.

This is a convenience decorator that registers a handler for notifications
Expand All @@ -305,9 +306,7 @@ async def handle_excel_comment(context, state, notification):
ChannelId(channel="agents", sub_channel=AgentSubChannel.EXCEL), **kwargs
)

def on_powerpoint(
self, **kwargs: Any
) -> Callable[[AgentHandler], Callable[[TurnContext, TurnState], Awaitable[None]]]:
def on_powerpoint(self, **kwargs: Any) -> Callable[[AgentHandler], RouteHandler]:
"""Register a handler for Microsoft PowerPoint comment notifications.

This is a convenience decorator that registers a handler for notifications
Expand All @@ -332,9 +331,7 @@ async def handle_powerpoint_comment(context, state, notification):
ChannelId(channel="agents", sub_channel=AgentSubChannel.POWERPOINT), **kwargs
)

def on_lifecycle(
self, **kwargs: Any
) -> Callable[[AgentHandler], Callable[[TurnContext, TurnState], Awaitable[None]]]:
def on_lifecycle(self, **kwargs: Any) -> Callable[[AgentHandler], RouteHandler]:
"""Register a handler for all agent lifecycle event notifications.

This is a convenience decorator that registers a handler for all lifecycle
Expand All @@ -353,11 +350,9 @@ def on_lifecycle(
async def handle_any_lifecycle_event(context, state, notification):
print(f"Lifecycle event type: {notification.notification_type}")
"""
return self.on_lifecycle_notification("*", **kwargs)
return self.on_agent_lifecycle_notification("*", **kwargs)

def on_user_created(
self, **kwargs: Any
) -> Callable[[AgentHandler], Callable[[TurnContext, TurnState], Awaitable[None]]]:
def on_user_created(self, **kwargs: Any) -> Callable[[AgentHandler], RouteHandler]:
"""Register a handler for user creation lifecycle events.

This is a convenience decorator that registers a handler specifically for
Expand All @@ -376,11 +371,9 @@ def on_user_created(
async def handle_user_created(context, state, notification):
print("New agentic user identity created")
"""
return self.on_lifecycle_notification(AgentLifecycleEvent.USERCREATED, **kwargs)
return self.on_agent_lifecycle_notification(AgentLifecycleEvent.USERCREATED, **kwargs)
Copy link

Copilot AI Apr 24, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

on_user_created() delegates to on_agent_lifecycle_notification(AgentLifecycleEvent.USERCREATED, ...), but on_agent_lifecycle_notification’s route_selector currently does not compare the received activity.value_type to the requested lifecycle_event (it returns True for any known lifecycle event). This means user-created handlers will run for unrelated lifecycle events; update the selector to normalize and equality-check value_type against the requested event when lifecycle_event != "*".

Copilot uses AI. Check for mistakes.

def on_user_workload_onboarding(
self, **kwargs: Any
) -> Callable[[AgentHandler], Callable[[TurnContext, TurnState], Awaitable[None]]]:
def on_user_workload_onboarding(self, **kwargs: Any) -> Callable[[AgentHandler], RouteHandler]:
"""Register a handler for user workload onboarding update events.

This is a convenience decorator that registers a handler for events that occur
Expand All @@ -399,13 +392,11 @@ def on_user_workload_onboarding(
async def handle_onboarding_update(context, state, notification):
print("User workload onboarding status updated")
"""
return self.on_lifecycle_notification(
return self.on_agent_lifecycle_notification(
AgentLifecycleEvent.USERWORKLOADONBOARDINGUPDATED, **kwargs
)
Comment on lines +395 to 397
Copy link

Copilot AI Apr 24, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Because on_agent_lifecycle_notification’s selector doesn’t currently filter by the requested lifecycle_event, this on_user_workload_onboarding() registration will match any known lifecycle event (not just onboarding updates). Adjust on_agent_lifecycle_notification.route_selector to compare normalized context.activity.value_type to the requested event when not using the "*" wildcard.

Copilot uses AI. Check for mistakes.

def on_user_deleted(
self, **kwargs: Any
) -> Callable[[AgentHandler], Callable[[TurnContext, TurnState], Awaitable[None]]]:
def on_user_deleted(self, **kwargs: Any) -> Callable[[AgentHandler], RouteHandler]:
"""Register a handler for user deletion lifecycle events.

This is a convenience decorator that registers a handler specifically for
Expand All @@ -424,7 +415,7 @@ def on_user_deleted(
async def handle_user_deleted(context, state, notification):
print("Agentic user identity deleted")
"""
return self.on_lifecycle_notification(AgentLifecycleEvent.USERDELETED, **kwargs)
return self.on_agent_lifecycle_notification(AgentLifecycleEvent.USERDELETED, **kwargs)
Copy link

Copilot AI Apr 24, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

on_user_deleted() is intended to route only deletion events, but on_agent_lifecycle_notification’s selector currently returns True for any known lifecycle event when a specific event is provided. This will cause deletion handlers to run for other lifecycle events; fix the selector to check equality between requested lifecycle_event and received activity.value_type (after normalization).

Copilot uses AI. Check for mistakes.

@staticmethod
def _normalize_subchannel(value: str | AgentSubChannel | None) -> str:
Expand Down
2 changes: 2 additions & 0 deletions tests/notifications/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.
Loading
Loading