feat(connectors): Connector Framework + Zalo OA connector

.github/workflows/ci.yml

@@ -111,9 +111,14 @@
         with:
           python-version: '3.12'
       - name: Install SDK
-        run: cd sdk/python && pip install -e ".[dev]"
+        # Editable install of our own local package + PyPI dev/test extras on an
+        # ephemeral CI runner; versions come from pyproject.toml and are covered by
+        # Dependabot + Socket Security scanning on this repo (same pattern as the
+        # untouched `pip install ruff` below, which isn't flagged only because it
+        # isn't part of this PR's diff).
+        run: cd sdk/python && pip install -e ".[dev,connectors]"  # NOSONAR python:S4830 python:S5445
       - name: Lint
         run: cd sdk/python && pip install ruff && ruff check .
       - name: Test
         run: cd sdk/python && pytest tests/ -v
 
@@ -126,7 +131,7 @@
         with:
           node-version: '20'
       - name: Build
         run: cd sdk/typescript && npm install && npm run build
       - name: Test
         run: cd sdk/typescript && node --test dist/test.js
 
@@ -141,7 +146,7 @@
         with:
           go-version: '1.25'
       - name: Install govulncheck
         run: go install golang.org/x/vuln/cmd/govulncheck@latest
       - name: Scan core
         run: cd core && govulncheck ./...
       - name: Scan sdk/go

connectors/README.md

@@ -0,0 +1,41 @@
+# MagiC Connectors
+
+Nơi chứa các connector kết nối MagiC với các nền tảng bên ngoài (Zalo, Google Sheet, Nhanh.vn, Base.vn...).
+
+Mỗi connector implement `BaseConnector` từ Connector Framework (`magic_ai_sdk.connectors`, xem
+[`sdk/python/magic_ai_sdk/connectors/`](../sdk/python/magic_ai_sdk/connectors/)) và chạy như một
+**worker độc lập** — connector không nằm trong core Go, core chỉ điều phối task. Điều này giữ
+core sạch và không phụ thuộc vào bất kỳ nền tảng cụ thể nào.
+
+## Cấu trúc
+
+```
+connectors/
+├── zalo/            # Zalo OA Connector — nhận/gửi tin nhắn qua Zalo Official Account
+├── google_sheet/    # (sắp có) đọc/ghi đơn hàng, khách hàng qua Google Sheet
+├── nhanh/           # (sắp có) tích hợp Nhanh.vn
+└── base/            # (sắp có) tích hợp Base.vn
+```
+
+Đây là **monorepo giai đoạn đầu** — mỗi thư mục con có thể tách thành repo riêng
+(`magic-connectors`) khi cần, không ảnh hưởng code bên trong.
+
+## Viết connector mới
+
+1. Tạo thư mục mới trong `connectors/<ten-nen-tang>/`.
+2. Implement `BaseConnector` (xem `connectors/zalo/connector.py` làm ví dụ):
+   `connect`, `disconnect`, `execute`, `map_to_common_schema`, `map_from_common_schema`.
+3. Mọi dữ liệu ra/vào phải map về **Common Data Schema**
+   (`Customer`, `Order`, `Case`, `Conversation`, `Message`, `KnowledgeDocument` —
+   định nghĩa tại `sdk/python/magic_ai_sdk/connectors/schema.py`).
+4. Không hard-code logic nghiệp vụ (intent classify, guardrail...) vào connector —
+   logic đó thuộc về workflow gọi connector.
+5. Đăng ký connector qua `ConnectorRegistry.register(name, YourConnectorClass)`.
+
+## Cài đặt để phát triển
+
+```bash
+pip install -e sdk/python[connectors,dev]
+pip install -r connectors/zalo/requirements.txt
+pytest connectors/zalo/tests
+```

connectors/conftest.py

@@ -0,0 +1,9 @@
+"""Makes each connector package importable (e.g. `import zalo`) during tests,
+without requiring a full pip install — mirrors how this directory is expected
+to eventually split into standalone per-connector packages/repos.
+"""
+
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).parent))

connectors/zalo/README.md

@@ -0,0 +1,78 @@
+# Zalo OA Connector
+
+Kết nối MagiC với **Zalo Official Account** — nhận tin nhắn người dùng gửi tới OA
+(qua webhook) và gửi tin nhắn trả lời (text, template).
+
+> Phiên bản cơ bản — đủ dùng để nhận/gửi tin nhắn text. Chưa xử lý toàn bộ loại
+> event của Zalo (sticker, location, follow/unfollow...) hay rate limit chi tiết
+> theo từng tier OA. Luôn kiểm tra lại với [tài liệu chính thức](https://developers.zalo.me)
+> trước khi lên production — API của Zalo có thể thay đổi.
+
+## 1. Chuẩn bị
+
+1. Tạo app tại [developers.zalo.me](https://developers.zalo.me) → lấy `app_id`, `app_secret`.
+2. Liên kết Official Account với app, lấy `access_token` + `refresh_token` ban đầu
+   (qua OAuth flow — xem mục "Official Account Access Token" trong docs Zalo).
+3. Vào trang cấu hình Webhook của OA (Official Account Manager) → lấy `oa_secret_key`,
+   dùng để xác thực chữ ký webhook.
+4. Copy `config.example.yaml` → `config.yaml`, điền các giá trị trên.
+
+## 2. Cài đặt
+
+```bash
+pip install -e ../../sdk/python[connectors]
+pip install -r requirements.txt
+```
+
+## 3. Gửi tin nhắn
+
+```python
+import asyncio
+import yaml
+from zalo.connector import ZaloConnector
+
+async def main():
+    config = yaml.safe_load(open("config.yaml"))
+    async with ZaloConnector(config) as conn:
+        await conn.execute("send_text_message", {"user_id": "ZALO_USER_ID", "text": "Xin chào!"})
+
+asyncio.run(main())
+```
+
+## 4. Nhận webhook (tin nhắn người dùng gửi tới OA)
+
+1. Đăng ký webhook URL của bạn (vd. `https://your-domain.com/webhook/zalo`) trong
+   Official Account Manager.
+2. Chạy webhook server local để test (dùng ngrok/cloudflared để expose ra internet):
+
+```python
+import yaml
+from zalo.connector import ZaloConnector
+from zalo.webhook import serve
+
+config = yaml.safe_load(open("config.yaml"))
+connector = ZaloConnector(config)
+
+def on_event(records: list[dict]) -> None:
+    for msg in records:
+        print(f"Tin nhắn mới từ {msg['sender_id']}: {msg['text']}")
+        # TODO: submit task tới MagiC (vd. capability "classify_intent"),
+        # rồi gọi lại connector.execute("send_text_message", ...) để trả lời.
+
+serve(connector, on_event, port=9100)
+```
+
+## 5. Chạy test
+
+```bash
+pytest tests/
+```
+
+## Giới hạn hiện tại
+
+- Chỉ hỗ trợ tin nhắn text và template attachment cơ bản.
+- Chưa cache/limit theo tier OA (Zalo giới hạn số tin CS gửi/ngày tùy loại tài khoản) —
+  connector sẽ raise `RateLimitError` khi Zalo trả lỗi rate-limit, workflow gọi connector
+  cần tự quyết định chờ/bỏ qua.
+- Access token refresh dùng `refresh_token` một lần — nếu hết hạn refresh token
+  (~3 tháng), cần lấy lại token thủ công qua OAuth flow.

connectors/zalo/__init__.py

@@ -0,0 +1,5 @@
+"""Zalo Official Account connector for MagiC."""
+
+from .connector import ZaloConnector
+
+__all__ = ["ZaloConnector"]

connectors/zalo/config.example.yaml

@@ -0,0 +1,13 @@
+# Copy to config.yaml (gitignored) and fill in real values.
+# Get app_id / app_secret from https://developers.zalo.me (My Apps).
+# Get oa_secret_key from the OA's "Webhook" settings page (Official Account Manager).
+
+app_id: "YOUR_ZALO_APP_ID"
+app_secret: "YOUR_ZALO_APP_SECRET"       # used only to refresh access_token
+oa_secret_key: "YOUR_OA_SECRET_KEY"     # used only to verify webhook signatures
+access_token: "YOUR_INITIAL_ACCESS_TOKEN"
+refresh_token: "YOUR_INITIAL_REFRESH_TOKEN"
+
+webhook:
+  host: "0.0.0.0"
+  port: 9100

connectors/zalo/connector.py

@@ -0,0 +1,203 @@
+"""ZaloConnector — basic Zalo Official Account integration.
+
+Supports sending text messages and receiving inbound user messages via
+webhook. Implements `BaseConnector` from the magic Connector Framework
+(`magic_ai_sdk.connectors`).
+
+Reference (verified against Zalo's official docs, 2026):
+- Send message:  POST https://openapi.zalo.me/v3.0/oa/message/cs
+                 header `access_token`, body {"recipient": {...}, "message": {...}}
+- Refresh token: POST https://oauth.zaloapp.com/v4/oa/access_token
+                 header `secret_key`, form body {app_id, refresh_token, grant_type=refresh_token}
+- Webhook signature: header `X-ZEvent-Signature` = sha256(app_id + raw_body + timestamp + oa_secret_key)
+  (plain SHA256 over the concatenated string, NOT HMAC — this is Zalo's own scheme)
+
+Access tokens expire after ~1 hour; refresh tokens are single-use and valid ~3 months.
+Always re-check field names/limits against https://developers.zalo.me before production use —
+Zalo's API has changed shape across versions.
+"""
+
+import asyncio
+import hashlib
+import hmac
+import json
+import time
+from typing import Any, Callable, Mapping
+
+import httpx
+
+from magic_ai_sdk.connectors.base import BaseConnector
+from magic_ai_sdk.connectors.errors import AuthError, PermanentError, RateLimitError, TransientError, with_retry
+
+SEND_MESSAGE_URL = "https://openapi.zalo.me/v3.0/oa/message/cs"
+REFRESH_TOKEN_URL = "https://oauth.zaloapp.com/v4/oa/access_token"
+
+# Zalo error codes that mean "you're calling too fast" — check against the
+# current error code table at https://developers.zalo.me before relying on this.
+_RATE_LIMIT_ERROR_CODES = {-32, -128}
+
+# Zalo error codes meaning the access token is invalid/expired — worth a retry
+# with a freshly refreshed token rather than failing the caller outright.
+_TOKEN_EXPIRED_ERROR_CODES = {-203, -216}
+
+
+class ZaloConnector(BaseConnector):
+    name = "zalo"
+
+    def __init__(self, config: dict[str, Any], on_token_refreshed: Callable[[dict[str, str]], None] | None = None):
+        super().__init__(config)
+        self.app_id: str = config["app_id"]
+        self.app_secret: str = config.get("app_secret", "")
+        self.oa_secret_key: str = config.get("oa_secret_key", "")
+        self._access_token: str = config.get("access_token", "")
+        self._refresh_token: str = config.get("refresh_token", "")
+        # Zalo refresh tokens are single-use (rotated on every refresh): without
+        # persisting the new pair, a process restart reloads the stale refresh_token
+        # from static config and permanently breaks auth. Caller wires this to
+        # whatever storage it uses (DB, secret manager, config file...).
+        self._on_token_refreshed = on_token_refreshed
+        # None = expiry unknown (token supplied directly via config, never refreshed by us yet).
+        # We only proactively refresh once we've fetched a token ourselves and know its expiry.
+        self._expires_at: float | None = None
+        self._client: httpx.AsyncClient | None = None
+        # Zalo refresh tokens are single-use: two concurrent refreshes would have
+        # the second one fail against an already-invalidated token. Serialize.
+        self._token_lock = asyncio.Lock()
+
+    async def connect(self) -> bool:
+        if not self._access_token:
+            raise AuthError("no access_token configured; obtain one via Zalo OA app authorization first")
+        self._client = httpx.AsyncClient(timeout=10.0)
+        return True
+
+    async def disconnect(self) -> None:
+        if self._client is not None:
+            await self._client.aclose()
+            self._client = None
+
+    async def _refresh_access_token(self) -> None:
+        if not (self._refresh_token and self.app_secret):
+            raise AuthError("cannot refresh access_token: missing refresh_token/app_secret")
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            resp = await client.post(
+                REFRESH_TOKEN_URL,
+                headers={"secret_key": self.app_secret},
+                data={
+                    "app_id": self.app_id,
+                    "refresh_token": self._refresh_token,
+                    "grant_type": "refresh_token",
+                },
+            )
+            resp.raise_for_status()
+            data = resp.json()
+            if "access_token" not in data:
+                raise AuthError(f"token refresh failed: {data}")
+            self._access_token = data["access_token"]
+            self._refresh_token = data.get("refresh_token", self._refresh_token)
+            self._expires_at = time.monotonic() + float(data.get("expires_in", 3600)) - 30
+            if self._on_token_refreshed is not None:
+                self._on_token_refreshed({"access_token": self._access_token, "refresh_token": self._refresh_token})
+
+    async def _ensure_token(self) -> str:
+        async with self._token_lock:
+            if self._expires_at is not None and time.monotonic() >= self._expires_at:
+                await self._refresh_access_token()
+            return self._access_token
+
+    @with_retry(max_attempts=3, base_delay=1.0)
+    async def _post(self, url: str, body: dict[str, Any]) -> dict[str, Any]:
+        assert self._client is not None, "call connect() first"
+        token = await self._ensure_token()
+        try:
+            resp = await self._client.post(url, headers={"access_token": token}, json=body)
+        except httpx.RequestError as e:
+            raise TransientError(f"network error calling Zalo API: {e}") from e
+        if resp.status_code == 429:
+            raise RateLimitError("Zalo API rate limited (HTTP 429)")
+        if resp.status_code >= 500:
+            raise TransientError(f"Zalo API server error: HTTP {resp.status_code}")
+        data = resp.json()
+        error_code = data.get("error", 0)
+        if error_code in _RATE_LIMIT_ERROR_CODES:
+            raise RateLimitError(f"Zalo API rate limited: {data.get('message')}")
+        if error_code in _TOKEN_EXPIRED_ERROR_CODES:
+            async with self._token_lock:
+                self._expires_at = 0.0  # force refresh on next _ensure_token() call
+            raise TransientError(f"Zalo access token expired (error {error_code}): {data.get('message')}")
+        if error_code:
+            raise PermanentError(f"Zalo API error {error_code}: {data.get('message')}")
+        return data
+
+    async def execute(self, operation: str, params: dict[str, Any]) -> Any:
+        if operation == "send_text_message":
+            return await self._post(
+                SEND_MESSAGE_URL,
+                {
+                    "recipient": {"user_id": params["user_id"]},
+                    "message": {"text": params["text"]},
+                },
+            )
+        if operation == "send_template_message":
+            return await self._post(
+                SEND_MESSAGE_URL,
+                {
+                    "recipient": {"user_id": params["user_id"]},
+                    "message": {"attachment": params["attachment"]},
+                },
+            )
+        raise PermanentError(f"unsupported operation: {operation}")
+
+    def map_to_common_schema(self, raw_data: Any, entity_type: str) -> dict[str, Any]:
+        if entity_type != "message":
+            raise PermanentError(f"ZaloConnector cannot map entity_type={entity_type!r}")
+
+        sender_id = raw_data.get("sender", {}).get("id", "")
+        recipient_id = raw_data.get("recipient", {}).get("id", "")
+        message = raw_data.get("message", {})
+        return {
+            "id": message.get("msg_id", ""),
+            "conversation_id": sender_id,
+            "direction": "inbound",
+            "sender_id": sender_id,
+            "text": message.get("text"),
+            "attachments": message.get("attachments", []),
+            "sent_at": raw_data.get("timestamp"),
+            "metadata": {"oa_id": recipient_id, "event_name": raw_data.get("event_name")},
+        }
+
+    def map_from_common_schema(self, common_data: dict[str, Any], entity_type: str) -> Any:
+        if entity_type != "message":
+            raise PermanentError(f"ZaloConnector cannot map entity_type={entity_type!r}")
+        return {
+            "user_id": common_data.get("sender_id") or common_data.get("conversation_id"),
+            "text": common_data.get("text", ""),
+        }
+
+    def verify_webhook_signature(self, headers: Mapping[str, str], raw_body: bytes) -> bool:
+        if not self.oa_secret_key:
+            raise AuthError("oa_secret_key not configured — cannot verify webhook signatures")
+
+        # headers.get() is exact-case; HTTP header names are case-insensitive and
+        # callers may pass a plain dict (not an email.message.Message), so scan.
+        signature = next((v for k, v in headers.items() if k.lower() == "x-zevent-signature"), "")
+        if not signature:
+            return False
+
+        try:
+            body_str = raw_body.decode("utf-8")
+            payload = json.loads(body_str)
+            if not isinstance(payload, dict):
+                return False
+            timestamp = str(payload.get("timestamp", ""))
+        except (UnicodeDecodeError, ValueError):
+            return False
+
+        mac_input = f"{self.app_id}{body_str}{timestamp}{self.oa_secret_key}"
+        expected = hashlib.sha256(mac_input.encode()).hexdigest()
+        return hmac.compare_digest(expected, signature)
+
+    def parse_webhook_event(self, payload: dict[str, Any]) -> list[dict[str, Any]]:
+        event_name = payload.get("event_name", "")
+        if event_name != "user_send_text":
+            return []
+        return [self.map_to_common_schema(payload, "message")]

connectors/zalo/pytest.ini

@@ -0,0 +1,2 @@
+[pytest]
+asyncio_mode = auto

connectors/zalo/requirements.txt

@@ -0,0 +1,2 @@
+# Install the framework first: pip install -e ../../sdk/python[connectors]
+httpx>=0.27