feat(msgraph): add webhook listener platform #1
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,299 @@ | ||
| """Microsoft Graph webhook adapter for change-notification ingress.""" | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| import asyncio | ||
| import json | ||
| import logging | ||
| from collections import deque | ||
| from hashlib import sha1 | ||
| from typing import Any, Awaitable, Callable, Dict, Optional | ||
|
|
||
| try: | ||
| from aiohttp import web | ||
|
|
||
| AIOHTTP_AVAILABLE = True | ||
| except ImportError: | ||
| AIOHTTP_AVAILABLE = False | ||
| web = None # type: ignore[assignment] | ||
|
|
||
| from gateway.config import Platform, PlatformConfig | ||
| from gateway.platforms.base import ( | ||
| BasePlatformAdapter, | ||
| MessageEvent, | ||
| MessageType, | ||
| SendResult, | ||
| ) | ||
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
| DEFAULT_HOST = "0.0.0.0" | ||
| DEFAULT_PORT = 8646 | ||
| DEFAULT_WEBHOOK_PATH = "/msgraph/webhook" | ||
| DEFAULT_MAX_SEEN_RECEIPTS = 5000 | ||
| NotificationScheduler = Callable[[Dict[str, Any], MessageEvent], Awaitable[None] | None] | ||
|
|
||
|
|
||
| def check_msgraph_webhook_requirements() -> bool: | ||
| """Return whether required webhook dependencies are available.""" | ||
| return AIOHTTP_AVAILABLE | ||
|
|
||
|
|
||
| class MSGraphWebhookAdapter(BasePlatformAdapter): | ||
| """Receive Microsoft Graph change notifications and surface them internally.""" | ||
|
|
||
| def __init__(self, config: PlatformConfig): | ||
| super().__init__(config, Platform.MSGRAPH_WEBHOOK) | ||
| extra = config.extra or {} | ||
| self._host: str = str(extra.get("host", DEFAULT_HOST)) | ||
| self._port: int = int(extra.get("port", DEFAULT_PORT)) | ||
| self._webhook_path: str = self._normalize_path( | ||
| extra.get("webhook_path", DEFAULT_WEBHOOK_PATH) | ||
| ) | ||
| self._health_path: str = self._normalize_path(extra.get("health_path", "/health")) | ||
| self._accepted_resources: list[str] = [ | ||
| str(value).strip() | ||
| for value in (extra.get("accepted_resources") or []) | ||
| if str(value).strip() | ||
| ] | ||
| self._client_state: Optional[str] = self._string_or_none(extra.get("client_state")) | ||
| self._max_seen_receipts = max( | ||
| 1, int(extra.get("max_seen_receipts", DEFAULT_MAX_SEEN_RECEIPTS)) | ||
| ) | ||
| self._runner = None | ||
| self._notification_scheduler: Optional[NotificationScheduler] = None | ||
| self._seen_receipts: set[str] = set() | ||
| self._seen_receipt_order: deque[str] = deque() | ||
| self._accepted_count = 0 | ||
| self._duplicate_count = 0 | ||
|
|
||
| @staticmethod | ||
| def _string_or_none(value: Any) -> Optional[str]: | ||
| if value is None: | ||
| return None | ||
| text = str(value).strip() | ||
| return text or None | ||
|
|
||
| @staticmethod | ||
| def _normalize_path(path: Any) -> str: | ||
| raw = str(path or "").strip() or "/" | ||
| return raw if raw.startswith("/") else f"/{raw}" | ||
|
|
||
| @staticmethod | ||
| def _build_receipt_key(notification: Dict[str, Any]) -> str: | ||
| explicit_id = str(notification.get("id") or "").strip() | ||
| if explicit_id: | ||
| return f"id:{explicit_id}" | ||
| payload = "|".join( | ||
| [ | ||
| str(notification.get("subscriptionId") or ""), | ||
| str(notification.get("changeType") or ""), | ||
| str(notification.get("resource") or ""), | ||
| json.dumps(notification.get("resourceData") or {}, sort_keys=True), | ||
| ] | ||
| ) | ||
| return f"sha1:{sha1(payload.encode('utf-8')).hexdigest()}" | ||
|
|
||
| def set_notification_scheduler(self, scheduler: Optional[NotificationScheduler]) -> None: | ||
| self._notification_scheduler = scheduler | ||
|
|
||
| async def connect(self) -> bool: | ||
| app = web.Application() | ||
| app.router.add_get(self._health_path, self._handle_health) | ||
| app.router.add_get(self._webhook_path, self._handle_notification) | ||
| app.router.add_post(self._webhook_path, self._handle_notification) | ||
|
|
||
| self._runner = web.AppRunner(app) | ||
| await self._runner.setup() | ||
| site = web.TCPSite(self._runner, self._host, self._port) | ||
| await site.start() | ||
| self._mark_connected() | ||
| logger.info( | ||
| "[msgraph_webhook] Listening on %s:%d%s", | ||
| self._host, | ||
| self._port, | ||
| self._webhook_path, | ||
| ) | ||
| return True | ||
|
|
||
| async def disconnect(self) -> None: | ||
| if self._runner is not None: | ||
| await self._runner.cleanup() | ||
| self._runner = None | ||
| self._mark_disconnected() | ||
|
|
||
| async def send( | ||
| self, | ||
| chat_id: str, | ||
| content: str, | ||
| reply_to: Optional[str] = None, | ||
| metadata: Optional[Dict[str, Any]] = None, | ||
| ) -> SendResult: | ||
| logger.info("[msgraph_webhook] Response for %s: %s", chat_id, content[:200]) | ||
| return SendResult(success=True) | ||
|
|
||
| async def get_chat_info(self, chat_id: str) -> Dict[str, Any]: | ||
| return {"name": chat_id, "type": "webhook"} | ||
|
|
||
| async def _handle_health(self, request: "web.Request") -> "web.Response": | ||
| return web.json_response( | ||
| { | ||
| "status": "ok", | ||
| "platform": self.platform.value, | ||
| "webhook_path": self._webhook_path, | ||
| "accepted": self._accepted_count, | ||
| "duplicates": self._duplicate_count, | ||
| } | ||
| ) | ||
|
|
||
| async def _handle_notification(self, request: "web.Request") -> "web.Response": | ||
| validation_token = request.query.get("validationToken", "") | ||
| if validation_token: | ||
| return web.Response(text=validation_token, content_type="text/plain") | ||
|
|
||
| try: | ||
| body = await request.json() | ||
| except Exception: | ||
| return web.json_response({"error": "Invalid JSON body"}, status=400) | ||
|
|
||
| notifications = body.get("value") | ||
| if not isinstance(notifications, list): | ||
| return web.json_response({"error": "Missing notification batch"}, status=400) | ||
|
|
||
| accepted = 0 | ||
| duplicates = 0 | ||
| rejected = 0 | ||
| scheduled = 0 | ||
|
|
||
| for raw_notification in notifications: | ||
| if not isinstance(raw_notification, dict): | ||
| rejected += 1 | ||
| continue | ||
| notification = dict(raw_notification) | ||
| if not self._resource_accepted(str(notification.get("resource") or "")): | ||
| rejected += 1 | ||
| continue | ||
| if not self._verify_client_state(notification): | ||
| rejected += 1 | ||
| continue | ||
|
|
||
| receipt_key = self._build_receipt_key(notification) | ||
| if self._has_seen_receipt(receipt_key): | ||
| duplicates += 1 | ||
| continue | ||
| self._remember_receipt(receipt_key) | ||
|
|
||
| accepted += 1 | ||
| scheduled += 1 | ||
| self._accepted_count += 1 | ||
| event = self._build_message_event(notification, receipt_key) | ||
| self._schedule_notification(notification, event) | ||
|
|
||
| self._duplicate_count += duplicates | ||
| status = 202 if accepted or duplicates else 403 | ||
| return web.json_response( | ||
| { | ||
| "status": "accepted" if accepted or duplicates else "rejected", | ||
| "accepted": accepted, | ||
| "duplicates": duplicates, | ||
| "rejected": rejected, | ||
| "scheduled": scheduled, | ||
| }, | ||
| status=status, | ||
| ) | ||
|
|
||
| def _resource_accepted(self, resource: str) -> bool: | ||
| if not self._accepted_resources: | ||
| return True | ||
| for pattern in self._accepted_resources: | ||
| if pattern.endswith("*") and resource.startswith(pattern[:-1]): | ||
| return True | ||
| if resource == pattern or resource.startswith(f"{pattern}/"): | ||
| return True | ||
|
Comment on lines
+211
to
+212
There was a problem hiding this comment.
Resource filtering compares raw strings, but Graph subscription resources are commonly configured with a leading Useful? React with 👍 / 👎. |
||
| return False | ||
|
|
||
| def _verify_client_state(self, notification: Dict[str, Any]) -> bool: | ||
| expected = self._client_state | ||
| if expected is None: | ||
| return True | ||
| provided = self._string_or_none(notification.get("clientState")) | ||
| return provided == expected | ||
|
|
||
| def _has_seen_receipt(self, receipt_key: str) -> bool: | ||
| return receipt_key in self._seen_receipts | ||
|
|
||
| def _remember_receipt(self, receipt_key: str) -> None: | ||
| self._seen_receipts.add(receipt_key) | ||
| self._seen_receipt_order.append(receipt_key) | ||
| while len(self._seen_receipt_order) > self._max_seen_receipts: | ||
| oldest = self._seen_receipt_order.popleft() | ||
| self._seen_receipts.discard(oldest) | ||
|
|
||
| def _build_message_event( | ||
| self, | ||
| notification: Dict[str, Any], | ||
| receipt_key: str, | ||
| ) -> MessageEvent: | ||
| source = self.build_source( | ||
| chat_id=f"msgraph:{notification.get('subscriptionId', 'unknown')}", | ||
| chat_name="msgraph/webhook", | ||
| chat_type="webhook", | ||
| user_id="msgraph", | ||
| user_name="Microsoft Graph", | ||
| ) | ||
| return MessageEvent( | ||
| text=self._render_prompt(notification), | ||
| message_type=MessageType.TEXT, | ||
| source=source, | ||
| raw_message=notification, | ||
| message_id=receipt_key, | ||
| internal=True, | ||
| ) | ||
|
|
||
| def _render_prompt(self, notification: Dict[str, Any]) -> str: | ||
| template = self.config.extra.get("prompt", "") | ||
| if template: | ||
| payload = { | ||
| "notification": notification, | ||
| "resource": notification.get("resource", ""), | ||
| "change_type": notification.get("changeType", ""), | ||
| "subscription_id": notification.get("subscriptionId", ""), | ||
| } | ||
| return self._render_template(template, payload) | ||
| rendered = json.dumps(notification, indent=2, sort_keys=True)[:4000] | ||
| return f"Microsoft Graph change notification:\n\n```json\n{rendered}\n```" | ||
|
|
||
| def _render_template(self, template: str, payload: Dict[str, Any]) -> str: | ||
| import re | ||
|
|
||
| def _resolve(match: "re.Match[str]") -> str: | ||
| key = match.group(1) | ||
| value: Any = payload | ||
| for part in key.split("."): | ||
| if isinstance(value, dict): | ||
| value = value.get(part, f"{{{key}}}") | ||
| else: | ||
| return f"{{{key}}}" | ||
| if isinstance(value, (dict, list)): | ||
| return json.dumps(value, sort_keys=True)[:2000] | ||
| return str(value) | ||
|
|
||
| return re.sub(r"\{([a-zA-Z0-9_.]+)\}", _resolve, template) | ||
|
|
||
| def _schedule_notification( | ||
| self, | ||
| notification: Dict[str, Any], | ||
| event: MessageEvent, | ||
| ) -> None: | ||
| scheduler = self._notification_scheduler | ||
| if scheduler is not None: | ||
| result = scheduler(notification, event) | ||
| if asyncio.iscoroutine(result): | ||
| task = asyncio.create_task(result) | ||
| self._background_tasks.add(task) | ||
| task.add_done_callback(self._background_tasks.discard) | ||
| return | ||
|
|
||
| task = asyncio.create_task(self.handle_message(event)) | ||
| self._background_tasks.add(task) | ||
| task.add_done_callback(self._background_tasks.discard) | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
_build_receipt_key()falls back to hashing onlysubscriptionId,changeType,resource, andresourceDatawhennotification.idis absent. Because Graph marksidas optional, repeated legitimate updates for the same resource (for example multipleupdatedevents with the same resource path/resourceData) produce the same key and are dropped as duplicates, which silently loses real events until eviction from the 5,000-entry window.Useful? React with 👍 / 👎.