Skip to main content

SDK for SAP IoT Services on Cloud Foundry. Wraps all Device Management APIs as well as both Cloud Gateways.

Project description

Table of Contents


This package wraps the SAP IoT Services, Cloud Foundry Edition Device Management APIs as well as the REST and the MQTT Cloud Gateway APIs in easy-to-use Python code. You can find more information on the APIs here and here.

Features include:

  • Wrapper around every functionality of the Device Management API
  • Better error feedback
  • Wrapper around both Cloud Gateways
  • Both clients, REST and MQTT, are extended so that they work directly with the encrypted keys in the pem file
  • Better error handling and error feedback for REST and MQTT
    • The MQTT client has an on_error and an on_command callback which directly return the parsed message
    • The REST client parses the error for batched and single uploads and throws an error with the respective message
    • Both clients return the failed message to enable the user to retry easily


You will need an instance of IoT Services, Cloud Foundry to use this SDK. You can find more information on how to activate the service here.


The package will be available at pip:

pip install iot-services-sdk

If you want to download it manually, please make sure to install paho-mqtt and requests into your Python environment.

How to obtain support

Please use GitHub Issues to file a bug.


Copyright (c) 2019 SAP SE or an SAP affiliate company. All rights reserved. This file is licensed under the SAP Sample Code License except as noted otherwise in the LICENSE file.


Please refer to the API documentation in the PDF file for details. You can check the tests for sample code.

Here is a small example displaying the following functionality:

  • Creating capabilities, a sensor type, a sensor and a device.
  • Downloading the certificate and creating a MQTT client.
  • Publishing valid and invalid measures as well as publishing and receiving commands.
  • Deleting every entity which was created.
import os

from iot_services_sdk import CapabilityService
from iot_services_sdk import DeviceService
from iot_services_sdk import GatewayService
from iot_services_sdk import SensorTypeService
from iot_services_sdk import SensorService

from iot_services_sdk import MQTTClient

# Initialize services
device_service = DeviceService(instance =, user = myuser, password = mypassword)
capability_service = CapabilityService(instance =, user = myuser, password = mypassword)
gateway_service = GatewayService(instance =, user = myuser, password = mypassword)
sensor_type_service = SensorTypeService(instance =, user = myuser, password = mypassword)
sensor_service = SensorService(instance =, user = myuser, password = mypassword)

# Turn requests debug functionality on

# Get Gateway
filters = ["status eq 'online'"]
gateways_response = gateway_service.get_gateways(filters=filters)
gateways = gateways_response.get_result()
mqtt_gateway = next(gateway for gateway in gateways if gateway['protocolId'] == 'mqtt')

# Create Measure Capability
properties = [
    {"name": "sdk_test_temp", "dataType": "double", "unitOfMeasure": "Celcius"}
create_response = capability_service.create_capability('sdk_test_measure_id', 'sdk_test_measure_cap', properties)

capability_measure_id = create_response.get_result().get('id')
capability_measure_alternate_id = create_response.get_result().get('alternateId')

# Create Command Capability
properties = [
    {"name": "sdk_test_cmd", "dataType": "string"}
create_response = capability_service.create_capability('sdk_test_cmd_id', 'sdk_test_cmd_cap', properties)

capability_cmd_id = create_response.get_result().get('id')
capability_cmd_alternate_id = create_response.get_result().get('alternateId')

# Create Device
create_response = device_service.create_device(mqtt_gateway['id'], 'sdk_device')

device = create_response.get_result()
device_id = device['id']
device_alternate_id = device['alternateId']

# Create Sensor Type
capabilities = [
    {'id': capability_measure_id, 'type': 'measure'},
    {'id': capability_cmd_id, 'type': 'command'}
create_response = sensor_type_service.create_sensor_type('123456', 'sdk_mqtt_test_type', capabilities)

sensor_type_id = create_response.get_result().get('id')
sensor_type_alternate_id = create_response.get_result().get('alternateId')

# Create Sensor
create_response = sensor_service.create_sensor(device_id, 'sdk-test-mqtt-sensor-id', 'sdk-test-mqtt-sensor-name', sensor_type_id)

sensor_id = create_response.get_result()['id']
sensor_alternate_id = create_response.get_result()['alternateId']

# Download Certificate
res = device_service.get_device_pem(device_id).get_result()

pem = res.get('pem')
secret = res.get('secret')
pem_filepath = 'cert.pem'

# Write Certificate to PEM file
pem_file = open(pem_filepath, 'w')

# Get MQTT Client for Device
mqtt_client = device_service.get_mqtt_client(device_alternate_id, pem_filepath, secret)

# Publish a measure
measures = [
    {'sdk_test_temp': 30}
message_info_measures = mqtt_client.publish(capability_measure_alternate_id, sensor_alternate_id, measures, device_alternate_id)

# Read measures
get_measures_response = device_service.get_measures(device_id)
measures = get_measures_response.get_result()

# Publish invalid measure and receive error
def on_error(client, userdata, report):
    print('ERROR RECEIVED: ')

mqtt_client.on_error = on_error

measures = [
    {'i_am_invalid': 30}
message_info_measures = mqtt_client.publish(capability_measure_alternate_id, sensor_alternate_id, measures, device_alternate_id)

# Publish and receive command
def on_command(client, userdata, command):
    print('COMMAND RECEIVED: ')

mqtt_client.on_command = on_command


command = {'sdk_test_cmd': 'I am a SDK!'}

send_command_res = device_service.send_command_to_device(device_id, capability_cmd_id, sensor_id, command)

# Stop MQTT loop

# Delete Sensor
delete_sensor_res = sensor_service.delete_sensor(sensor_id)

# Delete Device
delete_response = device_service.delete_device(device_id)

# Delete SensorType
delete_sensor_type_response = sensor_type_service.delete_sensor_type(sensor_type_id)

# Delete Measure Capability
delete_response = capability_service.delete_capability(capability_measure_id)

# Delete Command Capability
delete_response = capability_service.delete_capability(capability_cmd_id)

# Delete Certfile
except OSError:


To run the tests, you have to place a config.ini in the root directory. It has to contain the following information:


Then, run


in the root directory. The tests will clean up after themselves - there is no need to check your system afterwards.

Build Docs

Use Sphinx to build the documentation directly from the docstrings:

cd docs

sphinx-apidoc -f -o source/ ../sap_iot_services_sdk

make latexpdf

Copy the SAPIoTServicesSDK.pdf file into the root directory.

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for iot-services-sdk, version 2.0.2
Filename, size File type Python version Upload date Hashes
Filename, size iot_services_sdk-2.0.2.tar.gz (20.4 kB) File type Source Python version None Upload date Hashes View

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring Facebook / Instagram Facebook / Instagram PSF Sponsor Fastly Fastly CDN Google Google Object Storage and Download Analytics Huawei Huawei PSF Sponsor Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Salesforce Salesforce PSF Sponsor Sentry Sentry Error logging StatusPage StatusPage Status page