Task Execution

Poll for provisioning tasks, execute them in your external system, and report results back to Iden.

When administrators make changes in Iden -- granting access, revoking licenses, resetting passwords -- Iden creates provisioning tasks for your connector to execute. Your connector polls for pending tasks, performs the action in the external system, and reports back success or failure.

What Are Tasks?

Tasks are provisioning actions that Iden schedules for your connector. Each task represents a single operation to perform in the external application:

Action	Description
`create_account`	Create a new user account
`update_account`	Modify an existing account's attributes
`delete_account`	Remove or deprovision an account
`suspend_account`	Temporarily disable an account
`unsuspend_account`	Reactivate a previously suspended account
`reset_password`	Reset the account's password

Iden creates these tasks automatically based on governance decisions -- access reviews, approval workflows, manual admin actions, and automated policies. Your connector does not need to know why a task was created, only how to execute it.

Task Statuses

Each task has one of four statuses:

Status	Description
`pending`	Ready for your connector to pick up and execute
`scheduled`	Has a future `execute_after` date. Will become `pending` once that date passes
`completed`	Successfully executed by your connector
`failed`	Execution failed. Includes an error code and message

Your connector should only pick up tasks with status=pending. Scheduled tasks are not yet ready for execution -- Iden will transition them to pending when their execute_after date arrives.

Polling for Tasks

Fetch pending tasks by calling the tasks endpoint with status=pending.

curl:

curl "https://developer.idenhq.com/org/acme/api/v1/bridge/apps/${APP_ID}/tasks/?status=pending" \
  -H "Authorization: Api-Key <your-api-key>"

Python:

import requests

BASE_URL = "https://developer.idenhq.com/org/acme"
HEADERS = {
    "Authorization": "Api-Key <your-api-key>",
    "Content-Type": "application/json",
}

response = requests.get(
    f"{BASE_URL}/api/v1/bridge/apps/{app_id}/tasks/",
    headers=HEADERS,
    params={"status": "pending"},
)
response.raise_for_status()
tasks = response.json()["tasks"]
print(f"Found {len(tasks)} pending tasks")

Response:

{
  "tasks": [
    {
      "id": "t1a2b3c4-...",
      "action": "create_account",
      "resource_type": "accounts",
      "status": "pending",
      "execute_after": "2025-01-15T10:00:00Z",
      "created_at": "2025-01-15T09:55:00Z",
      "payload": {
        "account": {
          "id": "ext-123",
          "email": "jane@acme.com",
          "first_name": "Jane",
          "last_name": "Doe"
        }
      }
    }
  ]
}

Polling interval

Poll every 30 to 60 seconds for most use cases. If your application needs faster provisioning, you can poll as frequently as every 10 seconds. Avoid polling more aggressively than that -- it will not improve responsiveness and adds unnecessary load.

Task Payload Format

The payload field contains the data your connector needs to execute the action. The shape varies by action type.

`create_account`

Create a new user in the external system.

{
  "account": {
    "id": "ext-123",
    "email": "jane@acme.com",
    "first_name": "Jane",
    "last_name": "Doe",
    "username": "jane.doe"
  },
  "memberships": {
    "team": [
      {"id": "g1", "name": "Engineering"}
    ]
  }
}

Field	Type	Description
`account`	object	The account to create
`account.id`	string	The desired external ID for the new account
`account.email`	string	Email address
`account.first_name`	string	First name
`account.last_name`	string	Last name
`account.username`	string	Username (if applicable)
`memberships`	object	Group memberships to assign, keyed by resource type slug. Each value is an array of `{id, name}` references

`update_account`

Update memberships or assignments on an existing account. The changes object is keyed by resource type slug, with each value containing added and removed arrays of {id, name} references.

{
  "account": {
    "id": "ext-123",
    "email": "jane@acme.com",
    "first_name": "Jane",
    "last_name": "Doe"
  },
  "changes": {
    "team": {
      "added": [
        {"id": "g2", "name": "Design"}
      ],
      "removed": [
        {"id": "g1", "name": "Engineering"}
      ]
    }
  }
}

Field	Type	Description
`account`	object	The account to update
`account.id`	string	The external ID of the account to update
`changes`	object	Membership/assignment changes keyed by resource type slug. Each value has `added` and `removed` arrays of `{id, name}` references

`delete_account`

Remove or deprovision an account.

{
  "account": {
    "id": "ext-123"
  }
}

Field	Type	Description
`account`	object	The account to delete
`account.id`	string	The external ID of the account to delete

`suspend_account`

Temporarily disable an account without deleting it.

{
  "account": {
    "id": "ext-123"
  }
}

Field	Type	Description
`account`	object	The account to suspend
`account.id`	string	The external ID of the account to suspend

`unsuspend_account`

Reactivate a previously suspended account.

{
  "account": {
    "id": "ext-123"
  }
}

Field	Type	Description
`account`	object	The account to reactivate
`account.id`	string	The external ID of the account to reactivate

`reset_password`

Reset the account's password. The connector should generate a new password or trigger the external system's password reset flow.

{
  "account": {
    "id": "ext-123"
  }
}

Field	Type	Description
`account`	object	The account whose password should be reset
`account.id`	string	The external ID of the account whose password should be reset

Reporting Success

After executing a task, report success by patching the task status to completed.

curl:

curl -X PATCH "https://developer.idenhq.com/org/acme/api/v1/bridge/apps/${APP_ID}/tasks/${TASK_ID}/status/" \
  -H "Authorization: Api-Key <your-api-key>" \
  -H "Content-Type: application/json" \
  -d '{"status": "completed", "result": {}}'

Python:

response = requests.patch(
    f"{BASE_URL}/api/v1/bridge/apps/{app_id}/tasks/{task_id}/status/",
    headers=HEADERS,
    json={"status": "completed", "result": {}},
)
response.raise_for_status()
print(f"Task {task_id} marked as completed")

The result field is an empty object for now. It is reserved for future use (e.g., returning the newly created account ID on create_account).

Reporting Failure

If your connector cannot execute a task, report failure with an error code and a human-readable message.

curl:

curl -X PATCH "https://developer.idenhq.com/org/acme/api/v1/bridge/apps/${APP_ID}/tasks/${TASK_ID}/status/" \
  -H "Authorization: Api-Key <your-api-key>" \
  -H "Content-Type: application/json" \
  -d '{
    "status": "failed",
    "error": {
      "code": "ACCOUNT_NOT_FOUND",
      "message": "No account with ID ext-123 exists in the target system."
    }
  }'

Python:

response = requests.patch(
    f"{BASE_URL}/api/v1/bridge/apps/{app_id}/tasks/{task_id}/status/",
    headers=HEADERS,
    json={
        "status": "failed",
        "error": {
            "code": "ACCOUNT_NOT_FOUND",
            "message": "No account with ID ext-123 exists in the target system.",
        },
    },
)
response.raise_for_status()
print(f"Task {task_id} marked as failed")

Error Codes

Use these standard error codes when reporting task failures. The code field in the error object must be one of:

Code	When to use
`ACCOUNT_NOT_FOUND`	The account does not exist in the target system. Use for `update_account`, `delete_account`, `suspend_account`, `unsuspend_account`, and `reset_password` when the target account is missing
`ACCOUNT_ALREADY_EXISTS`	The account already exists. Use for `create_account` when the external system rejects a duplicate
`LICENSE_EXHAUSTED`	No available license seats. Use when a `create_account` or `update_account` task requires a license that has no remaining capacity
`PERMISSION_DENIED`	The connector's credentials do not have sufficient permissions to perform the action
`RATE_LIMITED`	The external system rejected the request due to rate limiting. Iden may retry the task later
`TIMEOUT`	The operation timed out before completing
`INTERNAL_ERROR`	An unspecified error occurred. Use as a catch-all when no other code applies

Always include a descriptive message alongside the code. The message is shown to administrators in the Iden dashboard and helps them diagnose issues.

409 Conflict: Task Already in Terminal State

If you attempt to update a task that is already completed or failed, the API returns 409 Conflict. This can happen if:

Your connector processed the task but the status report was delayed, and Iden already marked it via another mechanism.
A retry in your connector attempts to report the same task twice.

Handle 409 gracefully -- the task is already done, so no further action is needed.

Python:

response = requests.patch(
    f"{BASE_URL}/api/v1/bridge/apps/{app_id}/tasks/{task_id}/status/",
    headers=HEADERS,
    json={"status": "completed", "result": {}},
)

if response.status_code == 409:
    print(f"Task {task_id} already in terminal state, skipping")
elif response.ok:
    print(f"Task {task_id} marked as completed")
else:
    response.raise_for_status()

Building a Task Execution Loop

Here is a complete Python example that continuously polls for tasks, executes them, and reports results. Adapt the execute_task function to perform the actual operations in your external system.

import time
import logging
import requests

BASE_URL = "https://developer.idenhq.com/org/acme"
HEADERS = {
    "Authorization": "Api-Key <your-api-key>",
    "Content-Type": "application/json",
}
POLL_INTERVAL = 30  # seconds

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("bridge-connector")


def execute_task(task):
    """
    Execute a provisioning task in the external system.
    Returns (True, result_dict) on success or (False, error_dict) on failure.

    Replace the placeholder logic below with your actual integration code.
    """
    action = task["action"]
    payload = task["payload"]
    account = payload.get("account", {})
    account_id = account.get("id")

    try:
        if action == "create_account":
            # TODO: Create the account in your external system
            # e.g., external_api.create_user(
            #     email=account["email"],
            #     first_name=account["first_name"],
            #     last_name=account["last_name"],
            # )
            # memberships = payload.get("memberships", {})
            logger.info(f"Created account {account_id}")
            return True, {}

        elif action == "update_account":
            # TODO: Apply the membership changes to the existing account
            # changes = payload["changes"]
            # e.g., for slug, diff in changes.items():
            #     for ref in diff["added"]:
            #         external_api.add_membership(account_id, slug, ref["id"])
            #     for ref in diff["removed"]:
            #         external_api.remove_membership(account_id, slug, ref["id"])
            logger.info(f"Updated account {account_id}")
            return True, {}

        elif action == "delete_account":
            # TODO: Delete or deprovision the account
            # e.g., external_api.delete_user(account_id)
            logger.info(f"Deleted account {account_id}")
            return True, {}

        elif action == "suspend_account":
            # TODO: Suspend the account
            # e.g., external_api.suspend_user(account_id)
            logger.info(f"Suspended account {account_id}")
            return True, {}

        elif action == "unsuspend_account":
            # TODO: Reactivate the account
            # e.g., external_api.unsuspend_user(account_id)
            logger.info(f"Unsuspended account {account_id}")
            return True, {}

        elif action == "reset_password":
            # TODO: Reset the account's password
            # e.g., external_api.reset_password(account_id)
            logger.info(f"Reset password for {account_id}")
            return True, {}

        else:
            return False, {
                "code": "INTERNAL_ERROR",
                "message": f"Unknown action: {action}",
            }

    except Exception as e:
        logger.exception(f"Failed to execute {action} for {account_id}")
        return False, {
            "code": "INTERNAL_ERROR",
            "message": str(e),
        }


def report_task_status(app_id, task_id, success, data):
    """Report task completion or failure to Iden."""
    if success:
        body = {"status": "completed", "result": data}
    else:
        body = {"status": "failed", "error": data}

    response = requests.patch(
        f"{BASE_URL}/api/v1/bridge/apps/{app_id}/tasks/{task_id}/status/",
        headers=HEADERS,
        json=body,
    )

    if response.status_code == 409:
        logger.warning(f"Task {task_id} already in terminal state")
    else:
        response.raise_for_status()
        logger.info(f"Reported {body['status']} for task {task_id}")


def poll_and_execute(app_id):
    """Main loop: poll for pending tasks and execute them."""
    logger.info(f"Starting task loop for app {app_id}")

    while True:
        try:
            response = requests.get(
                f"{BASE_URL}/api/v1/bridge/apps/{app_id}/tasks/",
                headers=HEADERS,
                params={"status": "pending"},
            )
            response.raise_for_status()
            tasks = response.json()["tasks"]

            if tasks:
                logger.info(f"Found {len(tasks)} pending tasks")

            for task in tasks:
                task_id = task["id"]
                logger.info(
                    f"Executing task {task_id}: {task['action']}"
                )
                success, data = execute_task(task)
                report_task_status(app_id, task_id, success, data)

        except requests.RequestException as e:
            logger.error(f"API error during polling: {e}")

        time.sleep(POLL_INTERVAL)


if __name__ == "__main__":
    import sys

    app_id = sys.argv[1] if len(sys.argv) > 1 else "your-app-id-here"
    poll_and_execute(app_id)

Run the connector:

python connector.py a1b2c3d4-...

The loop will poll every 30 seconds, pick up any pending tasks, execute them, and report results back to Iden. Adjust POLL_INTERVAL based on how quickly you need tasks to be processed.

Best Practices

Poll at a reasonable interval. 30 seconds is a good default. Go as low as 10 seconds if you need near-real-time provisioning.
Handle 409 gracefully. A conflict response means the task is already done. Log it and move on.
Use specific error codes. ACCOUNT_NOT_FOUND is more useful to administrators than INTERNAL_ERROR. Only use INTERNAL_ERROR as a last resort.
Include descriptive error messages. Messages appear in the Iden dashboard. A message like "User ext-123 not found in Okta" is far more helpful than "Not found."
Process tasks sequentially within a poll cycle. Execute one task at a time to avoid overwhelming the external system. If you need parallelism, limit concurrency and handle rate limiting.
Log everything. Log each task pickup, execution attempt, and status report. When something goes wrong, logs are the first place your team will look.
Handle network errors. If the Iden API is temporarily unreachable, catch the exception, log it, and continue polling on the next cycle. Do not crash the connector.

Task Execution

On this page