Auto-Assign Tasks After Analysis - MindBridge Documentation

This example demonstrates how to automatically assign review tasks using the Python SDK after an analysis is completed. You will connect to an existing analysis, export a filtered set of rows from a data table, and assign entry tasks to appropriate MindBridge users based on values in your data, such as company code.

Prerequisites

Complete Get started so the Python SDK is installed and configured. You also need:

A MindBridge analysis URL for an engagement that you can access.
The logical name of the data table you want to export, such as gl_journal_lines.

Overview

In this guide you will:

Parse an analysis URL to extract the MindBridge host and analysis result ID.
Authenticate with the MindBridge API and map business keys to user IDs by email for task assignment.
Export filtered rows to a CSV file, and create tasks only when no matching task exists (idempotent).

Configure Your Inputs

Adjust the values in the Configure inputs section at the top of the script below to match your analysis and task assignment requirements. Typically, you only need to update that section; the rest of the script follows a standard process.

analysis_url: The full MindBridge URL for the analysis. The script parses this to extract the server host and analysis result ID. You can also supply these values directly if your workflow provides them.
data_table_query: Filters applied to the data table to identify which rows should receive tasks (for example, by date range or minimum risk). Modify the fields and operators as needed.
data_table_logical_name: The logical name of the data table to export. This table must exist in the analysis. The script will print available logical names to help confirm your selection.
column_for_assigning: The column whose values determine the assignee, for example, company_code. You must provide a mapping for each expected value.
assigned_user_map: A mapping from values in column_for_assigning to MindBridge user email addresses. The script will look up each email and assign the task to the associated user.

The listing below is one continuous script you can save and run. Step boundaries are marked with comments inside the code.

Complete example

import csv
import os
from datetime import date
from pathlib import Path
from tempfile import NamedTemporaryFile

from urllib3.util import parse_url

import mindbridgeapi as mbapi

# --- Configure inputs (edit for your environment) ---
# Replace with the full analysis URL from your browser.
# Host and path IDs must match your tenant.
analysis_url = (
    "https://{your-tenant}.mindbridge.ai"
    "/app/organization/{ORG_ID}"
    "/engagement/{ENGAGEMENT_ID}"
    "/analysis/{ANALYSIS_RESULT_ID}"
    "/analyze/financial-statements"
    "?productCode=GENERAL_LEDGER"
)

data_table_query = {
    "effective_date": {
        "$gte": date(2022, 4, 1),
        "$lt": date(2022, 6, 1),
    },
    "risk": {"$gte": 15_00},
}
data_table_logical_name = "gl_journal_lines"
column_for_assigning = "company_code"

# Keys must match values in your data column.
# Values must be real user emails in your tenant.
assigned_user_map = {
    "001": "reviewer-one@yourcompany.com",
    "002": "reviewer-two@yourcompany.com",
}

# --- Step 1: Parse the analysis URL ---
parsed_url = parse_url(analysis_url)
mindbridge_url = parsed_url.host

parsed_url_path = parsed_url.path.split("/")
analysis_idx = parsed_url_path.index("analysis")
analysis_result_id = parsed_url_path[analysis_idx + 1]

# --- Step 2: Connect, load analysis, resolve assignees ---
# Load the API token, connect, fetch the analysis and
# data tables, print logical names, restart data tables
# for a fresh export, and resolve each assignee email
# to a user ID. Errors if an email is missing or
# ambiguous.

token = os.environ.get("MINDBRIDGE_API_TOKEN", "")

server = mbapi.Server(url=mindbridge_url, token=token)

analysis_result = server.analysis_results.get_by_id(
    analysis_result_id
)
analysis = server.analyses.get_by_id(analysis_result.analysis_id)

print("Available data tables:")
for data_table in analysis.data_tables:
    print(
        f"- logical_name: {data_table.logical_name} "
        f"(type: {data_table.type}, id: {data_table.id})"
    )

server.analyses.restart_data_tables(analysis)
data_table = next(
    dt
    for dt in analysis.data_tables
    if dt.logical_name == data_table_logical_name
)
print(
    f"Using logical_name: {data_table.logical_name} "
    f"(type: {data_table.type}, id: {data_table.id})"
)

assigned_user_map_id = {}
for key, assigned_user_email in assigned_user_map.items():
    results = server.users.get(json={"email": assigned_user_email})
    user, *others = results
    if not user:
        raise ValueError(
            f"User with email '{assigned_user_email}' "
            "not found."
        )
    if others:
        raise Exception(
            f"Multiple users found with email "
            f"'{assigned_user_email}'. "
            "Please resolve duplicates."
        )
    assigned_user_map_id[key] = user.id

# --- Step 3: Export data and assign tasks ---
# Export the selected table to CSV. For each row, create
# an entry task for the mapped user unless a task already
# exists for the same analysis, result, transaction, and
# row (safe to rerun).

with NamedTemporaryFile(delete=False) as temp_file:
    temp_file_path = Path(temp_file.name)

print(f"Exporting to: {temp_file_path}")
async_result = server.data_tables.export(
    data_table,
    fields=[
        "rowid",
        "txid",
        "risk",
        "effective_date",
        column_for_assigning,
    ],
    query=data_table_query,
)
server.data_tables.wait_for_export(async_result)
temp_file_path = server.data_tables.download(
    async_result, output_file_path=temp_file_path
)

with temp_file_path.open(newline="", encoding="utf_8") as infile:
    reader = csv.DictReader(infile)
    print("Assigning tasks for the following rows:")
    for row in reader:
        row_id = row["rowid"]
        transaction_id = row["txid"]
        print(
            f"- row_id: {row_id}, "
            f"transaction_id: {transaction_id}, "
            f"risk: {row['risk']}, "
            f"effective_date: {row['effective_date']}"
        )
        assigned_id = assigned_user_map_id[row[column_for_assigning]]
        task = mbapi.TaskItem(
            row_id=row_id,
            transaction_id=transaction_id,
            type=mbapi.TaskType.ENTRY,
            status=mbapi.TaskStatus.OPEN,
            engagement_id=analysis.engagement_id,
            analysis_result_id=analysis_result.id,
            audit_areas=["Audit Area 1"],
            assigned_id=assigned_id,
        )
        existing_task = next(
            server.tasks.get(
                json={
                    "analysisId": analysis.id,
                    "analysisResultId": task.analysis_result_id,
                    "transactionId": task.transaction_id,
                    "rowId": task.row_id,
                }
            ),
            None,
        )
        if existing_task:
            print(
                f"   - Task already exists "
                f"(ID: {existing_task.id})"
            )
        else:
            created_task = server.tasks.create(task)
            print(f"    - Created task (ID: {created_task.id})")

print("Task assignment complete.")
temp_file_path.unlink()

When the script finishes, the filtered rows appear as tasks in MindBridge, assigned to the users specified in assigned_user_map. To modify this workflow, adjust the export fields, the data_table_query, or the TaskItem fields (such as audit areas or task type) to fit your organization’s review process.

​Prerequisites

​Overview

​Configure Your Inputs

​Complete example

Prerequisites

Overview

Configure Your Inputs

Complete example