Agentspace Custom Connector for Azure Blob Storage

This connector is designed to fetch data from an Azure Blob Storage container, transform it into the Google Agentspace Discovery Engine Document format, and ingest it for enterprise search.

It supports text-based files (e.g., .txt, .md, .html, .json) and PDF files, applying a uniform Access Control List (ACL) to all ingested documents based on the configuration.

• Python 3.8+
• An active Azure account with a Storage Account and a container.
• A Google Cloud Project with the Discovery Engine API enabled.
• Appropriate credentials for both Azure (Connection String) and Google Cloud (Application Default Credentials).

Before you configure and run the connector, you must prepare your Google Cloud environment.

The connector uses a Google Cloud Storage (GCS) bucket to stage the data before importing it into Agentspace. This is a required step.

Create a bucket using the gcloud CLI. Remember to choose a globally unique name.

# Replace 'your-unique-bucket-name' with a unique name for your bucket
gcloud storage buckets create gs://your-unique-bucket-name --project="your-gcp-project-id" --location="us-central1"

You will add this bucket name to the config.yaml file in the next step.

You do not need to create the Data Store manually. The connector script (ingestor.py) will automatically create it on its first run if it doesn't exist. You simply need to decide on a unique data_store_id and specify it in the config.yaml file.

When deployed as a Cloud Run Job, the connector runs using a Service Account. This Service Account needs permissions to interact with other GCP services. Ensure the Service Account used by Cloud Run has the following IAM roles:

• roles/discoveryengine.admin: To create and manage the Agentspace Data Store.
• roles/storage.objectAdmin: To write staged data to the GCS bucket.
• roles/secretmanager.secretAccessor: To access the Azure connection string stored in Secret Manager.

To ensure uniqueness and prevent import errors, document IDs are not generated from the raw blob name. Instead, a MD5 hash of the full blob name (e.g., folder/document.pdf) is used as the final document.id. This guarantees that every document has a unique, validly formatted ID. The original, human-readable blob name is preserved in the document's title field for searching and display.

To ensure that links to private Azure blobs are accessible from the Agentspace search results, the connector automatically generates a Shared Access Signature (SAS) token for each document's URL.

• The validity period for these tokens is configurable in config.yaml via the sas_token_expiry_hours parameter (default is 24 hours).
• This provides secure, time-limited read access to the files directly from the search results page.

Currently, this connector applies a uniform ACL to all documents based on the acl.readers configuration. For more granular, per-document permissions, the AzureBlobEntity class would need to be extended. Common strategies for implementing dynamic ACLs include:

1. Path-Based Rules: Defining rules in config.yaml that map blob name prefixes (e.g., finance/) to specific reader groups.
2. Sidecar Metadata Files: Storing a .acl.json file alongside each blob in Azure to specify its unique permissions.
3. External Authority API: Calling an external service to retrieve the correct ACL for each blob during processing.

After completing the GCP prerequisites, configure your settings in the config.yaml file.

# config.yaml

# Google Cloud Platform Settings
gcp:
  project_id: "your-gcp-project-id"
  location: "global"
  data_store_id: "your-chosen-data-store-id" # Will be created automatically
  gcs_bucket: "your-unique-bucket-name"     # The bucket you created in the previous step

# Azure Blob Storage Settings
azure:
  # For local runs, place the connection string here.
  # For cloud deployment, this is ignored and the value is read from a secret.
  connection_string: "DefaultEndpointsProtocol=https..."
  container_name: "your-container-name"

# Access Control List (ACL) Settings
acl:
  readers:
    - "user:your-user@example.com"
    - "group:your-team@example.com"

This project uses Python packages listed in requirements.txt. The dependencies are pinned to specific versions to ensure reproducible builds. It is highly recommended to use a virtual environment.

# Create and activate a virtual environment (recommended)
python3 -m venv venv
source venv/bin/activate

# Install the exact dependency versions
pip install -r requirements.txt

Once the configuration and installation are complete, you can run the ingestion process.

# Make sure you are in the project's root directory
# and your virtual environment is activated.

python -m main

The script will connect to your Azure container, process the supported files, and upload them to your specified Agentspace Data Store.

A comprehensive test suite is included to verify the connector's logic. The tests use pytest and pytest-cov.

To run the tests:

# Make sure you are in the project's root directory
# and your virtual environment is activated.

python -m pytest

To run the tests and generate a coverage report:

# The --cov flag will generate a report detailing test coverage.
python -m pytest --cov=.

## 3. Running the Connector

Once the configuration and installation are complete, you can run the ingestion process.

```bash
# Make sure you are in the azure_object_storage directory
# and your virtual environment is activated.

python -m azure_object_storage.main

The script will connect to your Azure container, process the supported files, and upload them to your specified Agentspace Data Store.

To run this connector automatically on a schedule, deploy it as a serverless Cloud Run Job and trigger it with Cloud Scheduler.

These variables will simplify the subsequent commands.

export GCP_PROJECT_ID="your-gcp-project-id"
export GCP_REGION="us-central1" # e.g., us-central1
export ARTIFACT_REGISTRY_REPO="agentspace-connectors"
export CLOUD_RUN_JOB_NAME="agentspace-azure-connector"
export AZURE_SECRET_NAME="azure-storage-connection-string"
export IMAGE_NAME="${GCP_REGION}-docker.pkg.dev/${GCP_PROJECT_ID}/${ARTIFACT_REGISTRY_REPO}/${CLOUD_RUN_JOB_NAME}:latest"

Store your Azure Storage connection string in Google Cloud Secret Manager. The Cloud Run Job will securely access this secret at runtime.

# Enable the API
gcloud services enable secretmanager.googleapis.com --project=${GCP_PROJECT_ID}

# Create a secret
gcloud secrets create "${AZURE_SECRET_NAME}" \
  --replication-policy="automatic" \
  --project=${GCP_PROJECT_ID}

# Add your connection string as a secret version
echo -n "PASTE_YOUR_AZURE_CONNECTION_STRING_HERE" | \
gcloud secrets versions add "${AZURE_SECRET_NAME}" --data-file=-
  --project=${GCP_PROJECT_ID}

This command builds the Docker image from the source code and pushes it to Artifact Registry.

# Enable the APIs
gcloud services enable artifactregistry.googleapis.com cloudbuild.googleapis.com --project=${GCP_PROJECT_ID}

# Create an Artifact Registry repository (if it doesn't exist)
gcloud artifacts repositories create "${ARTIFACT_REGISTRY_REPO}" \
    --repository-format=docker \
    --location=${GCP_REGION} \
    --project=${GCP_PROJECT_ID} || echo "Repository already exists."

# Build and submit the image
gcloud builds submit . --tag="${IMAGE_NAME}" --project=${GCP_PROJECT_ID}

Create a Cloud Run Job that uses the container image and is configured to access the Azure credential from Secret Manager.

# Enable the API
gcloud services enable run.googleapis.com --project=${GCP_PROJECT_ID}

# Deploy the job
gcloud run jobs create "${CLOUD_RUN_JOB_NAME}" \
  --project="${GCP_PROJECT_ID}" \
  --region="${GCP_REGION}" \
  --image="${IMAGE_NAME}" \
  --set-secrets="AZURE_STORAGE_CONNECTION_STRING=${AZURE_SECRET_NAME}:latest" \
  --task-timeout=1800 # 30 minutes

You can trigger the job manually to test the end-to-end process.

gcloud run jobs execute "${CLOUD_RUN_JOB_NAME}" --region="${GCP_REGION}" --project="${GCP_PROJECT_ID}" --wait

Create a Cloud Scheduler job to run the connector automatically (e.g., once a day).

# Enable the API
gcloud services enable cloudscheduler.googleapis.com --project=${GCP_PROJECT_ID}

# Create a service account for the scheduler
gcloud iam service-accounts create scheduler-run-invoker --project=${GCP_PROJECT_ID}

# Grant the service account permission to run the Cloud Run Job
gcloud run jobs add-iam-policy-binding "${CLOUD_RUN_JOB_NAME}" \
  --project="${GCP_PROJECT_ID}" \
  --region="${GCP_REGION}" \
  --member="serviceAccount:scheduler-run-invoker@${GCP_PROJECT_ID}.iam.gserviceaccount.com" \
  --role="roles/run.invoker"

# Create the scheduler job
gcloud scheduler jobs create http "${CLOUD_RUN_JOB_NAME}-schedule" \
  --project="${GCP_PROJECT_ID}" \
  --schedule="0 2 * * *" \  # Runs daily at 2 AM
  --uri="https://${GCP_REGION}-run.googleapis.com/apis/run.googleapis.com/v1/namespaces/${GCP_PROJECT_ID}/jobs/${CLOUD_RUN_JOB_NAME}:run" \
  --http-method=POST \
  --oauth-service-account-email="scheduler-run-invoker@${GCP_PROJECT_ID}.iam.gserviceaccount.com"