Skip to content


Quick Start

Connecting to a Diffusion server and listening to ping messages:

#  Copyright (c) 2021 Push Technology Ltd., All Rights Reserved.
#  Use is subject to license terms.
#  NOTICE: All information contained herein is, and remains the
#  property of Push Technology. The intellectual and technical
#  concepts contained herein are proprietary to Push Technology and
#  may be covered by U.S. and Foreign Patents, patents in process, and
#  are protected by trade secret or copyright law.

import asyncio
import logging
import os

import diffusion
import argparse

async def my_main(*, server_url: str, principal: str, password: str):

    credentials = diffusion.Credentials(password)

    async with diffusion.Session(server_url, principal, credentials) as session:

        await asyncio.sleep(10)

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
        default=os.getenv("SERVER_URL", "ws://your-diffusion-server:8080"),
URL of Diffusion Server. You can use a local/hosted URL or a Diffusion Cloud service URL,
e.g. ws://""",
        default=os.getenv("PRINCIPAL", "admin"),
        help="Username to connect with",
        default=os.getenv("PASSWORD", "password"),
        help="Password to connect with",

Notes on Usage

The Python SDK for Diffusion is built using the asynchronous programming paradigm, and concretely on top of the asyncio library.

In practice, it means that all the SDK code needs to be wrapped in a coroutine function (declared with async def) and executed using For more detailed information about using asyncio, we recommend this overview.