即時同步：在使用者事件中呼叫

您的產品是 Web 服務時，在新增／修改／停用使用者的事件中即時呼叫 SCIM，讓 MaiAgent 組織成員與您的系統隨時一致；附可直接引用的 client 與接法

適用情境：您的產品本身就是 Web 服務、有自己的使用者管理（ISV 的 SaaS、企業自建入口網站），使用者的新增、修改、停權都發生在您的系統裡。最直接的做法是在那一刻順手呼叫 SCIM 開通 API，MaiAgent 的成員狀態就和您的系統即時一致，不用等排程。

一、client

您的產品本身就是 Web 服務、有自己的使用者管理時，最直接的做法是在新增／修改／停用使用者的那一刻順手呼叫 SCIM，成員狀態就和您的系統即時一致。可直接引用，四個方法對應四個事件；把回傳的 MaiAgent 成員 ID 存進您的使用者表，之後更新與停用就不必再查詢。已在 MaiAgent dev 環境實跑：建立 → 重複建立回同一 ID → 更新改名換角色 → 停用 → 重新啟用。

"""maiagent_members.py：在您的使用者事件中即時同步 MaiAgent 組織成員（SCIM）"""
import os
import urllib.parse

import requests

USER_SCHEMA = "urn:ietf:params:scim:schemas:core:2.0:User"
GROUP_SCHEMA = "urn:ietf:params:scim:schemas:core:2.0:Group"
PATCH_SCHEMA = "urn:ietf:params:scim:api:messages:2.0:PatchOp"


class MaiAgentMembers:
    def __init__(self, scim_url=None, token=None):
        self.base = (scim_url or os.environ["MAIAGENT_SCIM_URL"]).rstrip("/")
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {token or os.environ['MAIAGENT_SCIM_TOKEN']}",
            "Content-Type": "application/scim+json",
            "Accept": "application/scim+json",
        })

    def _request(self, method, path, **kwargs):
        res = self.session.request(method, f"{self.base}{path}", timeout=30, **kwargs)
        res.raise_for_status()
        return res.json() if res.content else None

    def _find(self, resource, attr, value):
        query = urllib.parse.quote(f'{attr} eq "{value}"')
        resources = self._request("GET", f"/{resource}?filter={query}").get("Resources", [])
        return resources[0] if resources else None

    def ensure_group(self, name):
        """角色不存在就建立（權限為空，要到後台補）；已存在（後台先建好）就 PATCH 同名一次讓 SCIM 建立對應。回傳角色 ID。"""
        group = self._find("Groups", "displayName", name)
        if group is None:
            return self._request("POST", "/Groups", json={"schemas": [GROUP_SCHEMA], "displayName": name})["id"]
        self._request("PATCH", f"/Groups/{group['id']}", json={
            "schemas": [PATCH_SCHEMA],
            "Operations": [{"op": "replace", "path": "displayName", "value": name}],
        })
        return group["id"]

    # --- 使用者事件 -------------------------------------------------------

    def on_user_created(self, external_id, email, name, groups=()):
        """回傳 MaiAgent 成員 ID，請存進您的使用者表。同 Email 已是成員時會接管而不重複建立。"""
        for group in groups:
            self.ensure_group(group)
        payload = self._payload(external_id, email, name, groups)
        existing = self._find("Users", "externalId", external_id) or self._find("Users", "userName", email)
        if existing:
            return self._request("PUT", f"/Users/{existing['id']}", json=payload)["id"]
        return self._request("POST", "/Users", json=payload)["id"]

    def on_user_updated(self, member_id, external_id, email, name, groups=()):
        """整筆同步（改名、換角色）。Email 不能經 SCIM 變更：傳入不同的 Email 會被 MaiAgent 忽略，見「事件對應表」。"""
        for group in groups:
            self.ensure_group(group)
        self._request("PUT", f"/Users/{member_id}", json=self._payload(external_id, email, name, groups))

    def on_user_deactivated(self, member_id):
        """離職／停權：停用而非刪除，對話與額度紀錄保留，API Key 立即撤銷。"""
        self._set_active(member_id, False)

    def on_user_reactivated(self, member_id):
        self._set_active(member_id, True)

    # --- 內部 -------------------------------------------------------------

    def _set_active(self, member_id, active):
        self._request("PATCH", f"/Users/{member_id}", json={
            "schemas": [PATCH_SCHEMA],
            "Operations": [{"op": "replace", "path": "active", "value": active}],
        })

    @staticmethod
    def _payload(external_id, email, name, groups):
        return {
            "schemas": [USER_SCHEMA],
            "userName": email,
            "externalId": external_id,
            "displayName": name,
            "active": True,
            "groups": [{"display": g} for g in groups],
        }

二、接進您的事件

例如 Django signal、Rails callback、或使用者服務的 create／update／delete 方法尾端：

members = MaiAgentMembers()

def handle_user_created(user):                       # 您系統的新增事件
    user.maiagent_member_id = members.on_user_created(
        external_id=user.id, email=user.email, name=user.display_name, groups=[user.department],
    )
    user.save(update_fields=["maiagent_member_id"])

def handle_user_updated(user):                       # 改 Email／姓名／部門
    members.on_user_updated(user.maiagent_member_id, user.id, user.email, user.display_name, [user.department])

def handle_user_deleted(user):                       # 刪除或停權都對應「停用」
    members.on_user_deactivated(user.maiagent_member_id)

三、首次上線：把既有使用者批次補建

事件同步只會處理接上之後發生的異動；接上前就存在的使用者要補建一次。不用另外寫工具，把 on_user_created 對既有使用者跑一輪即可——它是冪等的（同 externalId 或同 Email 只會接管、不會重複建），而且會把 MaiAgent 成員 ID 寫回您的使用者表，之後的更新與停用就直接可用：

"""backfill_members.py：首次上線把既有使用者補建成 MaiAgent 成員（可重複執行、可中斷續跑）"""
import time

from maiagent_members import MaiAgentMembers

members = MaiAgentMembers()

for user in User.objects.filter(is_active=True, maiagent_member_id__isnull=True).iterator():
    user.maiagent_member_id = members.on_user_created(
        external_id=user.id, email=user.email, name=user.display_name, groups=[user.department],
    )
    user.save(update_fields=["maiagent_member_id"])
    time.sleep(0.7)  # SCIM 每組織每分鐘 100 次；每人約 2 次呼叫，控制在限流內

• 速度：限流下約每秒 1 人，1,000 人約 20 分鐘；放到背景任務跑，不要在請求裡跑。

• 可中斷續跑：只挑 maiagent_member_id 還是空的人，中途失敗重跑就會接著做；已建好的人再跑一次也只會回同一個 ID。

• 先在幾個人身上試：加個 [:5] 先跑五個，到後台「成員管理」確認姓名、角色都對，再全量。

• 順序：先補建、再打開事件同步；補建期間若有人異動，事件同步接上後對該使用者再呼叫一次即可，結果一樣。

• 不想寫程式也行：把既有使用者匯出成名單交給名單同步工具跑一次，之後再用 filter=externalId eq "…" 把成員 ID 撈回來存；人數多時上面的迴圈比較省事。

四、實務建議

• 角色先在後台建好、權限設好：SCIM 建的成員不會進預設角色，ensure_group 自動建立的角色權限是空的；先建好角色（含 MaiGPT 使用權）再用相同名稱，成員一建好就能用。

• 事件非同步化、失敗要重試：把呼叫放進您的任務佇列（Celery、Sidekiq、SQS…）而不是請求主線程，SCIM 端偶發 429／5xx 時重試即可；四個方法都是冪等的，重複執行沒有副作用。

• 您系統的「刪除」對應 MaiAgent 的「停用」：不要真的刪，之後回鍋能直接 on_user_reactivated，歷史對話也還在。

• 每天再對帳一次：事件可能漏（部署時、佇列失敗）。把名單同步工具排程每天跑一次當保險，它會把漏掉的差異補齊；兩者用同一個 externalId，不會互相打架。

五、事件對應表

您系統的事件	呼叫	MaiAgent 這邊發生什麼
新增使用者	on_user_created	建立帳號（無密碼）與成員、掛角色；回成員 ID
改姓名／部門	on_user_updated	整筆同步姓名與角色
改 Email	不能經 SCIM 改（會被忽略）	由管理員在後台修改；或 on_user_deactivated 舊成員後以新 Email on_user_created（歷史不延續）
停權、刪除、離職	on_user_deactivated	停用：API Key 撤銷、角色權限失效、換發 token 回 404；資料保留
復職	on_user_reactivated	重新啟用，角色恢復

錯誤碼與處理見名單同步工具的常見錯誤對照，兩者共用同一套 API。