名單同步工具：CSV／JSON／資料庫

沒有 IdP 也能自動開通成員：一支現成的名單同步工具，吃 CSV、JSON 或從資料庫匯出的名單，透過 SCIM 建立、更新、停用組織成員，可排程執行

適用情境：客戶沒有 Okta、Entra ID 這類會自動推送人員的 IdP，人員名單在 HR 系統、Excel、或您自家的資料庫裡。這支工具把名單同步到 MaiAgent 組織成員——名單上有的建立或更新、名單上消失的自動停用——底層走SCIM 開通 API，您不需要自己處理 API 細節。

跑一次是一次同步；排程每天跑或在人事異動時觸發即可。

一、準備名單

名單只有四個欄位。id 是您系統裡不會變的人員編號（員工編號、HR 系統 ID），工具用它認人；groups 填 MaiAgent 的角色名稱。離職的人從名單拿掉即可，工具會停用他。

角色請先在後台建好、權限設好（含 MaiGPT 使用權），名單再用相同名稱。工具遇到不存在的角色會自動建立，但建出來的角色權限是空的，成員會「建好了什麼都不能用」，得回後台補權限。

roster.csv，groups 多個角色用 ; 分隔：

id,email,name,groups
emp-10023,alice@acme.com,Alice Chen,客服
emp-10024,bob@acme.com,Bob Lin,客服;主管
emp-10031,carol@acme.com,Carol Wu,

roster.json，groups 是陣列：

[
  { "id": "emp-10023", "email": "alice@acme.com", "name": "Alice Chen", "groups": ["客服"] },
  { "id": "emp-10024", "email": "bob@acme.com",   "name": "Bob Lin",    "groups": ["客服", "主管"] },
  { "id": "emp-10031", "email": "carol@acme.com", "name": "Carol Wu",   "groups": [] }
]

名單在資料庫時，用一句 SQL 匯出成上面的 JSON 再交給工具，資料庫種類不限。以 PostgreSQL 為例（MySQL 換成 pymysql，SQL 同理）：

"""export_roster.py：從 HR 資料庫匯出在職名單成 roster.json"""
import json
import os

import psycopg  # pip install "psycopg[binary]"

SQL = """
SELECT e.employee_no                    AS id,
       e.email                          AS email,
       e.full_name                      AS name,
       COALESCE(array_agg(r.role_name) FILTER (WHERE r.role_name IS NOT NULL), '{}') AS groups
FROM   employees e
LEFT   JOIN employee_roles r ON r.employee_no = e.employee_no
WHERE  e.status = 'active'
GROUP  BY e.employee_no, e.email, e.full_name
"""

with psycopg.connect(os.environ["HR_DATABASE_URL"]) as conn, conn.cursor() as cur:
    cur.execute(SQL)
    rows = [{"id": i, "email": e, "name": n, "groups": list(g)} for i, e, n, g in cur.fetchall()]

with open("roster.json", "w", encoding="utf-8") as f:
    json.dump(rows, f, ensure_ascii=False, indent=2)
print(f"匯出 {len(rows)} 人")

排程時兩步串起來：python export_roster.py && python maiagent_scim_sync.py roster.json。只匯出在職者（status = 'active'），離職者就會被工具停用。

二、執行

設定連線資訊

到後台「組織設定 → 身份同步」選 SCIM，在「連線資訊」分頁複製 Endpoint URL 與 Token（取得方式與權限見啟用與取得 Token），設成環境變數。Token 等同組織的人事管理權限，只能放在跑這支工具的伺服器上：

export MAIAGENT_SCIM_URL="https://api.maiagent.ai/api/v1/scim/<組織 ID>"
export MAIAGENT_SCIM_TOKEN="<SCIM Token>"
pip install requests

先看它會做什麼

python maiagent_scim_sync.py roster.csv --dry-run     # 或 roster.json

只印計畫、不動任何資料：

[group]      建立角色 主管（權限為空，請到後台角色權限設定）
[create]     emp-10023 alice@acme.com  groups=客服
[update]     emp-10024 bob@acme.com    groups=客服;主管
[email]      emp-10031 carol.new@acme.com  groups=  ⚠ Email 不可經 SCIM 變更（MaiAgent 仍是 carol@acme.com），請在後台修改或停用舊帳號後以新 Email 建立
[deactivate] emp-10009 dave@acme.com   （不在名單上）
[skip]       owner@acme.com            （非本工具建立，不處理）
計畫：建立 1、更新 1、停用 1、不變 0、略過 1、需人工 1

真的跑，然後排程

拿掉 --dry-run 執行；跑完到後台「成員管理」應該看得到這些人。再跑一次應該全部是 [unchanged]——這就是冪等，可以放心排程，例如每天早上：

0 7 * * * cd /opt/maiagent-sync && python maiagent_scim_sync.py roster.csv >> sync.log 2>&1

三、工具本體

maiagent_scim_sync.py，Python 3.10+，只依賴 requests；名單副檔名是 .csv 或 .json 都吃。已在 MaiAgent dev 環境實跑驗證（建立、指派後台既有角色與自動建立角色、重跑不變、移除即停用、Email 變更警示）。

"""把名單（CSV 或 JSON）同步到 MaiAgent 組織成員（SCIM）。用法：python maiagent_scim_sync.py roster.csv|roster.json [--dry-run]"""
import csv
import json
import os
import sys
import time
import urllib.parse

import requests

SCIM_URL = os.environ["MAIAGENT_SCIM_URL"].rstrip("/")
HEADERS = {
    "Authorization": f"Bearer {os.environ['MAIAGENT_SCIM_TOKEN']}",
    "Content-Type": "application/scim+json",
    "Accept": "application/scim+json",
}
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"


def scim(method, path, **kwargs):
    """呼叫 SCIM 端點；撞到每分鐘限流（429）就等一下再試。"""
    for attempt in range(5):
        res = requests.request(method, f"{SCIM_URL}{path}", headers=HEADERS, timeout=30, **kwargs)
        if res.status_code != 429:
            return res
        time.sleep(2 ** attempt)
    res.raise_for_status()


def read_roster(path):
    """讀 CSV（groups 以 ; 分隔）或 JSON（groups 為陣列），回傳 [{id, email, name, groups}]。"""
    if path.lower().endswith(".json"):
        with open(path, encoding="utf-8") as f:
            rows = json.load(f)
    else:
        with open(path, newline="", encoding="utf-8-sig") as f:
            rows = list(csv.DictReader(f))
    roster = []
    for row in rows:
        groups = row.get("groups") or []
        if isinstance(groups, str):
            groups = [g.strip() for g in groups.split(";") if g.strip()]
        roster.append({"id": str(row["id"]).strip(), "email": row["email"].strip(), "name": (row.get("name") or "").strip(), "groups": groups})
    return roster


def list_all_members():
    members, start = [], 1
    while True:
        page = scim("GET", f"/Users?startIndex={start}&count=100").json()
        resources = page.get("Resources", [])
        members.extend(resources)
        if not resources or start + len(resources) > page.get("totalResults", 0):
            return members
        start += len(resources)


def find_by_email(email):
    query = urllib.parse.quote(f'userName eq "{email}"')
    resources = scim("GET", f"/Users?filter={query}").json().get("Resources", [])
    return resources[0] if resources else None


def ensure_groups(names, dry_run):
    """名單用到的角色先確保存在且已對應到 SCIM：
    - 不存在 → POST /Groups 建立（權限是空的，要到後台角色權限補）
    - 已存在（例如後台先建好、權限已設定）→ PATCH 同名一次，讓 SCIM 建立對應，之後 User 的 groups[] 才掛得上
    """
    for name in sorted(names):
        query = urllib.parse.quote(f'displayName eq "{name}"')
        found = scim("GET", f"/Groups?filter={query}").json().get("Resources")
        if found:
            if not dry_run:
                scim("PATCH", f"/Groups/{found[0]['id']}", json={"schemas": [PATCH_SCHEMA], "Operations": [{"op": "replace", "path": "displayName", "value": name}]}).raise_for_status()
            continue
        print(f"[group]      建立角色 {name}（權限為空，請到後台角色權限設定）")
        if not dry_run:
            scim("POST", "/Groups", json={"schemas": [GROUP_SCHEMA], "displayName": name}).raise_for_status()


def to_payload(person):
    return {
        "schemas": [USER_SCHEMA],
        "userName": person["email"],
        "externalId": person["id"],
        "displayName": person["name"],
        "active": True,
        "groups": [{"display": g} for g in person["groups"]],
    }


def same_as(member, person):
    current_groups = sorted(g["display"] for g in member.get("groups", []))
    return (
        member.get("userName", "").lower() == person["email"].lower()
        and member.get("displayName", "") == person["name"]
        and member.get("active") is True
        and current_groups == sorted(person["groups"])
    )


def main(roster_path, dry_run):
    roster = read_roster(roster_path)
    ensure_groups({g for person in roster for g in person["groups"]}, dry_run)
    existing = list_all_members()
    by_external_id = {m["externalId"]: m for m in existing if m.get("externalId")}
    counts = {"create": 0, "update": 0, "deactivate": 0, "unchanged": 0, "skip": 0, "warn": 0}

    for person in roster:
        member = by_external_id.get(person["id"]) or find_by_email(person["email"])
        label = f"{person['id']} {person['email']}  groups={';'.join(person['groups'])}"
        if member is None:
            counts["create"] += 1
            print(f"[create]     {label}")
            if not dry_run:
                scim("POST", "/Users", json=to_payload(person)).raise_for_status()
        elif member.get("userName", "").lower() != person["email"].lower() and member.get("externalId") == person["id"]:
            counts["warn"] += 1
            print(f"[email]      {label}  ⚠ Email 不可經 SCIM 變更（MaiAgent 仍是 {member['userName']}），請在後台修改或停用舊帳號後以新 Email 建立")
        elif same_as(member, person) and member.get("externalId") == person["id"]:
            counts["unchanged"] += 1
            print(f"[unchanged]  {label}")
        else:
            counts["update"] += 1
            print(f"[update]     {label}")
            if not dry_run:
                scim("PUT", f"/Users/{member['id']}", json=to_payload(person)).raise_for_status()

    wanted = {person["id"] for person in roster}
    for member in existing:
        if not member.get("externalId"):
            counts["skip"] += 1
            print(f"[skip]       {member.get('userName')}  （非本工具建立，不處理）")
        elif member["externalId"] not in wanted and member.get("active"):
            counts["deactivate"] += 1
            print(f"[deactivate] {member['externalId']} {member.get('userName')}  （不在名單上）")
            if not dry_run:
                scim(
                    "PATCH",
                    f"/Users/{member['id']}",
                    json={"schemas": [PATCH_SCHEMA], "Operations": [{"op": "replace", "path": "active", "value": False}]},
                ).raise_for_status()

    mode = "計畫" if dry_run else "完成"
    print(f"{mode}：建立 {counts['create']}、更新 {counts['update']}、停用 {counts['deactivate']}、不變 {counts['unchanged']}、略過 {counts['skip']}、需人工 {counts['warn']}")


if __name__ == "__main__":
    args = [a for a in sys.argv[1:] if not a.startswith("--")]
    if len(args) != 1:
        sys.exit("用法：python maiagent_scim_sync.py roster.csv|roster.json [--dry-run]")
    main(args[0], dry_run="--dry-run" in sys.argv)

四、它怎麼運作、為什麼安全

只動它自己建的人：以 externalId（您的人員編號）當鍵；後台手動邀請進來、沒有 externalId 的成員一律 [skip]，不會被誤停用。同一個 Email 已經是成員（例如先前用邀請加入）時，會補上 externalId 接管，不會建重複。
停用不是刪除：離職者的對話與額度紀錄保留，只是不能再用、API Key 被撤銷；重新加回名單就重新啟用。
角色先確保存在且已對應：後台先建好的角色，工具會 PATCH 一次同名讓 SCIM 建立對應再指派；不存在的角色才 POST /Groups 建立（權限為空）。成員只會進名單指定的角色、不會進預設角色；要用 MaiGPT，角色需具 MaiGPT 使用權。
Email 不可經 SCIM 變更：名單裡某人的 Email 變了，工具會印 [email] 警示並跳過，請在後台修改，或把舊的人從名單拿掉（停用）、以新 Email 當新人加入。
Email 是登入帳號：SCIM 建的帳號沒有密碼；員工要登入 MaiAgent 網頁需另設 SSO，只做成員身分嵌入則不需登入。
不是 Python 也一樣：每個動作只是幾支 HTTP 呼叫（GET /Users?filter=、POST /Users、PUT／PATCH /Users/{id}、GET／POST /Groups），照這支的順序用任何語言重寫即可；API 細節見SCIM 開通 API。

五、常見錯誤對照

回應

原因

處理

401

Token 錯、已重新產生、或組織 ID 與 Token 不符

到後台「連線資訊」確認 Endpoint URL 與 Token 是同一組織的

403／找不到頁面

組織方案未含身分同步

洽業務或對接窗口

400 userName or emails is required

名單少了 Email

補上 email 欄位

409 already exists

同 Email 已是本組織成員

工具已處理（改用 PUT 接管）；自己呼叫 API 時用 filter=userName eq "…" 找到後改 PUT／PATCH

404 on /Users/{id}

id 不是 MaiAgent 成員 ID（常見是誤用 externalId）

先用 filter=externalId eq "…" 換成成員 ID

429

每組織每分鐘 100 次

工具會自動退避；自己呼叫時批次控制在每秒 1 次左右

415

Content-Type 不是 application/scim+json（application/json 也接受）

改 header

角色沒掛上

角色不存在，或後台建的角色尚未與 SCIM 建立對應（User 的 groups[] 只對應已對應的角色）

先 POST /Groups，或對既有角色 PATCH 同名 displayName；工具已內建

成員建好但什麼都不能用

沒進任何角色，或角色權限是空的

後台角色權限設定（含 MaiGPT 使用權）；建議先建好角色再同步

改了 Email 沒生效

SCIM 不允許變更 userName

後台修改，或停用舊帳號以新 Email 建立

建了成員但登入不了

SCIM 建的帳號沒有密碼

設定 SSO，或只用成員身分嵌入不登入

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

下一頁SSO 與身分整合總覽

最後更新於 2 天前

這有幫助嗎？