Real-Time Synchronization: Calls from User Events

If your product is a web service, call SCIM in real time when users are created, updated, or deactivated to keep MaiAgent organization members synchronized with your system at all times. Includes a re

Use this approach when your product is itself a web service with its own user management, such as an ISV SaaS product or an enterprise-built portal, and user creation, updates, and deactivation all occur in your system. The most direct solution is to call the SCIM provisioning API at that moment. This keeps MaiAgent member status synchronized with your system in real time without waiting for a scheduled job.

1. Client

If your product is itself a web service with its own user management, the most direct solution is to call SCIM when a user is created, updated, or deactivated. This keeps member status synchronized with your system in real time. You can use the following client directly. Its four methods correspond to four events. Save the returned MaiAgent member ID in your user table so subsequent updates and deactivation do not require another lookup. This flow has been tested in the MaiAgent dev environment: create → create again and receive the same ID → update the name and role → deactivate → reactivate.

"""maiagent_members.py: Synchronize MaiAgent organization members in real time from your user events (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):
        """Create the role if it does not exist (it will have no permissions, so add them in the admin console). If it already exists (created in the admin console), PATCH it once with the same name to create its SCIM mapping. Return the role 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"]

    # --- User events -----------------------------------------------------

    def on_user_created(self, external_id, email, name, groups=()):
        """Return the MaiAgent member ID. Save it in your user table. If a member with the same email already exists, take ownership of that member instead of creating a duplicate."""
        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=()):
        """Synchronize the complete record (name and role changes). Email cannot be changed through SCIM: MaiAgent ignores a different email passed here. See the event mapping table."""
        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):
        """Deactivate instead of deleting on departure or suspension. Conversation and quota records are retained, and API keys are revoked immediately."""
        self._set_active(member_id, False)

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

    # --- Internal --------------------------------------------------------

    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. Connect the client to your events

For example, add it at the end of a Django signal, Rails callback, or your user service's create, update, and delete methods:

members = MaiAgentMembers()

def handle_user_created(user):                       # Create event in your system
    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):                       # Change email, name, or department
    members.on_user_updated(user.maiagent_member_id, user.id, user.email, user.display_name, [user.department])

def handle_user_deleted(user):                       # Both deletion and suspension map to deactivation
    members.on_user_deactivated(user.maiagent_member_id)

3. Initial rollout: Batch-provision existing users

Event synchronization only processes changes that occur after it is connected. Users who already existed must be provisioned once. You do not need a separate tool—simply run on_user_created for each existing user. It is idempotent: a matching externalId or email takes ownership of the existing account instead of creating a duplicate. It also writes the MaiAgent member ID back to your user table, so subsequent updates and deactivation can use it directly:

"""backfill_members.py: Provision existing users as MaiAgent members during initial rollout (restartable and safe to run repeatedly)."""
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 allows 100 requests per organization per minute. Each user requires about two calls, so stay within the rate limit.

• Speed: At the rate limit, the process handles about one person per second, or 1,000 people in about 20 minutes. Run it as a background task, not inside a request.

• Restartable: Only users whose maiagent_member_id is still empty are selected. If the process fails, rerun it to continue. Running it again for an existing user simply returns the same ID.

• Test with a few users first: Add [:5] to process five users, then verify their names and roles under Member Management in the admin console before processing everyone.

• Order: Complete the backfill before enabling event synchronization. If a user changes during the backfill, call the event synchronization method for that user once after it is enabled; the result will be the same.

• No-code alternative: Export existing users as a list and run the roster synchronization tool once. Then retrieve each member ID with filter=externalId eq "…" and save it. For large numbers of users, the loop above requires less effort.

4. Practical recommendations

• Create roles and configure their permissions in the admin console first: Members created through SCIM are not assigned to the default role, and roles created automatically by ensure_group have no permissions. Create each role first, including MaiGPT access, and use the same name so members can use it immediately after they are created.

• Process events asynchronously and retry failures: Put calls in your task queue, such as Celery, Sidekiq, or SQS, instead of the main request thread. Retry if the SCIM endpoint occasionally returns 429 or 5xx. All four methods are idempotent, so repeated execution has no side effects.

• Map “delete” in your system to “deactivate” in MaiAgent: Do not actually delete the member. If the user returns, call on_user_reactivated directly; their conversation history will still be available.

• Reconcile once a day: Events can be missed during deployments or queue failures. Schedule the roster synchronization tool to run once a day as a safeguard. It fills in any missed differences. Both approaches use the same externalId, so they do not conflict.

5. Event mapping table

Event in your system	Call	Result in MaiAgent
Create user	on_user_created	Creates an account (without a password) and member, assigns roles, and returns the member ID
Change name or department	on_user_updated	Synchronizes the complete name and role record
Change email	Cannot be changed through SCIM (the change is ignored)	An administrator must change it in the admin console, or call on_user_deactivated for the old member and then on_user_created with the new email (history does not carry over)
Suspend, delete, or offboard	on_user_deactivated	Deactivates the member: revokes API keys, invalidates role permissions, and causes token issuance to return 404; data is retained
Reinstate	on_user_reactivated	Reactivates the member and restores roles

For error codes and handling instructions, see the common error reference for the roster synchronization tool. Both tools use the same API.