Use for Python

Real-time application development framework for Python. XConn enables backend APIs that are fast, support PubSub and are secure.

Prerequisites

Before running a client, you need have a WAMP router running. The client needs to connect to a router to send and receive messages, so this step is essential.

We recommend using the NXT router, a lightweight and high-performance WAMP router built for flexibility and speed.

Getting started

Install xconn from pypi

uv venv
uv pip install xconn

Save the below code in sample.py

from xconn import App

app = App()

@app.register("io.xconn.hello")
async def my_procedure(first_name: str, last_name: str, age: int):
    print(first_name + " " + last_name + " " + str(age))
    return first_name, last_name, age


@app.subscribe("io.xconn.publish")
async def my_topic():
    print("received event...")

run the app with xcorn command (note: this automatically starts a debug router)

uv run xcorn sample:app --start-router
starting server on 127.0.0.1:8080
connected realm1
Registered procedure io.xconn.hello
Subscribed topic io.xconn.publish

Client

This library provides two types of clients: * Client for synchronous operations. * AsyncClient for asynchronous operations.

Sync Client

from xconn.session import Session
from xconn.client import connect_anonymous

session: Session = connect_anonymous("ws://localhost:8080/ws", "realm1")

Once the session is established, you can perform WAMP actions. Below are examples of all four WAMP operations:

Subscribe to a topic

from xconn.types import Event
from xconn.session import Session

def example_subscribe(session: Session):
    session.subscribe("io.xconn.example", event_handler)


def event_handler(event: Event):
    print(f"Event Received: args={event.args}, kwargs={event.kwargs}, details={event.details}")

Publish to a topic

from xconn.session import Session

def example_publish(session: Session):
    session.publish("io.xconn.example", ["test"])

Register a procedure

from xconn.session import Session
from xconn.types import Invocation

def example_register(session: Session):
    session.register("io.xconn.example", invocation_handler)


def invocation_handler(invocation: Invocation):
    print(f"Received args={invocation.args}, kwargs={invocation.kwargs}, details={invocation.details}")

Call a procedure

from xconn.session import Session

def example_call(session: Session):
    session.call("io.xconn.example", ["1", "2"], {"key": "value"})

Authentication

Authentication is straightforward.

Ticket Auth

from xconn.session import Session
from xconn.client import connect_ticket

session: Session = connect_ticket("ws://localhost:8080/ws", "realm1", "authid", "ticket")

Challenge Response Auth

from xconn.session import Session
from xconn.client import connect_wampcra

session: Session = connect_wampcra("ws://localhost:8080/ws", "realm1", "authid", "secret")

Cryptosign Auth

from xconn.session import Session
from xconn.client import connect_cryptosign

session: Session = connect_cryptosign("ws://localhost:8080/ws", "realm1", "authid", "d850fff4ff199875c01d3e652e7205309dba2f053ae813c3d277609150adff13")

Async Client

from xconn import run
from xconn.async_session import AsyncSession
from xconn.async_client import connect_anonymous

async def main():
    session: AsyncSession = await connect_anonymous("ws://localhost:8080/ws", "realm1")

if __name__ == "__main__":
    run(main())

Once the session is established, you can perform WAMP actions. Below are examples of all 4 WAMP operations:

Subscribe to a topic

from xconn.types import Event
from xconn.async_session import AsyncSession

async def example_subscribe(session: AsyncSession):
    await session.subscribe("io.xconn.example", event_handler)


async def event_handler(event: Event):
    print(f"Event Received: args={event.args}, kwargs={event.kwargs}, details={event.details}")

Publish to a topic

from xconn.async_session import AsyncSession

async def example_publish(session: AsyncSession):
    await session.publish("io.xconn.example", ["test"])

Register a procedure

from xconn.types import Invocation
from xconn.async_session import AsyncSession

async def example_register(session: AsyncSession):
    await session.register("io.xconn.example", invocation_handler)


async def invocation_handler(invocation: Invocation):
    print(f"Received args={invocation.args}, kwargs={invocation.kwargs}, details={invocation.details}")

Call a procedure

from xconn.async_session import AsyncSession

async def example_call(session: AsyncSession):
    await session.call("io.xconn.example", ["1", "2"], {"key": "value"})

Authentication

Authentication is straightforward.

Ticket Auth

from xconn import run
from xconn.async_session import AsyncSession
from xconn.async_client import connect_ticket

async def connect():
    session: AsyncSession = await connect_ticket("ws://localhost:8080/ws", "realm1", "authid", "ticket")

if __name__ == "__main__":
    run(connect())

Challenge Response Auth

from xconn import run
from xconn.async_session import AsyncSession
from xconn.async_client import connect_wampcra

async def connect():
    session: AsyncSession = await connect_wampcra("ws://localhost:8080/ws", "realm1", "authid", "secret")

if __name__ == "__main__":
    run(connect())

Cryptosign Auth

from xconn import run
from xconn.async_session import AsyncSession
from xconn.async_client import connect_cryptosign

async def connect():
    session: AsyncSession = await connect_cryptosign("ws://localhost:8080/ws", "realm1", "authid", "d850fff4ff199875c01d3e652e7205309dba2f053ae813c3d277609150adff13")

if __name__ == "__main__":
    run(connect())

look at examples directory for more examples