Detecting the User Who Manually Triggered a Fabric Pipeline or Notebook

Update – Finally Solved for Both Pipelines and Notebooks

After further investigation and testing, I finally found a reliable way to determine who manually triggered both Microsoft Fabric Pipelines and Notebooks.

The original approach I shared above worked in some scenarios by using:

• Job instance metadata

• Runtime context

• Microsoft Graph /me

However, those methods only identify the currently executing identity and can be unreliable for auditing purposes, especially when notebooks or pipelines are triggered by different users.

Final Solution

The most reliable approach is to use the Power BI / Fabric Admin Activity Events API:

GET /admin/activityevents

This API exposes Fabric audit events and allows us to identify the user who actually initiated a run.

Pipeline Detection Logic

1. Get the exact pipeline run start time using:

queryactivityruns

1. Retrieve audit events around that execution time.

2. Find the matching:

Operation = RunArtifact
ObjectId  = Pipeline ID

1. Return:

UserId

which contains the triggering user's email address (UPN).

Notebook Detection Logic

Standalone notebooks do not have a queryactivityruns endpoint, so the process is slightly different:

1. Get the current notebook Job Instance.

2. Resolve the notebook start time.

3. Query Admin Activity Events.

4. Find matching notebook execution events:

Operation = StartRunNotebook
or
Operation = RunArtifact

1. Match on Notebook ID.

2. Return the triggering user's email address.

Converting Email Address to Display Name

The audit log returns:

[email protected]

To make logs more readable, I then call Microsoft Graph:

GET /users/{userPrincipalName}

and retrieve:

displayName
userPrincipalName

which allows logging values such as:

User Name ([email protected])

instead of only the email address.

Important Discovery – Audit Log Visibility Delay

During testing I discovered that Fabric audit events are not always immediately available through the Admin Activity Events API.

For example:

Notebook Start Time : 05:21:41
Audit Event Time    : 05:21:51

The audit event itself was generated almost immediately.

However, the event was not visible through the API until approximately 5–7 minutes later.

Because of this, a simple one-time lookup can fail even though the audit event already exists internally.

Final Reliability Improvement

To make the solution production-ready, I implemented a retry mechanism:

retry_count=10
retry_interval_seconds=60

The lookup now retries every minute until the matching audit event becomes available.

This ensures that:

• Pipelines resolve the correct triggering user.

• Notebooks resolve the correct triggering user.

• Delayed audit ingestion does not cause false "Unknown" results.

• Security and operational reporting remain accurate.

Result

The final implementation can now reliably populate logging tables and System Run Reports with:

Trigger Type
Trigger Name
Run Started By

for:

• Scheduled runs

• Manual runs

• Orchestrator pipeline runs

• Standalone notebook executions

This has significantly improved operational auditing and troubleshooting in our Fabric environment.

Hopefully this helps others looking for a reliable way to track who actually executed Fabric workloads.

One important thing I found is that audit events are not always visible immediately through the Admin Activity Events API. The event timestamp can be accurate, but the API visibility can be delayed by several minutes. Because of that, I added a retry loop.

Below is the sanitised sample code structure:

import requests
import pandas as pd
import time

# ------------------------------------------------------------------
# Service Principal Inputs - replace with secure values
# ------------------------------------------------------------------
tenant_id = "<TENANT_ID>"
client_id = "<CLIENT_ID>"
client_secret = "<CLIENT_SECRET>"


def get_powerbi_admin_token(tenant_id, client_id, client_secret):
    token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"

    token_body = {
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret,
        "scope": "https://analysis.windows.net/powerbi/api/.default"
    }

    response = requests.post(
        token_url,
        data=token_body,
        headers={"Content-Type": "application/x-www-form-urlencoded"}
    )

    response.raise_for_status()
    return response.json()["access_token"]


def get_graph_display_name_from_upn(tenant_id, client_id, client_secret, upn):
    if not upn:
        return "Unknown"

    token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"

    token_body = {
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret,
        "scope": "https://graph.microsoft.com/.default"
    }

    token_response = requests.post(
        token_url,
        data=token_body,
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        timeout=30
    )

    if token_response.status_code != 200:
        return upn

    graph_token = token_response.json()["access_token"]

    graph_response = requests.get(
        f"https://graph.microsoft.com/v1.0/users/{upn}?$select=displayName,userPrincipalName",
        headers={"Authorization": f"Bearer {graph_token}"},
        timeout=30
    )

    if graph_response.status_code != 200:
        return upn

    user = graph_response.json()
    display_name = user.get("displayName")
    user_principal_name = user.get("userPrincipalName") or upn

    if display_name:
        return f"{display_name} ({user_principal_name})"

    return user_principal_name


def get_manual_pipeline_triggered_by(
    tenant_id,
    client_id,
    client_secret,
    workspace_id,
    pipeline_id,
    pipeline_run_id,
    window_minutes=15,
    retry_count=10,
    retry_interval_seconds=60
):
    """
    Resolve who manually triggered a Fabric pipeline run.

    Logic:
    1. Use Fabric queryactivityruns API to find the actual pipeline run start time.
    2. Use Power BI Admin Activity Events API to search audit events around that time.
    3. Find Operation = RunArtifact where ObjectId = pipeline_id.
    4. Retry because Admin Activity Events API visibility can be delayed.
    5. Return UserId from the matching audit event.
    """

    fabric_token = mssparkutils.credentials.getToken(
        "https://api.fabric.microsoft.com"
    )

    fabric_headers = {
        "Authorization": f"Bearer {fabric_token}",
        "Content-Type": "application/json"
    }

    activity_runs_url = (
        f"https://api.fabric.microsoft.com/v1/workspaces/{workspace_id}"
        f"/datapipelines/pipelineruns/{pipeline_run_id}/queryactivityruns"
    )

    activity_payload = {
        "lastUpdatedAfter": "2026-01-01T00:00:00Z",
        "lastUpdatedBefore": "2026-12-31T23:59:59Z",
        "filters": [],
        "orderBy": [
            {
                "orderBy": "ActivityRunStart",
                "order": "ASC"
            }
        ]
    }

    activity_response = requests.post(
        activity_runs_url,
        headers=fabric_headers,
        json=activity_payload
    )

    activity_response.raise_for_status()

    activity_rows = activity_response.json().get("value", [])

    if not activity_rows:
        return None

    df_activities = pd.json_normalize(activity_rows)

    start_col = None

    for col in ["activityRunStart", "ActivityRunStart", "startTime", "StartTime"]:
        if col in df_activities.columns:
            start_col = col
            break

    if not start_col:
        return None

    run_start_utc = pd.to_datetime(
        df_activities[start_col],
        utc=True,
        errors="coerce"
    ).min()

    if pd.isna(run_start_utc):
        return None

    powerbi_token = get_powerbi_admin_token(
        tenant_id,
        client_id,
        client_secret
    )

    powerbi_headers = {
        "Authorization": f"Bearer {powerbi_token}",
        "Content-Type": "application/json"
    }

    for retry_attempt in range(1, retry_count + 1):

        print(f"Pipeline audit lookup retry {retry_attempt} of {retry_count}")

        search_start_dt = run_start_utc - pd.Timedelta(minutes=window_minutes)
        search_end_dt = run_start_utc + pd.Timedelta(minutes=window_minutes)

        events = query_admin_activity_events(
            powerbi_headers,
            search_start_dt,
            search_end_dt
        )

        candidates = []

        for event in events:
            operation = event.get("Operation", "")

            object_id = str(
                event.get("ObjectId")
                or event.get("ItemId")
                or event.get("ArtifactId")
                or ""
            ).lower()

            if operation == "RunArtifact" and object_id == pipeline_id.lower():
                event_creation_dt = pd.to_datetime(
                    event.get("CreationTime"),
                    utc=True,
                    errors="coerce"
                )

                if pd.notna(event_creation_dt):
                    event["_time_diff_seconds"] = abs(
                        (event_creation_dt - run_start_utc).total_seconds()
                    )

                    candidates.append(event)

        if candidates:
            best_event = sorted(
                candidates,
                key=lambda x: x["_time_diff_seconds"]
            )[0]

            return best_event.get("UserId")

        if retry_attempt < retry_count:
            time.sleep(retry_interval_seconds)

    return None


def get_manual_notebook_triggered_by(
    tenant_id,
    client_id,
    client_secret,
    notebook_id,
    notebook_start_time_utc,
    window_minutes=15,
    max_allowed_time_diff_seconds=60,
    retry_count=10,
    retry_interval_seconds=60
):
    """
    Resolve who manually triggered a standalone Fabric notebook run.

    Logic:
    1. Use the notebook job start time.
    2. Use Power BI Admin Activity Events API to search audit events around that time.
    3. Match Operation = StartRunNotebook or RunArtifact.
    4. Match ObjectId = notebook_id.
    5. Retry because audit events may not be immediately visible.
    6. Reject old matches using max_allowed_time_diff_seconds.
    """

    run_start_utc = pd.to_datetime(
        notebook_start_time_utc,
        utc=True,
        errors="coerce"
    )

    if pd.isna(run_start_utc):
        return None

    powerbi_token = get_powerbi_admin_token(
        tenant_id,
        client_id,
        client_secret
    )

    powerbi_headers = {
        "Authorization": f"Bearer {powerbi_token}",
        "Content-Type": "application/json"
    }

    for retry_attempt in range(1, retry_count + 1):

        print(f"Notebook audit lookup retry {retry_attempt} of {retry_count}")

        search_start_dt = run_start_utc - pd.Timedelta(minutes=window_minutes)
        search_end_dt = run_start_utc + pd.Timedelta(minutes=window_minutes)

        events = query_admin_activity_events(
            powerbi_headers,
            search_start_dt,
            search_end_dt
        )

        candidates = []

        for event in events:
            operation = event.get("Operation", "")

            object_id = str(
                event.get("ObjectId")
                or event.get("ItemId")
                or event.get("ArtifactId")
                or ""
            ).lower()

            if (
                operation in ["StartRunNotebook", "RunArtifact"]
                and object_id == notebook_id.lower()
            ):
                event_creation_dt = pd.to_datetime(
                    event.get("CreationTime"),
                    utc=True,
                    errors="coerce"
                )

                if pd.notna(event_creation_dt):
                    event["_time_diff_seconds"] = abs(
                        (event_creation_dt - run_start_utc).total_seconds()
                    )

                    event["_operation_priority"] = {
                        "StartRunNotebook": 1,
                        "RunArtifact": 2
                    }.get(operation, 99)

                    candidates.append(event)

        if candidates:
            best_event = sorted(
                candidates,
                key=lambda x: (
                    x["_operation_priority"],
                    x["_time_diff_seconds"]
                )
            )[0]

            if best_event["_time_diff_seconds"] <= max_allowed_time_diff_seconds:
                return best_event.get("UserId")

        if retry_attempt < retry_count:
            time.sleep(retry_interval_seconds)

    return None


def query_admin_activity_events(powerbi_headers, search_start_dt, search_end_dt):
    """
    Query Power BI / Fabric Admin Activity Events API.
    Splits the search if the time window crosses a UTC date boundary.
    """

    search_windows = []

    if search_start_dt.date() == search_end_dt.date():
        search_windows.append((search_start_dt, search_end_dt))
    else:
        end_of_first_day = (
            search_start_dt.normalize()
            + pd.Timedelta(days=1)
            - pd.Timedelta(seconds=1)
        )

        start_of_second_day = search_end_dt.normalize()

        search_windows.append((search_start_dt, end_of_first_day))
        search_windows.append((start_of_second_day, search_end_dt))

    events = []

    for window_start, window_end in search_windows:

        search_start_str = window_start.strftime("%Y-%m-%dT%H:%M:%S.000Z")
        search_end_str = window_end.strftime("%Y-%m-%dT%H:%M:%S.000Z")

        activity_events_url = (
            "https://api.powerbi.com/v1.0/myorg/admin/activityevents"
            f"?startDateTime='{search_start_str}'"
            f"&endDateTime='{search_end_str}'"
        )

        response = requests.get(
            activity_events_url,
            headers=powerbi_headers
        )

        response.raise_for_status()

        data = response.json()
        events.extend(data.get("activityEventEntities", []))

        continuation_uri = data.get("continuationUri")

        while continuation_uri:
            continuation_response = requests.get(
                continuation_uri,
                headers=powerbi_headers
            )

            continuation_response.raise_for_status()

            continuation_data = continuation_response.json()
            events.extend(continuation_data.get("activityEventEntities", []))

            continuation_uri = continuation_data.get("continuationUri")

    return events

Required permissions:

• Fabric access to query pipeline activity runs
• Power BI / Fabric Admin Activity Events API access
• Microsoft Graph application permission such as User.Read.All with admin consent, if display name resolution is required

Final output example:

Trigger Type : Manual
Trigger Name : User Name ([email protected])

This gives a much more reliable result than using runtime context or Graph /me, because those can return the current execution identity rather than the user who actually triggered the pipeline or notebook.