How to Extract CPI Message Logs with Custom Header Properties and Report in SAP Analytics Cloud

MJUDI

How to Extract CPI Message Logs with Custom Header Properties and Report in SAP Analytics Cloud

1. Problem Statement: Why You Can’t Rely on CPI Monitoring Alone

SAP’s Integration Suite (CPI) gives you visibility — but only just. If you want to check whether a message was delivered or inspect its Custom Header Properties, you can. But it’s manual. One message at a time. No consolidated view. No report.

Cloud ALM improves things slightly by centralising errors and alerts, but when it comes to reporting on message metadata like Custom Header Properties, you're stuck. Neither CPI’s monitor nor Cloud ALM’s Integration & Exception Monitor gives you a way to extract all this data in bulk.

And that’s a problem. Because sometimes your business needs answers like:

“Did all outbound invoice messages include the expected reference IDs?”
“Which messages failed for a specific customer or process?”
“Can we see everything we sent for this business object last month in one view?”

For many organisations, this isn’t just a nice-to-have — it’s a compliance or contractual requirement.

This is where the gap becomes critical: the data is there, but not accessible in a way that supports operational oversight, reporting, or auditing.

So, what’s the fix?

You build your own reporting pipeline — extracting message data via CPI’s APIs, enriching it with Custom Header Properties, and loading it into SAP Analytics Cloud (SAC).

2. Extracting Messages and Header Properties

To get the data you need out of SAP Cloud Integration (CPI), you’ll be working with two key APIs:

/MessageProcessingLogs to retrieve a list of processed messages. It is always advisable to test your APIs before using them in your application in Postman or Bruno for example:

/MessageProcessingLogs('{MessageGuid}')/CustomHeaderProperties to fetch Custom Header Properties for each message.

Step 1: Authenticate Against the CPI API

With basic authentication method, you can quickly test the API, however it's recommended to use a more secured method like OAuth 2.0 the way it's demonstrated in the attached script.

import requests
from requests.auth import HTTPBasicAuth

cpi_host = 'https://your-cpi-host.it-cpi.eu10.hana.ondemand.com'
username = 'your-technical-user'
password = 'your-password'

Step 2: Get a List of Message GUIDs

response = requests.get(
    f"{cpi_host}/api/v1/MessageProcessingLogs?$format=json",
    auth=HTTPBasicAuth(username, password)
)

messages = response.json().get('d', {}).get('results', [])

Step 3: Loop Through Messages to Fetch Header Properties

all_data = []

for msg in messages:
    guid = msg.get('MessageGuid')
    hdr_response = requests.get(
        f"{cpi_host}/api/v1/MessageProcessingLogs('{guid}')/CustomHeaderProperties?$format=json",
        auth=HTTPBasicAuth(username, password)
    )
    
    custom_headers = hdr_response.json().get('d', {}).get('results', [])
    
    record = {
        'MessageGuid': guid,
        'IntegrationFlowName': msg.get('IntegrationFlowName'),
        'Timestamp': msg.get('LogStart'),
        'CustomHeaderProperties': {hdr['Name']: hdr['Value'] for hdr in custom_headers}
    }
    
    all_data.append(record)

Step 4: Store the Data

import pandas as pd

df = pd.json_normalize(all_data, sep='_')
df.to_csv('cpi_message_headers.csv', index=False)

3. Preparing Your SAC Model

To import the data you extracted out of SAP Cloud Integration (CPI) in SAP Analytics Cloud (SAC), you’ll be working with the following APIs:

GET all Models with api/v1/dataimport/models endpoint to return a list of all Models
GET Model Details with api/v1/dataimport/models/{modelID} endpoint to give us the types of data we can import into this model
GET Model Metadata with api/v1/dataimport/models/{modelID}/metadata endpoint
POST - Create a Job to be used later to import the data with api/v1/dataimport/models/{modelID}/importType endpoint
POST - Load data into a job with api/v1/dataimport/jobs/{jobID} endpoint
GET the list of jobs with api/v1/dataimport/jobs endpoint
POST - Validate with api/v1/dataimport/jobs/{jobID}/validate endpoint
POST - Run the job with api/v1/dataimport/jobs/{jobID}/run endpoint
GET the status of the run with api/v1/dataimport/jobs/{jobID}/status endpoint

However, there is a simple and quick way to import the data in one click mode with the api/v1/import/{modelID} endpoint. This method is demonstrated in the sample attached script.

Before you can load data into SAC, you must ensure the model exists and that you’ve verified its metadata.

Step 1: Get Model Info

model_id = 'your-model-id'
sac_host = 'https://<your-sac-tenant>.sapbusinessobjects.cloud'
token = 'your-bearer-token'

headers = {
    'Authorization': f'Bearer {token}'
}

response = requests.get(
    f'{sac_host}/api/v1/dataimport/models/{model_id}',
    headers=headers
)

print(response.json())

Step 2: Get Model Metadata (Field Types)

meta_response = requests.get(
    f'{sac_host}/api/v1/dataimport/models/{model_id}/metadata',
    headers=headers
)

metadata = meta_response.json()

4. Setting Up the Data Import Job

Step 1: Create the Import Job

import json

job_payload = {
    "importType": "Append"  # or "Full"
}

create_job_response = requests.post(
    f'{sac_host}/api/v1/dataimport/models/{model_id}/importType',
    headers={**headers, 'Content-Type': 'application/json'},
    data=json.dumps(job_payload)
)

job_info = create_job_response.json()
job_id = job_info.get('id')

print(f"Created import job with ID: {job_id}")

5. Loading Data into the Job

Step 1: Prepare the Payload

data_payload = {
    "data": [
        {
            "MessageGuid": "12345",
            "IntegrationFlowName": "InvoiceDispatch",
            "Timestamp": "2024-12-01T10:15:00Z",
            "BusinessObjectID": "INV-1001"
        },
        {
            "MessageGuid": "12346",
            "IntegrationFlowName": "InvoiceDispatch",
            "Timestamp": "2024-12-01T10:16:00Z",
            "BusinessObjectID": "INV-1002"
        }
    ]
}

Step 2: Load the Data Into the Job

load_response = requests.post(
    f'{sac_host}/api/v1/dataimport/jobs/{job_id}',
    headers={**headers, 'Content-Type': 'application/json'},
    data=json.dumps(data_payload)
)

if load_response.status_code == 202:
    print("Data loaded into job successfully.")
else:
    print(f"Error loading data: {load_response.status_code} - {load_response.text}")

6. Managing and Monitoring Jobs

Step 1: Validate the Job

validate_response = requests.post(
    f'{sac_host}/api/v1/dataimport/jobs/{job_id}/validate',
    headers=headers
)

if validate_response.status_code == 202:
    print("Validation successful. Ready to run the job.")
else:
    print(f"Validation failed: {validate_response.status_code} - {validate_response.text}")

Step 2: Run the Job

run_response = requests.post(
    f'{sac_host}/api/v1/dataimport/jobs/{job_id}/run',
    headers=headers
)

if run_response.status_code == 202:
    print("Job started successfully.")
else:
    print(f"Failed to start the job: {run_response.status_code} - {run_response.text}")

7. Checking Job Status

Step 1: Poll the Job Status

import time

while True:
    status_response = requests.get(
        f'{sac_host}/api/v1/dataimport/jobs/{job_id}/status',
        headers=headers
    )

    status_info = status_response.json()
    status = status_info.get('status')

    print(f"Current job status: {status}")

    if status in ["Completed", "Failed"]:
        break

    time.sleep(5)

Conclusion

If you’ve ever felt stuck manually digging through CPI logs just to answer simple business questions, now you know there’s a better way.

By using SAP’s APIs, a little Python, and the SAC Data Import API, you can build a repeatable pipeline to extract, load, and report on Custom Header Properties — at scale.

The steps you followed — extracting with /MessageProcessingLogs, staging data for SAC, validating, running, and monitoring — create a blueprint you can extend for other use cases too.

The data was always there. Now, you control it.

A sample Python script is attached to this blog, however, the script is following the one-click approach to import the data into SAC.

Common Mistakes and How to Avoid Them

Mismatched Field Names: Match your payload keys exactly to the SAC model field names (case-sensitive).
Wrong Date Format: Use ISO 8601 date format (YYYY-MM-DDTHH:MM:SSZ).
Not Validating Before Running: Always validate your job to catch issues early.
Overwriting When You Meant to Append: Use Append unless you intend to replace all data.
Token Expiry: Ensure your OAuth token is valid or refresh it if needed.

Automate CPI Log Reporting: Extract Custom Header Properties and Load into SAP Analytics Cloud

How to Extract CPI Message Logs with Custom Header Properties and Report in SAP Analytics Cloud

1. Problem Statement: Why You Can’t Rely on CPI Monitoring Alone

2. Extracting Messages and Header Properties

Step 1: Authenticate Against the CPI API

Step 2: Get a List of Message GUIDs

Step 3: Loop Through Messages to Fetch Header Properties

Step 4: Store the Data

3. Preparing Your SAC Model

Step 1: Get Model Info

Step 2: Get Model Metadata (Field Types)

4. Setting Up the Data Import Job

Step 1: Create the Import Job

5. Loading Data into the Job

Step 1: Prepare the Payload

Step 2: Load the Data Into the Job

6. Managing and Monitoring Jobs

Step 1: Validate the Job

Step 2: Run the Job

7. Checking Job Status

Step 1: Poll the Job Status

Conclusion

Common Mistakes and How to Avoid Them

Get Your SAP HANA Idea Incubator Badge Today!

SCN Mission - SAP HANA Quiz Challenge is now retired

Share your #HANAStory and Win