---
title: Example: Integration
slug: sdk/example-usages/integration
docTags: 
createdAt: 2025-03-20T19:52:46.556Z
---

End-to-end walkthrough: configure a cloud connector instance, add topics that map device tags to remote names, enable streaming.

## Setup

```python
from litmussdk.integrations.common import _integrations, _topics
```

## 1. List Available Providers

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

# What does this provider need for instance config?
schema = _integrations.cloud_config_schema(provider_id)

# What does each topic need?
tag_schema = _integrations.cloud_config_tag_schema(provider_id)
```

## 2. Create an Instance

```python
instance = _integrations.cloud_connector_create(
    provider_id=provider_id,
    provider_config={
        "name": "Plant-Floor-REST",
        "url": "https://my-cloud.example.com/ingest",
        "auth": {"type": "bearer", "token": "..."},
    },
)
instance_id = instance["InstanceID"]
```

## 3. Add Topics (Outbound)

Each topic maps a local Litmus tag/topic to a remote name in the cloud system.

```python
topic_a = _topics.create_topic(
    integration_instance_id=instance_id,
    is_local_to_remote=True,                # outbound
    local_name="plant/pump/01/pressure",
    remote_name="pump-pressure",
    enabled=True,
    description="Pump pressure",
    config={},
)

topic_b = _topics.create_topic(
    integration_instance_id=instance_id,
    is_local_to_remote=True,
    local_name="plant/pump/01/flow_rate",
    remote_name="pump-flow",
    enabled=True,
    description="Pump flow rate",
    config={},
)
```

## 4. Enable the Instance

```python
_integrations.cloud_connector_enable_disable(instance_id, enable=True)
```

Topics defined with `enabled=True` start streaming immediately.

## 5. Toggle Topics in Bulk

```python
all_topics = _topics.list_topics(instance_id)
all_topic_ids = [t["TopicID"] for t in all_topics]

# Disable all temporarily (e.g. during a maintenance window)
_topics.enable_disable_multiple_topics(instance_id, all_topic_ids, enable=False)

# Re-enable
_topics.enable_disable_multiple_topics(instance_id, all_topic_ids, enable=True)
```

## 6. Import Topics from DeviceHub

Bulk-import all tags on a device into this instance as topics:

```python
_topics.import_from_dh(
    integration_instance_id=instance_id,
    device_id=device_id,
    wildcard=False,   # or True to also include sub-topics
)
```

## 7. CSV Export / Import

```python
csv_text = _topics.download_csv(instance_id)
with open("topics.csv", "w") as f:
    f.write(csv_text)

# Apply on another Edge:
_topics.upload_csv(instance_id, file_path="topics.csv")
```

## 8. GraphQL Subscription API (4.0.x only)

The same flow is also available via GraphQL with finer control. This is the recommended path on 4.0.x:

```python
from litmussdk.integrations.common import _gql_streaming

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

## 9. Cleanup

```python
_topics.delete_multiple_topics(instance_id, all_topic_ids)
_integrations.cloud_connector_instance_delete(instance_id)
```

## See Also

- [Managing Connections](docId\:vwtFo9SXcMRbFX9cAfENw) - LE vs LEM