Installation

Follow the steps to set up and start using the Python ADK in your project.

1. Install the ADK Package

Create a Python project and install the ADK package:

pip install art_adk

2. Import the ADK into your project

Import the ADK in your project files to use the features provided by ART:

# Synchronous (blocking)
from art_adk import Adk

# Asynchronous (non-blocking)
from art_adk import AsyncAdk

3. Generate Client Secret

Once after gaining access to the ART Live Dashboard, generate the Client Secret and add it in the root of your project directory. The client credential JSON will look something like this:

{
    "Client-ID": "xxxxxxxxxx",
    "Client-Secret": "xxxxxxxxxxx",
    "Org-Title": "YOUR_ORG",
    "ProjectKey": "YOUR_PROJECT_KEY",
    "Environment": "YOUR_ENV_NAME"
};

important

The adk-service.json needs to be updated with ProjectKey and Environment, manually. To learn how to obtain the values, click here

4. Get Passcode

Before authenticating, obtain a passcode through the WebSocket secure line. Initialize the ADK without an AuthToken, then request the passcode:

import os
from art_adk import Adk, AsyncAdk

# Configure without passcode
config = {
    "ROOT": os.getcwd(),
    "Uri": "ws.arealtimetech.com"  # Replace with ART server WebSocket URI
}

# Synchronous
adk = Adk(config)
adk.connect()

user_payload = {
    "Username": "your_username",  # Replace with your username
    "FirstName": "first_name",    # Replace with your first name
    "LastName": "last_name"       # Replace with your last name
}

try:
    passcode = adk.get_security_code(user_payload)
    print("Passcode obtained:", passcode)
except Exception as error:
    print("Failed to get passcode:", error)

# Asynchronous
async def get_passcode():
    adk = AsyncAdk(config)
    await adk.connect()

    user_payload = {
        "Username": "your_username",  # Replace with your username
        "FirstName": "first_name",    # Replace with your first name
        "LastName": "last_name"       # Replace with your last name
    }

    try:
        passcode = await adk.get_security_code(user_payload)
        print("Passcode obtained:", passcode)
        return passcode
    except Exception as error:
        print("Failed to get passcode:", error)

5. Authenticate

Once the Client Credentials are added to your project and you have obtained a passcode, the Client needs to be authenticated

import os
from art_adk import Adk, AsyncAdk

config = {
    "ROOT": os.getcwd(),
    "Uri": "ws.arealtimetech.com",  # Replace with ART server WebSocket URI
    "AuthToken": passcode  # Use the passcode obtained from step 4
}

# Synchronous - Context manager (auto connect/disconnect)
with Adk(config) as adk:
    # Your logic here
    pass

# Synchronous - Manual connection
adk = Adk(config)
adk.connect()
# Your logic here
adk.disconnect()

# Asynchronous - Context manager (auto connect/disconnect)
async with AsyncAdk(config) as adk:
    # Your logic here
    pass

# Asynchronous - Manual connection
adk = AsyncAdk(config)
await adk.connect()
# Your logic here
await adk.disconnect()

6. Subscribe to Channel

A Channel is a communication pathway that provides passage for data transfer. Subscribing to channel will allow you to get started with using the realtime messaging in your application.

# Synchronous
subscription = adk.subscribe("your-channel-name")

# Asynchronous
subscription = await adk.subscribe("your-channel-name")

You can learn more about Channel Configuration from here

7. Push Messages

Messages are pushed to a channel using the push() ADK method. This method allows you to send various event types with associated data, and optionally target specific users or groups within the channel for more granular delivery.

# Define the message payload
payload = {"content": "Hello from ART ADK!"}

# Optionally define specific target users within the channel
target_users = ["username1", "username2"]

# Synchronous
try:
    response = subscription.push("message", payload, {
        "to": target_users  # Target specific recipients within the channel
    })
    print("Message acknowledged by ART:", response)
except Exception as error:
    print("Failed to push message:", error)

# Asynchronous
try:
    response = await subscription.push("message", payload, {
        "to": target_users  # Target specific recipients within the channel
    })
    print("Message acknowledged by ART:", response)
except Exception as error:
    print("Failed to push message:", error)

8. Listen to All Events and Messages

The listen() method allows you to capture all events and messages published within the subscribed channel. This is a catch-all listener that delivers every incoming payload, regardless of event type.

# Listen to all events and messages in the channel
def message_handler(data):
    print("Received:", data)

subscription.listen(message_handler)

note

You can refer the Quick Start Guide to understand the steps better.