リアルタイム同期：ユーザーイベントからの呼び出し

製品が Web サービスの場合、ユーザーの追加・変更・無効化イベントで SCIM をリアルタイムに呼び出し、MaiAgent の組織メンバーとお客様のシステムを常に同期します。そのまま利用できるクライアントと組み込み方法も掲載しています

お客様の製品自体が Web サービスであり、独自のユーザー管理機能を備えている場合（ISV の SaaS、企業が独自に構築したポータルなど）に適しています。ユーザーの追加・変更・無効化がすべてお客様のシステム内で発生する場合、最も直接的な方法は、そのタイミングで SCIM プロビジョニング API を呼び出すことです。これにより、スケジュールされた処理を待つことなく、MaiAgent のメンバー状態をお客様のシステムとリアルタイムに同期できます。

1. クライアント

お客様の製品自体が Web サービスであり、独自のユーザー管理機能を備えている場合、最も直接的な方法は、ユーザーを追加・変更・無効化したそのタイミングで SCIM を呼び出すことです。これにより、メンバーの状態をお客様のシステムとリアルタイムに同期できます。以下のクライアントはそのまま利用でき、4 つのメソッドが 4 つのイベントに対応しています。返された 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],
        }

2. イベントへの組み込み

たとえば、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)

3. 初回導入：既存ユーザーを一括で追加する

イベント同期で処理されるのは、接続後に発生した変更だけです。接続前から存在するユーザーは、一度追加する必要があります。別のツールを作成する必要はなく、既存ユーザーごとに 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 は組織ごとに 1 分あたり 100 回までです。1 人につき約 2 回呼び出すため、レート制限内に収めます。

• 速度：レート制限内では 1 秒あたり約 1 人、1,000 人なら約 20 分です。リクエスト内ではなく、バックグラウンドタスクとして実行してください。

• 中断後に再開可能：maiagent_member_id が空のユーザーだけを選択します。途中で失敗した場合は、再実行すると続きから処理されます。作成済みのユーザーに対して再実行しても、同じ ID が返るだけです。

• まず数人で試す：[:5] を追加して最初に 5 人だけ処理し、管理画面の「メンバー管理」で名前とロールが正しいことを確認してから、全員を処理してください。

• 順序：先に一括追加を完了してから、イベント同期を有効にしてください。一括追加中にユーザー情報が変更された場合は、イベント同期を接続した後にそのユーザーについてもう一度呼び出せば、結果は同じになります。

• コードを書かない方法：既存ユーザーをリストとしてエクスポートし、メンバーリスト同期ツールを一度実行します。その後、filter=externalId eq "…" を使ってメンバー ID を取得し、保存します。人数が多い場合は、上記のループの方が手間を減らせます。

4. 実運用での推奨事項

• ロールを管理画面で事前に作成し、権限を設定する：SCIM で作成したメンバーはデフォルトロールに入りません。また、ensure_group が自動作成したロールの権限は空です。MaiGPT の利用権限を含めてロールを事前に作成し、同じ名前を使用すると、メンバーは作成直後から利用できます。

• イベントを非同期で処理し、失敗時に再試行する：呼び出しはメインのリクエストスレッドではなく、Celery、Sidekiq、SQS などのタスクキューに入れてください。SCIM エンドポイントが一時的に 429 または 5xx を返した場合は再試行します。4 つのメソッドはいずれも冪等であるため、繰り返し実行しても副作用はありません。

• お客様のシステムでの「削除」を MaiAgent の「無効化」に対応させる：実際には削除しないでください。ユーザーが復帰した場合は、そのまま on_user_reactivated を呼び出せます。過去の会話も残っています。

• 1 日 1 回照合する：デプロイ時やキューの失敗によってイベントが失われる場合があります。保険として、メンバーリスト同期ツールを 1 日 1 回実行するようにスケジュールしてください。見落とした差分が補完されます。どちらも同じ externalId を使用するため、競合しません。

5. イベント対応表

お客様のシステムでのイベント	呼び出し	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 を使用します。