feat(teams-pipeline): add plugin runtime and operator cli

gateway/config.py

@@ -101,6 +101,7 @@ class Platform(Enum):
     DINGTALK = "dingtalk"
     API_SERVER = "api_server"
     WEBHOOK = "webhook"
+    MSGRAPH_WEBHOOK = "msgraph_webhook"
     FEISHU = "feishu"
     WECOM = "wecom"
     WECOM_CALLBACK = "wecom_callback"
@@ -376,6 +377,7 @@ def from_dict(cls, data: Dict[str, Any]) -> "StreamingConfig":
     Platform.SMS: lambda cfg: bool(os.getenv("TWILIO_ACCOUNT_SID")),
     Platform.API_SERVER: lambda cfg: True,
     Platform.WEBHOOK: lambda cfg: True,
+    Platform.MSGRAPH_WEBHOOK: lambda cfg: True,
     Platform.FEISHU: lambda cfg: bool(cfg.extra.get("app_id")),
     Platform.WECOM: lambda cfg: bool(cfg.extra.get("bot_id")),
     Platform.WECOM_CALLBACK: lambda cfg: bool(
@@ -1366,6 +1368,48 @@ def _apply_env_overrides(config: GatewayConfig) -> None:
         if webhook_secret:
             config.platforms[Platform.WEBHOOK].extra["secret"] = webhook_secret
 
+    # Microsoft Graph webhook platform
+    msgraph_webhook_enabled = os.getenv("MSGRAPH_WEBHOOK_ENABLED", "").lower() in (
+        "true",
+        "1",
+        "yes",
+    )
+    msgraph_webhook_port = os.getenv("MSGRAPH_WEBHOOK_PORT")
+    msgraph_webhook_client_state = os.getenv("MSGRAPH_WEBHOOK_CLIENT_STATE", "")
+    msgraph_webhook_resources = os.getenv("MSGRAPH_WEBHOOK_ACCEPTED_RESOURCES", "")
+    if (
+        msgraph_webhook_enabled
+        or Platform.MSGRAPH_WEBHOOK in config.platforms
+        or msgraph_webhook_port
+        or msgraph_webhook_client_state
+        or msgraph_webhook_resources
+    ):
+        if Platform.MSGRAPH_WEBHOOK not in config.platforms:
+            config.platforms[Platform.MSGRAPH_WEBHOOK] = PlatformConfig()
+        if msgraph_webhook_enabled:
+            config.platforms[Platform.MSGRAPH_WEBHOOK].enabled = True
+        if msgraph_webhook_port:
+            try:
+                config.platforms[Platform.MSGRAPH_WEBHOOK].extra["port"] = int(
+                    msgraph_webhook_port
+                )
+            except ValueError:
+                pass
+        if msgraph_webhook_client_state:
+            config.platforms[Platform.MSGRAPH_WEBHOOK].extra["client_state"] = (
+                msgraph_webhook_client_state
+            )
+        if msgraph_webhook_resources:
+            resources = [
+                resource.strip()
+                for resource in msgraph_webhook_resources.split(",")
+                if resource.strip()
+            ]
+            if resources:
+                config.platforms[Platform.MSGRAPH_WEBHOOK].extra[
+                    "accepted_resources"
+                ] = resources
+
     # DingTalk
     dingtalk_client_id = os.getenv("DINGTALK_CLIENT_ID")
     dingtalk_client_secret = os.getenv("DINGTALK_CLIENT_SECRET")

gateway/platforms/msgraph_webhook.py

@@ -0,0 +1,307 @@
+"""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]) -> Optional[str]:
+        explicit_id = str(notification.get("id") or "").strip()
+        if explicit_id:
+            return f"id:{explicit_id}"
+        return None
+
+    @staticmethod
+    def _normalize_resource_value(resource: str) -> str:
+        return str(resource or "").strip().strip("/")
+
+    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 receipt_key is not None:
+                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
+        normalized_resource = self._normalize_resource_value(resource)
+        for pattern in self._accepted_resources:
+            normalized_pattern = self._normalize_resource_value(pattern)
+            if not normalized_pattern:
+                continue
+            if normalized_pattern.endswith("*"):
+                prefix = normalized_pattern[:-1].rstrip("/")
+                if normalized_resource == prefix or normalized_resource.startswith(f"{prefix}/"):
+                    return True
+                continue
+            if (
+                normalized_resource == normalized_pattern
+                or normalized_resource.startswith(f"{normalized_pattern}/")
+            ):
+                return True
+        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: Optional[str],
+    ) -> MessageEvent:
+        message_id = receipt_key or f"sha1:{sha1(json.dumps(notification, sort_keys=True).encode('utf-8')).hexdigest()}"
+        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=message_id,
+            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)