---
title: Integrations
slug: sdk/integrations
docTags: 
createdAt: 2026-05-08T21:18:28.940Z
---

The integrations module manages cloud connector instances, subscriptions (topics), and object storage on Litmus Edge.

## Imports

The integrations surface lives under `litmussdk.integrations.common`:

```python
# Instances and topics (REST API, 3.x and 4.0.x)
from litmussdk.integrations.common import _integrations, _topics, _dashboard

# GraphQL streaming subscriptions (4.0.x only)
from litmussdk.integrations.common import _gql_streaming

# Object storage (4.0.x only)
from litmussdk.integrations.common import _object_storage
```

:::BlockQuote
**Note:** GraphQL and object storage functions raise `UnsupportedVersionError` when called against a 3.x device.
:::

***

## Providers

### Listing Providers

```python
providers = _integrations.cloud_providers()  # list[dict]
provider = next(p for p in providers if p["Name"] == "REST")
provider_id = provider["ID"]
```

### Provider Schema

```python
schema = _integrations.cloud_config_schema(provider_id)
tag_schema = _integrations.cloud_config_tag_schema(provider_id)
description = _integrations.cloud_config_description(provider_id)
```

***

## Instances

```python
instances = _integrations.cloud_connector_instances()
instance = next(i for i in instances if i["Config"]["name"] == "My REST Connector")
instance_id = instance["InstanceID"]

# Create
instance = _integrations.cloud_connector_create(
    provider_id=provider_id,
    provider_config={"name": "My REST Connector", "url": "https://example.com/api"},
)
instance_id = instance["InstanceID"]

# Read details
details = _integrations.cloud_connector_instance_details(instance_id)

# Enable / disable
_integrations.cloud_connector_enable_disable(instance_id, enable=True)

# Update
_integrations.cloud_connector_instance_update(
    integration_instance_id=instance_id,
    update_body={"Config": {"name": "...", "url": "..."}},
)

# Delete
_integrations.cloud_connector_instance_delete(instance_id)

# Purge persistence layer
_integrations.purge_persistence_layer(instance_id)
```

***

## Topics (REST API, 3.x and 4.0.x)

Topics define data flows. Outbound = local -> remote. Inbound = remote -> local.

```python
topics = _topics.list_topics(instance_id)

topic = _topics.create_topic(
    integration_instance_id=instance_id,
    is_local_to_remote=True,
    local_name="plant/pump/01/pressure",
    remote_name="pump-pressure",
    enabled=True,
    description="Pump pressure",
    config={},
)
topic_id = topic["TopicID"]

_topics.update_topic(
    integration_instance_id=instance_id,
    integration_topic_id=topic_id,
    local_name="plant/pump/01/pressure",
    remote_name="pump-pressure-v2",
    is_local_to_remote=True,
    enabled=True,
)

_topics.enable_disable_topic(instance_id, topic_id, enable=True)
_topics.enable_disable_multiple_topics(instance_id, [topic_id_1, topic_id_2], enable=False)

_topics.remove_topic(instance_id, topic_id)
_topics.delete_multiple_topics(instance_id, [topic_id_1, topic_id_2])

# Import from DeviceHub
_topics.import_from_dh(instance_id, device_id=device_id, wildcard=False)

# CSV export / import
csv_text = _topics.download_csv(instance_id)
_topics.upload_csv(instance_id, file_path="/path/to/topics.csv")
```

***

## Subscriptions (GraphQL API, 4.0.x only)

```python
subs = _gql_streaming.list_subscriptions(instance_id)

_gql_streaming.create_subscriptions(
    instance_id=instance_id,
    subscriptions=[{
        "LocalName": "plant/pump/01/pressure",
        "RemoteName": "pump-pressure",
        "IsLocalToRemote": True,
        "Enabled": True,
        "Description": "",
        "Config": {},
    }],
)

# Each entry in update must include TopicID
_gql_streaming.update_subscriptions(
    instance_id=instance_id,
    subscriptions=[{"TopicID": topic_id, "RemoteName": "pump-pressure-v2", ...}],
)

_gql_streaming.enable_subscriptions(instance_id, [topic_id])
_gql_streaming.disable_subscriptions(instance_id, [topic_id])
_gql_streaming.delete_subscriptions(instance_id, [topic_id])

# Bulk
_gql_streaming.enable_all_topics(instance_id)
_gql_streaming.disable_all_topics(instance_id)
_gql_streaming.delete_all_topics(instance_id)

# Bulk import from other services
_gql_streaming.import_topics(
    instance_id=instance_id,
    device_hub=[{"deviceId": device_id, "tagIds": [tag_id]}],
    digital_twins=[{"instanceId": dt_instance_id}],
    analytics=[{"processorId": proc_id}],
)

# Instance management (GraphQL)
instances = _gql_streaming.list_instances_gql()
_gql_streaming.update_instance_gql(instance_id, config="{...}", error_logging=False)
result = _gql_streaming.clear_instance_persistence_layer(instance_id)
_gql_streaming.delete_instance_gql(instance_id)

# CSV
csv = _gql_streaming.download_subscriptions_csv([instance_id])
```

***

## Object Storage (4.0.x only)

```python
status = _object_storage.get_cloud_storage_availability()
providers = _object_storage.get_cloud_storage_providers()
jobs = _object_storage.get_cloud_storage_jobs()
```

***

## Dashboard Stats

```python
stats = _dashboard.all_stats(instance_id)
_dashboard.mean_disk_usage(instance_id)
_dashboard.persistence_size(instance_id)
_dashboard.live_count(instance_id)
_dashboard.drop_count(instance_id)
```

***

## Prometheus Metrics

```python
metrics = _integrations.prometheus_metrics()  # str, raw Prometheus exposition format
```

***

## License

Copyright (c) Litmus Automation Inc.