Notebook with sempy_labs failed when executed in DataPipeline

Hello Fabric Community,

I’ve been struggling with the complex execution context in Fabric (p.s. Who's Calling?).

I created a notebook to deploy a (template) Semantic Model from a central workspace to a target workspace using sempy_labs.labs.

When I execute this notebook using a user identity, it works ✅

When I execute the notebook inside a pipeline with a connection (authenticated using a Service Principal), it fails ❌ with the following error (though the SP has contributor access on both source and target WS):

FabricHTTPException
403 Forbidden for URL: https://api.fabric.microsoft.com/v1.0/myorg/groups/xxxxxxxx-yyyyy-zzzzz-000000/datasets
Headers: {'Content-Length': '0', 'Strict-Transport-Security': 'max-age=31536000; includeSubDomains', 'X-Frame-Options': 'deny', 'X-Content-Type-Options': 'nosniff', 'Access-Control-Expose-Headers': 'RequestId', 'RequestId': 'a9bd2bed-ff7b-4c1c-b28d-551727ebe685', 'Date': 'Mon, 20 Apr 2026 18:56:10 GMT'}

When I execute the notebook inside a data pipeline without a connection, it occasionally works, but mostly throws this error:

WorkspaceNotFoundException
Workspace 'xxxxxxxx-yyyyy-zzzzz-000000' not found.

import sempy_labs as labs
from sempy_labs import directlake
import sempy_labs.report as rep
import time
import logging
import sys
import os
import requests

# all the variables and parameters are collected and passed correctly (ignore them)

# Helper functions

def get_workspace_id(workspace_name: str | None):
    if not workspace_name:
        return None
    df = fabric.list_workspaces()
    row = df.loc[df["Name"] == workspace_name, "Id"]
    return row.iloc[0] if not row.empty else None

def get_item_id(workspace_id, item_name, item_type):
    df = fabric.list_items(workspace=workspace_id)
    row = df.loc[
        (df["Type"] == item_type) &
        (df["Display Name"] == item_name),
        "Id"
    ]
    return row.iloc[0] if not row.empty else None

def semantic_model_exists(workspace_id, dataset_name):
    return get_item_id(workspace_id, dataset_name, "SemanticModel") is not None


def report_exists(workspace_id, report_name):
    return get_item_id(workspace_id, report_name, "Report") is not None

def stage(name):
    logger.info(f"=== {name} ===")

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)]
)

logger = logging.getLogger("deployment")

def main():

    stage("Resolve Context")

    target_workspace_id = workspaceId
    lakehouse_id = lakehouseId

    if not target_workspace_id:
        raise ValueError(f"Workspace not found: {TARGET_WORKSPACE_NAME}")

    if not lakehouse_id:
        raise ValueError(f"Lakehouse not found: {LAKEHOUSE_NAME}")

    stage("Check Existing Assets")

    sm_exists = semantic_model_exists(
        target_workspace_id,
        TARGET_SEMANTICMODEL_NAME
    )

    rep_exists = report_exists(
        target_workspace_id,
        SOURCE_REPORT_NAME
    )

    logger.info(f"Semantic model exists: {sm_exists}")
    logger.info(f"Report exists: {rep_exists}")

    # --------------------------------------------------
    # SHORT CIRCUIT
    # --------------------------------------------------

    if sm_exists and rep_exists:
        logger.info("Nothing to deploy")
    else:

        if not sm_exists:
            stage("Deploy Semantic Model")

            labs.deploy_semantic_model(
                source_dataset=SOURCE_DATASET_NAME,
                source_workspace=SOURCE_WORKSPACE_NAME,
                target_dataset=TARGET_SEMANTICMODEL_NAME,
                target_workspace=TARGET_WORKSPACE_NAME,
                refresh_target_dataset=False,
                overwrite=True
            )

            directlake.update_direct_lake_model_connection(
                dataset=TARGET_SEMANTICMODEL_NAME,
                workspace=TARGET_WORKSPACE_NAME,
                source=LAKEHOUSE_NAME,
                source_type="Lakehouse",
                source_workspace=TARGET_WORKSPACE_NAME,
                use_sql_endpoint=False
            )

        # --------------------------------------------------
        # CLONE REPORT
        # --------------------------------------------------

        if not rep_exists:
            stage("Clone Report")

            rep.clone_report(
                report=SOURCE_REPORT_NAME,
                cloned_report=SOURCE_REPORT_NAME,
                workspace=SOURCE_WORKSPACE_NAME,
                target_workspace=TARGET_WORKSPACE_NAME,
                target_dataset=TARGET_SEMANTICMODEL_NAME,
                target_dataset_workspace=TARGET_WORKSPACE_NAME
            )

            rep.report_rebind(
                report=SOURCE_REPORT_NAME,
                dataset=TARGET_SEMANTICMODEL_NAME,
                report_workspace=TARGET_WORKSPACE_NAME,
                dataset_workspace=TARGET_WORKSPACE_NAME
            )

    # --------------------------------------------------
    # FINAL REFRESH
    # --------------------------------------------------

    stage("Refresh Semantic Model")

    time.sleep(10)

    try:
        labs.refresh_semantic_model(
        dataset=TARGET_SEMANTICMODEL_NAME,
        workspace=TARGET_WORKSPACE_NAME
        )
    except Exception as e:
        raise


    logger.info("Deployment completed successfully")
​

Can someone help me how can I achieve the deployment by the pipeline execution, preferably, using SP. Because my pipeline triggers via Fabric API from an external application using SP.