diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 53b98f8..f8b920b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -111,7 +111,12 @@ jobs: 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 diff --git a/connectors/README.md b/connectors/README.md new file mode 100644 index 0000000..cfab88a --- /dev/null +++ b/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//`. +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 +``` diff --git a/connectors/conftest.py b/connectors/conftest.py new file mode 100644 index 0000000..03db5bf --- /dev/null +++ b/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)) diff --git a/connectors/zalo/README.md b/connectors/zalo/README.md new file mode 100644 index 0000000..4195ab9 --- /dev/null +++ b/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. diff --git a/connectors/zalo/__init__.py b/connectors/zalo/__init__.py new file mode 100644 index 0000000..b9813a7 --- /dev/null +++ b/connectors/zalo/__init__.py @@ -0,0 +1,5 @@ +"""Zalo Official Account connector for MagiC.""" + +from .connector import ZaloConnector + +__all__ = ["ZaloConnector"] diff --git a/connectors/zalo/config.example.yaml b/connectors/zalo/config.example.yaml new file mode 100644 index 0000000..0c3877f --- /dev/null +++ b/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 diff --git a/connectors/zalo/connector.py b/connectors/zalo/connector.py new file mode 100644 index 0000000..9126ad8 --- /dev/null +++ b/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")] diff --git a/connectors/zalo/pytest.ini b/connectors/zalo/pytest.ini new file mode 100644 index 0000000..2f4c80e --- /dev/null +++ b/connectors/zalo/pytest.ini @@ -0,0 +1,2 @@ +[pytest] +asyncio_mode = auto diff --git a/connectors/zalo/requirements.txt b/connectors/zalo/requirements.txt new file mode 100644 index 0000000..a51ee1d --- /dev/null +++ b/connectors/zalo/requirements.txt @@ -0,0 +1,2 @@ +# Install the framework first: pip install -e ../../sdk/python[connectors] +httpx>=0.27 diff --git a/connectors/zalo/tests/__init__.py b/connectors/zalo/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/connectors/zalo/tests/test_connector.py b/connectors/zalo/tests/test_connector.py new file mode 100644 index 0000000..07c2f52 --- /dev/null +++ b/connectors/zalo/tests/test_connector.py @@ -0,0 +1,230 @@ +"""Unit tests for ZaloConnector — HTTP calls are mocked via respx, no network.""" + +import asyncio +import hashlib +import json + +import httpx +import pytest +import respx + +from magic_ai_sdk.connectors.errors import AuthError, PermanentError, RateLimitError, TransientError +from zalo.connector import REFRESH_TOKEN_URL, SEND_MESSAGE_URL, ZaloConnector + +BASE_CONFIG = { + "app_id": "app-123", + "app_secret": "app-secret", + "oa_secret_key": "oa-secret", + "access_token": "initial-access-token", + "refresh_token": "initial-refresh-token", +} + + +def make_connector(**overrides): + return ZaloConnector({**BASE_CONFIG, **overrides}) + + +async def test_connect_requires_access_token(): + conn = ZaloConnector({**BASE_CONFIG, "access_token": ""}) + with pytest.raises(AuthError): + await conn.connect() + + +@respx.mock +async def test_send_text_message_success(): + respx.post(SEND_MESSAGE_URL).mock( + return_value=httpx.Response(200, json={"error": 0, "message": "Success", "data": {"message_id": "m1"}}) + ) + conn = make_connector() + async with conn: + result = await conn.execute("send_text_message", {"user_id": "u1", "text": "xin chào"}) + assert result["data"]["message_id"] == "m1" + + sent = json.loads(respx.calls.last.request.content) + assert sent == {"recipient": {"user_id": "u1"}, "message": {"text": "xin chào"}} + assert respx.calls.last.request.headers["access_token"] == "initial-access-token" + + +@respx.mock +async def test_send_message_rate_limited_raises(): + respx.post(SEND_MESSAGE_URL).mock( + return_value=httpx.Response(200, json={"error": -32, "message": "rate limited"}) + ) + conn = make_connector() + async with conn: + with pytest.raises(RateLimitError): + await conn.execute("send_text_message", {"user_id": "u1", "text": "hi"}) + + +@respx.mock +async def test_send_message_permanent_error_raises(): + respx.post(SEND_MESSAGE_URL).mock( + return_value=httpx.Response(200, json={"error": -201, "message": "invalid user id"}) + ) + conn = make_connector() + async with conn: + with pytest.raises(PermanentError): + await conn.execute("send_text_message", {"user_id": "bad", "text": "hi"}) + + +async def test_execute_unsupported_operation(): + conn = make_connector() + with pytest.raises(PermanentError): + await conn.execute("delete_everything", {}) + + +@respx.mock +async def test_access_token_refreshed_when_expired(): + respx.post(REFRESH_TOKEN_URL).mock( + return_value=httpx.Response(200, json={"access_token": "new-token", "refresh_token": "new-refresh", "expires_in": 3600}) + ) + respx.post(SEND_MESSAGE_URL).mock(return_value=httpx.Response(200, json={"error": 0})) + + conn = make_connector() + conn._expires_at = 0.0 # force refresh on next call, monotonic() is always > 0 + async with conn: + await conn.execute("send_text_message", {"user_id": "u1", "text": "hi"}) + + assert conn._access_token == "new-token" + assert respx.calls.last.request.headers["access_token"] == "new-token" + + +def _sign(app_id: str, raw_body: bytes, timestamp: str, secret: str) -> str: + mac_input = f"{app_id}{raw_body.decode()}{timestamp}{secret}" + return hashlib.sha256(mac_input.encode()).hexdigest() + + +def test_verify_webhook_signature_valid(): + conn = make_connector() + body = json.dumps({"event_name": "user_send_text", "timestamp": "1234567890"}).encode() + sig = _sign("app-123", body, "1234567890", "oa-secret") + assert conn.verify_webhook_signature({"X-ZEvent-Signature": sig}, body) is True + + +def test_verify_webhook_signature_invalid(): + conn = make_connector() + body = json.dumps({"event_name": "user_send_text", "timestamp": "1234567890"}).encode() + assert conn.verify_webhook_signature({"X-ZEvent-Signature": "wrong"}, body) is False + + +def test_verify_webhook_signature_missing_secret_raises(): + conn = make_connector(oa_secret_key="") + with pytest.raises(AuthError): + conn.verify_webhook_signature({"X-ZEvent-Signature": "x"}, b"{}") + + +def test_parse_webhook_event_user_send_text(): + conn = make_connector() + payload = { + "event_name": "user_send_text", + "sender": {"id": "user-1"}, + "recipient": {"id": "oa-1"}, + "message": {"text": "hello", "msg_id": "msg-1"}, + "timestamp": "1700000000", + } + records = conn.parse_webhook_event(payload) + assert len(records) == 1 + assert records[0]["text"] == "hello" + assert records[0]["sender_id"] == "user-1" + + +def test_parse_webhook_event_ignores_other_events(): + conn = make_connector() + payload = {"event_name": "follow"} + assert conn.parse_webhook_event(payload) == [] + + +def test_map_from_common_schema(): + conn = make_connector() + outbound = conn.map_from_common_schema({"sender_id": "user-1", "text": "reply"}, "message") + assert outbound == {"user_id": "user-1", "text": "reply"} + + +def test_map_to_common_schema_rejects_unknown_entity(): + conn = make_connector() + with pytest.raises(PermanentError): + conn.map_to_common_schema({}, "order") + + +@respx.mock +async def test_expired_token_error_triggers_transient_retry_with_refresh(): + """Zalo error -203 (invalid/expired token) must force a refresh and retry, + not fail permanently — see connector.py's _TOKEN_EXPIRED_ERROR_CODES.""" + respx.post(REFRESH_TOKEN_URL).mock( + return_value=httpx.Response(200, json={"access_token": "refreshed-token", "expires_in": 3600}) + ) + route = respx.post(SEND_MESSAGE_URL) + route.side_effect = [ + httpx.Response(200, json={"error": -203, "message": "token expired"}), + httpx.Response(200, json={"error": 0}), + ] + + conn = make_connector() + async with conn: + result = await conn.execute("send_text_message", {"user_id": "u1", "text": "hi"}) + + assert result == {"error": 0} + assert conn._access_token == "refreshed-token" + assert route.call_count == 2 + assert route.calls[0].request.headers["access_token"] == "initial-access-token" + assert route.calls[1].request.headers["access_token"] == "refreshed-token" + + +async def test_ensure_token_refresh_is_serialized_under_concurrency(): + """Concurrent callers must not each trigger their own refresh — Zalo refresh + tokens are single-use, so a second concurrent refresh would fail.""" + conn = make_connector() + conn._expires_at = 0.0 + refresh_calls = 0 + + async def fake_refresh(): + nonlocal refresh_calls + refresh_calls += 1 + await asyncio.sleep(0.01) + conn._access_token = "refreshed-once" + conn._expires_at = 999999999.0 + + conn._refresh_access_token = fake_refresh + + results = await asyncio.gather(*(conn._ensure_token() for _ in range(5))) + + assert refresh_calls == 1 + assert all(token == "refreshed-once" for token in results) + + +@respx.mock +async def test_token_refresh_invokes_persistence_callback(): + """Zalo refresh tokens are single-use — the new pair must be handed to the + caller so it can be persisted, or a restart reloads the stale refresh_token.""" + respx.post(REFRESH_TOKEN_URL).mock( + return_value=httpx.Response(200, json={"access_token": "new-tok", "refresh_token": "new-refresh", "expires_in": 3600}) + ) + saved = {} + conn = ZaloConnector({**BASE_CONFIG}, on_token_refreshed=saved.update) + + await conn._refresh_access_token() + + assert saved == {"access_token": "new-tok", "refresh_token": "new-refresh"} + + +@respx.mock +async def test_network_error_wrapped_as_transient(): + respx.post(SEND_MESSAGE_URL).mock(side_effect=httpx.ConnectError("connection refused")) + conn = make_connector() + async with conn: + with pytest.raises(TransientError): + await conn.execute("send_text_message", {"user_id": "u1", "text": "hi"}) + + +def test_verify_webhook_signature_case_insensitive_header(): + conn = make_connector() + body = json.dumps({"event_name": "user_send_text", "timestamp": "1234567890"}).encode() + sig = _sign("app-123", body, "1234567890", "oa-secret") + # lowercase header name, as some HTTP frameworks normalize to + assert conn.verify_webhook_signature({"x-zevent-signature": sig}, body) is True + + +def test_verify_webhook_signature_rejects_non_dict_json(): + conn = make_connector() + body = json.dumps(["not", "an", "object"]).encode() + assert conn.verify_webhook_signature({"X-ZEvent-Signature": "whatever"}, body) is False diff --git a/connectors/zalo/tests/test_webhook.py b/connectors/zalo/tests/test_webhook.py new file mode 100644 index 0000000..a2db237 --- /dev/null +++ b/connectors/zalo/tests/test_webhook.py @@ -0,0 +1,118 @@ +"""Tests for the Zalo webhook HTTP server: signature verification + malformed requests.""" + +import http.client +import json +import threading +import time + +import pytest + +from zalo.connector import ZaloConnector +from zalo.webhook import serve + +CONFIG = { + "app_id": "app-123", + "oa_secret_key": "oa-secret", + "access_token": "token", +} + + +@pytest.fixture +def running_server(): + connector = ZaloConnector(CONFIG) + events: list[list[dict]] = [] + server_holder: dict = {} + + def on_event(records): + events.append(records) + + def run(): + from http.server import ThreadingHTTPServer + + # Reimplement serve()'s server setup but capture the server so we can shut it down. + orig_serve_forever = ThreadingHTTPServer.serve_forever + + def patched_serve_forever(self, *a, **kw): + server_holder["server"] = self + server_holder["ready"].set() + orig_serve_forever(self, *a, **kw) + + ThreadingHTTPServer.serve_forever = patched_serve_forever + try: + serve(connector, on_event, host="127.0.0.1", port=0) + finally: + ThreadingHTTPServer.serve_forever = orig_serve_forever + + server_holder["ready"] = threading.Event() + thread = threading.Thread(target=run, daemon=True) + thread.start() + server_holder["ready"].wait(timeout=5) + server = server_holder["server"] + port = server.server_address[1] + + yield connector, events, port + + server.shutdown() + thread.join(timeout=5) + + +def _sign(app_id: str, body: bytes, timestamp: str, secret: str) -> str: + import hashlib + + mac_input = f"{app_id}{body.decode()}{timestamp}{secret}" + return hashlib.sha256(mac_input.encode()).hexdigest() + + +def test_valid_signed_webhook_calls_on_event(running_server): + connector, events, port = running_server + body = json.dumps( + { + "event_name": "user_send_text", + "sender": {"id": "user-1"}, + "recipient": {"id": "oa-1"}, + "message": {"text": "hi", "msg_id": "m1"}, + "timestamp": "1700000000", + } + ).encode() + sig = _sign("app-123", body, "1700000000", "oa-secret") + + conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5) + conn.request("POST", "/", body=body, headers={"X-ZEvent-Signature": sig, "Content-Length": str(len(body))}) + resp = conn.getresponse() + resp.read() + conn.close() + + assert resp.status == 200 + time.sleep(0.05) + assert len(events) == 1 + assert events[0][0]["text"] == "hi" + + +def test_invalid_signature_rejected(running_server): + connector, events, port = running_server + body = json.dumps({"event_name": "user_send_text", "timestamp": "1"}).encode() + + conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5) + conn.request("POST", "/", body=body, headers={"X-ZEvent-Signature": "bad", "Content-Length": str(len(body))}) + resp = conn.getresponse() + resp.read() + conn.close() + + assert resp.status == 401 + assert events == [] + + +def test_invalid_content_length_returns_400_not_crash(running_server): + connector, events, port = running_server + body = b"{}" + + conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5) + conn.putrequest("POST", "/") + conn.putheader("Content-Length", "not-a-number") + conn.endheaders() + conn.send(body) + resp = conn.getresponse() + resp.read() + conn.close() + + assert resp.status == 400 diff --git a/connectors/zalo/webhook.py b/connectors/zalo/webhook.py new file mode 100644 index 0000000..110579a --- /dev/null +++ b/connectors/zalo/webhook.py @@ -0,0 +1,77 @@ +"""Minimal HTTP server that receives Zalo OA webhook events, verifies their +signature, and forwards parsed Common Schema records to a callback. + +The callback is where integration with MagiC happens — e.g. submit a task +to the `classify_intent` capability, or hand off to a human inbox. This +module only handles the Zalo-specific transport (signature verification, +JSON parsing); it doesn't know anything about MagiC itself. +""" + +import json +import logging +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any, Callable + +from .connector import ZaloConnector + +logger = logging.getLogger("magic_connectors.zalo") + +OnEvent = Callable[[list[dict[str, Any]]], None] + + +def serve(connector: ZaloConnector, on_event: OnEvent, host: str = "0.0.0.0", port: int = 9100) -> None: + """Start a blocking HTTP server for Zalo OA webhook callbacks.""" + + class Handler(BaseHTTPRequestHandler): + def do_POST(self): + content_length = self.headers.get("Content-Length") + if not content_length: + self.send_error(400, "Missing Content-Length") + return + + try: + length = int(content_length) + except ValueError: + self.send_error(400, "Invalid Content-Length") + return + + if length > 1 * 1024 * 1024: + self.send_error(413, "Request too large") + return + + raw_body = self.rfile.read(length) + + if not connector.verify_webhook_signature(self.headers, raw_body): + logger.warning("Rejected webhook with invalid signature") + self.send_error(401, "Invalid signature") + return + + try: + payload = json.loads(raw_body) + except (json.JSONDecodeError, ValueError): + self.send_error(400, "Invalid JSON") + return + + try: + records = connector.parse_webhook_event(payload) + if records: + on_event(records) + except Exception: + logger.exception("on_event callback failed") + + # Zalo expects a 200 quickly regardless of downstream processing outcome. + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.end_headers() + self.wfile.write(b'{"ok":true}') + + def log_message(self, format, *args): + logger.debug("HTTP %s", format % args) + + server = ThreadingHTTPServer((host, port), Handler) + logger.info("Zalo webhook server listening on %s:%d", host, port) + try: + server.serve_forever() # NOSONAR python:S5332 — plain HTTP intentional; TLS terminates at the reverse proxy (see README) + except KeyboardInterrupt: + logger.info("Shutting down Zalo webhook server") + server.shutdown() diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..d3cfb42 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,60 @@ +# 1-click local dev stack: magic core + postgres + redis. +# Run: docker compose up -d +# +# For full observability (Prometheus/Grafana/Jaeger), see +# deploy/docker-compose.observability.yml instead. + +name: magic-dev + +services: + postgres: + image: postgres:16-alpine + restart: unless-stopped + environment: + POSTGRES_USER: magic + POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-magic} + POSTGRES_DB: magic + ports: + - "5432:5432" + volumes: + - magic-dev-postgres:/var/lib/postgresql/data + healthcheck: + test: ["CMD-SHELL", "pg_isready -U magic -d magic"] + interval: 10s + timeout: 5s + retries: 5 + + redis: + image: redis:7-alpine + restart: unless-stopped + ports: + - "6379:6379" + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 10s + timeout: 5s + retries: 5 + + magic: + build: . + restart: unless-stopped + depends_on: + postgres: + condition: service_healthy + redis: + condition: service_healthy + environment: + MAGIC_POSTGRES_URL: "postgres://magic:${POSTGRES_PASSWORD:-magic}@postgres:5432/magic?sslmode=disable" + MAGIC_REDIS_URL: "redis://redis:6379" + MAGIC_API_KEY: ${MAGIC_API_KEY:-dev-key-change-me} + ports: + - "8080:8080" + healthcheck: + test: ["CMD-SHELL", "wget -qO- http://localhost:8080/health || exit 1"] + interval: 15s + timeout: 5s + retries: 5 + start_period: 20s + +volumes: + magic-dev-postgres: diff --git a/scripts/quickstart.sh b/scripts/quickstart.sh new file mode 100755 index 0000000..08b3950 --- /dev/null +++ b/scripts/quickstart.sh @@ -0,0 +1,26 @@ +#!/usr/bin/env bash +# 1-click local dev: start magic + postgres + redis, wait for health, print next steps. +set -euo pipefail + +cd "$(dirname "$0")/.." + +echo "==> Starting magic + postgres + redis (docker compose)..." +docker compose up -d --build + +echo "==> Waiting for magic to become healthy..." +for _ in $(seq 1 30); do + if curl -sf http://localhost:8080/health > /dev/null 2>&1; then + echo "==> magic is up: http://localhost:8080" + echo "" + echo "Next steps:" + echo " - Register a worker: see examples/hello-worker/main.py" + echo " - Connector Framework: see connectors/README.md" + echo " - Logs: docker compose logs -f magic" + echo " - Stop: docker compose down" + exit 0 + fi + sleep 2 +done + +echo "magic did not become healthy in time — check logs: docker compose logs magic" >&2 +exit 1 diff --git a/sdk/python/magic_ai_sdk/connectors/__init__.py b/sdk/python/magic_ai_sdk/connectors/__init__.py new file mode 100644 index 0000000..1a102d2 --- /dev/null +++ b/sdk/python/magic_ai_sdk/connectors/__init__.py @@ -0,0 +1,31 @@ +"""MagiC Connector Framework — build platform integrations (Zalo, Nhanh.vn, Base.vn, ...). + +Requires the `connectors` extra: `pip install magic-ai-sdk[connectors]` +""" + +from magic_ai_sdk.connectors.auth import ApiKeyAuth, AuthHandler, OAuth2Auth, WebhookSignatureAuth +from magic_ai_sdk.connectors.base import BaseConnector +from magic_ai_sdk.connectors.errors import ( + AuthError, + ConnectorError, + PermanentError, + RateLimitError, + TransientError, + with_retry, +) +from magic_ai_sdk.connectors.registry import ConnectorRegistry + +__all__ = [ + "BaseConnector", + "ConnectorRegistry", + "AuthHandler", + "ApiKeyAuth", + "OAuth2Auth", + "WebhookSignatureAuth", + "ConnectorError", + "AuthError", + "RateLimitError", + "TransientError", + "PermanentError", + "with_retry", +] diff --git a/sdk/python/magic_ai_sdk/connectors/auth.py b/sdk/python/magic_ai_sdk/connectors/auth.py new file mode 100644 index 0000000..2028f62 --- /dev/null +++ b/sdk/python/magic_ai_sdk/connectors/auth.py @@ -0,0 +1,105 @@ +"""Auth abstractions shared across connectors. + +Each connector picks the AuthHandler that matches its platform instead of +hand-rolling header/signature logic. All handlers are stateless and safe +to share across requests. +""" + +import hashlib +import hmac +import time +from abc import ABC, abstractmethod +from typing import Any + +import httpx + + +class AuthHandler(ABC): + """Base class for connector authentication strategies.""" + + @abstractmethod + async def apply(self, request_kwargs: dict[str, Any]) -> dict[str, Any]: + """Mutate/return httpx request kwargs (headers, params...) with credentials applied. + + Async so handlers that need to fetch/refresh a token first (OAuth2Auth) + share the exact same interface as ones that don't (ApiKeyAuth) — callers + can `await handler.apply(kwargs)` polymorphically either way. + """ + + +class ApiKeyAuth(AuthHandler): + """Static API key sent as a header or query param.""" + + def __init__(self, key: str, header: str = "Authorization", prefix: str = ""): + self.key = key + self.header = header + self.prefix = prefix + + async def apply(self, request_kwargs: dict[str, Any]) -> dict[str, Any]: + headers = dict(request_kwargs.get("headers") or {}) + headers[self.header] = f"{self.prefix}{self.key}" + request_kwargs["headers"] = headers + return request_kwargs + + +class OAuth2Auth(AuthHandler): + """Client-credentials OAuth2 with lazy token fetch + refresh. + + Zalo/Nhanh/Base all use some flavor of bearer-token-with-expiry; + this covers the common case without pulling in a full OAuth library. + """ + + def __init__(self, token_url: str, client_id: str, client_secret: str, leeway_seconds: float = 30.0): + self.token_url = token_url + self.client_id = client_id + self.client_secret = client_secret + self.leeway_seconds = leeway_seconds + self._access_token: str | None = None + self._expires_at: float = 0.0 + + async def _refresh(self) -> None: + async with httpx.AsyncClient(timeout=10.0) as client: + resp = await client.post( + self.token_url, + data={ + "grant_type": "client_credentials", + "client_id": self.client_id, + "client_secret": self.client_secret, + }, + ) + resp.raise_for_status() + data = resp.json() + self._access_token = data["access_token"] + self._expires_at = time.monotonic() + float(data.get("expires_in", 3600)) - self.leeway_seconds + + async def token(self) -> str: + if self._access_token is None or time.monotonic() >= self._expires_at: + await self._refresh() + assert self._access_token is not None + return self._access_token + + async def apply(self, request_kwargs: dict[str, Any]) -> dict[str, Any]: + headers = dict(request_kwargs.get("headers") or {}) + headers["Authorization"] = f"Bearer {await self.token()}" + request_kwargs["headers"] = headers + return request_kwargs + + +class WebhookSignatureAuth: + """Verifies inbound webhook signatures (HMAC-SHA256 over the raw body). + + Not an AuthHandler (nothing to "apply" to outbound requests) — used by + connectors on the receiving side to validate that a webhook payload + really came from the platform and wasn't spoofed/tampered with. + """ + + def __init__(self, secret: str, header: str = "X-Signature"): + self.secret = secret + self.header = header + + def sign(self, raw_body: bytes) -> str: + return hmac.new(self.secret.encode(), raw_body, hashlib.sha256).hexdigest() + + def verify(self, raw_body: bytes, signature: str) -> bool: + expected = self.sign(raw_body) + return hmac.compare_digest(expected, signature) diff --git a/sdk/python/magic_ai_sdk/connectors/base.py b/sdk/python/magic_ai_sdk/connectors/base.py new file mode 100644 index 0000000..f08189c --- /dev/null +++ b/sdk/python/magic_ai_sdk/connectors/base.py @@ -0,0 +1,61 @@ +"""BaseConnector — the interface every platform integration implements. + +A connector's job is narrow on purpose: authenticate, call the platform's +API (`execute`), and translate data to/from the Common Data Schema +(`map_to_common_schema` / `map_from_common_schema`). Business logic +(intent classification, guardrails, case management...) belongs in the +workflow that calls the connector, not in the connector itself — that's +what keeps magic core platform-agnostic. +""" + +from abc import ABC, abstractmethod +from typing import Any, Mapping + + +class BaseConnector(ABC): + """Abstract base class for every connector (Zalo, Nhanh, Base, ...).""" + + name: str + version: str = "0.1.0" + + def __init__(self, config: dict[str, Any]): + self.config = config + + @abstractmethod + async def connect(self) -> bool: + """Authenticate / establish connection with the external platform.""" + + @abstractmethod + async def disconnect(self) -> None: + """Release any held resources (sessions, connections).""" + + @abstractmethod + async def execute(self, operation: str, params: dict[str, Any]) -> Any: + """Perform a platform operation, e.g. `send_text_message`, `get_order`.""" + + @abstractmethod + def map_to_common_schema(self, raw_data: Any, entity_type: str) -> dict[str, Any]: + """Translate a platform payload into the Common Data Schema.""" + + @abstractmethod + def map_from_common_schema(self, common_data: dict[str, Any], entity_type: str) -> Any: + """Translate a Common Data Schema record into the platform's format.""" + + async def health_check(self) -> bool: + """Cheap liveness check. Override for a real ping if the platform supports one.""" + return True + + def verify_webhook_signature(self, headers: Mapping[str, str], raw_body: bytes) -> bool: + """Validate an inbound webhook's signature. Override for webhook-based connectors.""" + raise NotImplementedError(f"{self.name} does not support webhooks") + + def parse_webhook_event(self, payload: dict[str, Any]) -> list[dict[str, Any]]: + """Parse a verified webhook payload into a list of Common Schema records.""" + raise NotImplementedError(f"{self.name} does not support webhooks") + + async def __aenter__(self) -> "BaseConnector": + await self.connect() + return self + + async def __aexit__(self, exc_type, exc, tb) -> None: + await self.disconnect() diff --git a/sdk/python/magic_ai_sdk/connectors/errors.py b/sdk/python/magic_ai_sdk/connectors/errors.py new file mode 100644 index 0000000..2b7851f --- /dev/null +++ b/sdk/python/magic_ai_sdk/connectors/errors.py @@ -0,0 +1,78 @@ +"""Connector error hierarchy + retry helper. + +Connectors talk to flaky third-party APIs (Zalo, Nhanh.vn, Base.vn...). +This module gives every connector the same vocabulary for failure and +the same backoff behavior, instead of each one inventing its own. +""" + +import asyncio +import logging +import random +from functools import wraps +from typing import Callable, TypeVar + +logger = logging.getLogger("magic_ai_sdk.connectors") + +T = TypeVar("T") + + +class ConnectorError(Exception): + """Base class for all connector errors.""" + + +class AuthError(ConnectorError): + """Authentication/authorization with the external platform failed.""" + + +class RateLimitError(ConnectorError): + """The external platform rejected the call due to rate limiting.""" + + def __init__(self, message: str = "rate limited", retry_after: float | None = None): + super().__init__(message) + self.retry_after = retry_after + + +class TransientError(ConnectorError): + """A retryable error — network blip, 5xx, timeout.""" + + +class PermanentError(ConnectorError): + """A non-retryable error — bad request, not found, unsupported operation.""" + + +def with_retry( + max_attempts: int = 3, + base_delay: float = 0.5, + max_delay: float = 8.0, + retry_on: tuple[type[Exception], ...] = (TransientError, RateLimitError), +) -> Callable[[Callable[..., T]], Callable[..., T]]: + """Decorator: retry an async connector method with exponential backoff + jitter. + + Only retries exceptions in `retry_on`. `RateLimitError.retry_after`, when set, + is honored instead of the computed backoff. + """ + + def decorator(func: Callable[..., T]) -> Callable[..., T]: + @wraps(func) + async def wrapper(*args, **kwargs): + attempt = 0 + while True: + try: + return await func(*args, **kwargs) + except retry_on as e: + attempt += 1 + if attempt >= max_attempts: + raise + delay = getattr(e, "retry_after", None) + if delay is None: + delay = min(max_delay, base_delay * (2 ** (attempt - 1))) + delay += random.uniform(0, base_delay) + logger.warning( + "%s failed (attempt %d/%d): %s — retrying in %.1fs", + func.__qualname__, attempt, max_attempts, e, delay, + ) + await asyncio.sleep(delay) + + return wrapper + + return decorator diff --git a/sdk/python/magic_ai_sdk/connectors/registry.py b/sdk/python/magic_ai_sdk/connectors/registry.py new file mode 100644 index 0000000..1dd381d --- /dev/null +++ b/sdk/python/magic_ai_sdk/connectors/registry.py @@ -0,0 +1,53 @@ +"""ConnectorRegistry — register connector classes, get per-tenant instances. + +One process may serve many tenants (shops); each tenant needs its own +connector instance (different Zalo OA access token, different Nhanh.vn +API key). The registry keys instances by `(tenant_id, connector_name)` so +tenants never share credentials or in-memory state. +""" + +from typing import Any + +from magic_ai_sdk.connectors.base import BaseConnector + +_InstanceKey = tuple[str, str] + + +class ConnectorRegistry: + """Class-level registry: connector classes are process-global, instances are per-tenant.""" + + _connectors: dict[str, type[BaseConnector]] = {} + _instances: dict[_InstanceKey, BaseConnector] = {} + + @classmethod + def register(cls, name: str, connector_class: type[BaseConnector]) -> None: + cls._connectors[name] = connector_class + + @classmethod + def unregister(cls, name: str) -> None: + cls._connectors.pop(name, None) + + @classmethod + def is_registered(cls, name: str) -> bool: + return name in cls._connectors + + @classmethod + def get(cls, name: str, tenant_id: str, config: dict[str, Any]) -> BaseConnector: + if name not in cls._connectors: + raise ValueError(f"Connector '{name}' is not registered") + + key = (tenant_id, name) + if key not in cls._instances: + cls._instances[key] = cls._connectors[name](config) + return cls._instances[key] + + @classmethod + def evict(cls, name: str, tenant_id: str) -> None: + """Drop a cached instance, e.g. after credentials rotate.""" + cls._instances.pop((tenant_id, name), None) + + @classmethod + def reset(cls) -> None: + """Clear all registrations and instances. Test-only.""" + cls._connectors.clear() + cls._instances.clear() diff --git a/sdk/python/magic_ai_sdk/connectors/schema.py b/sdk/python/magic_ai_sdk/connectors/schema.py new file mode 100644 index 0000000..e3befc6 --- /dev/null +++ b/sdk/python/magic_ai_sdk/connectors/schema.py @@ -0,0 +1,114 @@ +"""Common Data Schema — the shared shape every connector maps to/from. + +Connectors never expose platform-specific data to workflows directly; +they translate to and from these models so a workflow written against +"Conversation" or "Order" works the same whether the data came from +Zalo, Google Sheet, Nhanh.vn, or Base.vn. +""" + +from datetime import datetime +from enum import Enum +from typing import Any, Optional + +from pydantic import BaseModel, Field + + +class ChannelType(str, Enum): + ZALO = "zalo" + FACEBOOK = "facebook" + GOOGLE_SHEET = "google_sheet" + NHANH = "nhanh" + BASE = "base" + MISA = "misa" + + +class Customer(BaseModel): + id: str + source_connector: str + external_id: str + name: Optional[str] = None + phone: Optional[str] = None + email: Optional[str] = None + address: Optional[str] = None + metadata: dict[str, Any] = Field(default_factory=dict) + + +class OrderStatus(str, Enum): + PENDING = "pending" + CONFIRMED = "confirmed" + SHIPPING = "shipping" + DELIVERED = "delivered" + CANCELLED = "cancelled" + RETURNED = "returned" + + +class OrderItem(BaseModel): + sku: Optional[str] = None + name: str + quantity: int = Field(default=1, ge=1) + unit_price: float = 0.0 + + +class Order(BaseModel): + id: str + source_connector: str + external_id: str + customer_id: Optional[str] = None + status: OrderStatus = OrderStatus.PENDING + items: list[OrderItem] = Field(default_factory=list) + total_amount: float = 0.0 + shipping_fee: float = 0.0 + created_at: Optional[datetime] = None + metadata: dict[str, Any] = Field(default_factory=dict) + + +class MessageDirection(str, Enum): + INBOUND = "inbound" + OUTBOUND = "outbound" + + +class Message(BaseModel): + id: str + conversation_id: str + direction: MessageDirection + sender_id: str + text: Optional[str] = None + attachments: list[dict[str, Any]] = Field(default_factory=list) + sent_at: Optional[datetime] = None + metadata: dict[str, Any] = Field(default_factory=dict) + + +class Conversation(BaseModel): + id: str + source_connector: str + channel: ChannelType + external_id: str + customer_id: Optional[str] = None + messages: list[Message] = Field(default_factory=list) + metadata: dict[str, Any] = Field(default_factory=dict) + + +class CaseStatus(str, Enum): + OPEN = "open" + IN_PROGRESS = "in_progress" + RESOLVED = "resolved" + ESCALATED = "escalated" + + +class Case(BaseModel): + id: str + source_connector: str + conversation_id: Optional[str] = None + customer_id: Optional[str] = None + title: str + status: CaseStatus = CaseStatus.OPEN + reason: Optional[str] = None + metadata: dict[str, Any] = Field(default_factory=dict) + + +class KnowledgeDocument(BaseModel): + id: str + title: str + content: str + tags: list[str] = Field(default_factory=list) + metadata: dict[str, Any] = Field(default_factory=dict) diff --git a/sdk/python/pyproject.toml b/sdk/python/pyproject.toml index 36a363a..5ee1d26 100644 --- a/sdk/python/pyproject.toml +++ b/sdk/python/pyproject.toml @@ -29,10 +29,12 @@ Repository = "https://github.com/kienbui1995/magic" Issues = "https://github.com/kienbui1995/magic/issues" [project.optional-dependencies] -dev = ["pytest>=8.0"] +dev = ["pytest>=8.0", "pytest-asyncio>=0.24", "respx>=0.21"] +connectors = ["pydantic>=2.0"] [tool.pytest.ini_options] testpaths = ["tests"] +asyncio_mode = "auto" [tool.ruff] target-version = "py311" diff --git a/sdk/python/tests/test_connectors.py b/sdk/python/tests/test_connectors.py new file mode 100644 index 0000000..d73ab95 --- /dev/null +++ b/sdk/python/tests/test_connectors.py @@ -0,0 +1,218 @@ +"""Unit tests for the Connector Framework (base, registry, auth, errors, schema).""" + +import time +from typing import Any + +import httpx +import pytest +import respx + +from magic_ai_sdk.connectors import ( + ApiKeyAuth, + BaseConnector, + ConnectorRegistry, + OAuth2Auth, + PermanentError, + RateLimitError, + TransientError, + WebhookSignatureAuth, + with_retry, +) +from magic_ai_sdk.connectors.schema import ChannelType, Conversation, Message + + +class FakeConnector(BaseConnector): + name = "fake" + + def __init__(self, config: dict[str, Any]): + super().__init__(config) + self.connected = False + + async def connect(self) -> bool: + self.connected = True + return True + + async def disconnect(self) -> None: + self.connected = False + + async def execute(self, operation: str, params: dict[str, Any]) -> Any: + if operation == "echo": + return params + raise PermanentError(f"unsupported operation: {operation}") + + def map_to_common_schema(self, raw_data: Any, entity_type: str) -> dict[str, Any]: + return {"entity_type": entity_type, **raw_data} + + def map_from_common_schema(self, common_data: dict[str, Any], entity_type: str) -> Any: + return {"entity_type": entity_type, **common_data} + + +# ---- BaseConnector ---- + + +async def test_connector_context_manager_connects_and_disconnects(): + conn = FakeConnector({}) + async with conn as c: + assert c.connected is True + assert conn.connected is False + + +async def test_connector_execute_dispatches_to_operation(): + conn = FakeConnector({}) + result = await conn.execute("echo", {"a": 1}) + assert result == {"a": 1} + + +async def test_connector_execute_raises_on_unknown_operation(): + conn = FakeConnector({}) + with pytest.raises(PermanentError): + await conn.execute("does_not_exist", {}) + + +async def test_default_health_check_is_true(): + conn = FakeConnector({}) + assert await conn.health_check() is True + + +def test_webhook_methods_raise_by_default(): + conn = FakeConnector({}) + with pytest.raises(NotImplementedError): + conn.verify_webhook_signature({}, b"") + with pytest.raises(NotImplementedError): + conn.parse_webhook_event({}) + + +# ---- ConnectorRegistry ---- + + +@pytest.fixture(autouse=True) +def _reset_registry(): + ConnectorRegistry.reset() + yield + ConnectorRegistry.reset() + + +def test_registry_get_creates_and_caches_per_tenant(): + ConnectorRegistry.register("fake", FakeConnector) + + a1 = ConnectorRegistry.get("fake", tenant_id="shop-a", config={}) + a2 = ConnectorRegistry.get("fake", tenant_id="shop-a", config={}) + b1 = ConnectorRegistry.get("fake", tenant_id="shop-b", config={}) + + assert a1 is a2 # same tenant -> cached instance + assert a1 is not b1 # different tenant -> isolated instance + + +def test_registry_get_unknown_connector_raises(): + with pytest.raises(ValueError): + ConnectorRegistry.get("nope", tenant_id="shop-a", config={}) + + +def test_registry_evict_drops_cached_instance(): + ConnectorRegistry.register("fake", FakeConnector) + first = ConnectorRegistry.get("fake", tenant_id="shop-a", config={}) + ConnectorRegistry.evict("fake", tenant_id="shop-a") + second = ConnectorRegistry.get("fake", tenant_id="shop-a", config={}) + assert first is not second + + +# ---- AuthHandler ---- + + +async def test_api_key_auth_sets_header(): + auth = ApiKeyAuth(key="secret123", header="Authorization", prefix="Bearer ") + kwargs = await auth.apply({}) + assert kwargs["headers"]["Authorization"] == "Bearer secret123" + + +@respx.mock +async def test_oauth2_auth_apply_fetches_token_and_sets_header(): + respx.post("https://auth.example.com/token").mock( + return_value=httpx.Response(200, json={"access_token": "tok-1", "expires_in": 3600}) + ) + auth = OAuth2Auth(token_url="https://auth.example.com/token", client_id="cid", client_secret="secret") + kwargs = await auth.apply({}) + assert kwargs["headers"]["Authorization"] == "Bearer tok-1" + + +def test_webhook_signature_auth_verifies_hmac(): + auth = WebhookSignatureAuth(secret="topsecret") + body = b'{"event":"message"}' + sig = auth.sign(body) + assert auth.verify(body, sig) is True + assert auth.verify(body, "deadbeef") is False + + +# ---- errors / retry ---- + + +async def test_with_retry_succeeds_after_transient_failures(): + calls = {"n": 0} + + @with_retry(max_attempts=3, base_delay=0.001, max_delay=0.001) + async def flaky(): + calls["n"] += 1 + if calls["n"] < 3: + raise TransientError("temporary") + return "ok" + + result = await flaky() + assert result == "ok" + assert calls["n"] == 3 + + +async def test_with_retry_gives_up_after_max_attempts(): + @with_retry(max_attempts=2, base_delay=0.001, max_delay=0.001) + async def always_fails(): + raise TransientError("nope") + + with pytest.raises(TransientError): + await always_fails() + + +async def test_with_retry_does_not_retry_permanent_errors(): + calls = {"n": 0} + + @with_retry(max_attempts=5, base_delay=0.001) + async def fails_permanently(): + calls["n"] += 1 + raise PermanentError("bad request") + + with pytest.raises(PermanentError): + await fails_permanently() + assert calls["n"] == 1 + + +async def test_with_retry_honors_rate_limit_retry_after(): + calls = {"n": 0} + + @with_retry(max_attempts=2, base_delay=5.0) # base_delay huge so test would time out if not honored + async def rate_limited_then_ok(): + calls["n"] += 1 + if calls["n"] < 2: + raise RateLimitError("slow down", retry_after=0.001) + return "ok" + + start = time.monotonic() + result = await rate_limited_then_ok() + elapsed = time.monotonic() - start + assert result == "ok" + assert elapsed < 1.0 # honored the short retry_after, not the huge base_delay + + +# ---- Common Data Schema ---- + + +def test_conversation_schema_roundtrip(): + conv = Conversation( + id="c1", + source_connector="zalo", + channel=ChannelType.ZALO, + external_id="zalo-user-123", + messages=[ + Message(id="m1", conversation_id="c1", direction="inbound", sender_id="zalo-user-123", text="hi"), + ], + ) + dumped = conv.model_dump() + assert dumped["channel"] == "zalo" + assert dumped["messages"][0]["text"] == "hi"