XConn

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

Getting started

Install xconn from pypi

uv venv
uv pip install git+ssh://git@github.com/xconnio/xconn-python.git
source .venv/bin/activate

Save the below code in sample.py

fromxconnimportAppapp=App()
@app.register("io.xconn.hello")asyncdefmy_procedure(first_name: str, last_name: str, age: int):
print(first_name+" "+last_name+" "+str(age))
returnfirst_name, last_name, age@app.subscribe("io.xconn.publish")asyncdefmy_topic():
print("received event...")

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

(xconn-python) om26er@Home-PC:~$ 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

fromxconn.sessionimportSessionfromxconn.clientimportconnect_anonymoussession: 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

fromxconn.typesimportEventfromxconn.sessionimportSessiondefexample_subscribe(session: Session):
session.subscribe("io.xconn.example", event_handler)
print("Subscribed to topic 'io.xconn.example'")
defevent_handler(event: Event):
print(f"Event Received: args={event.args}, kwargs={event.kwargs}, details={event.details}")

Publish to a topic

fromxconn.sessionimportSessiondefexample_publish(session: Session):
session.publish("io.xconn.example", ["test"])
print("Published to topic io.xconn.example")

Register a procedure

fromxconn.sessionimportSessionfromxconn.typesimportInvocationdefexample_register(session: Session):
session.register("io.xconn.example", invocation_handler)
definvocation_handler(invocation: Invocation):
print(f"Received args={invocation.args}, kwargs={invocation.kwargs}, details={invocation.details}")

Call a procedure

fromxconn.sessionimportSessiondefexample_call(session: Session):
result=session.call("io.xconn.example", ["1", "2"], {"key": "value"})
print(f"Received args={result.args}, kwargs={result.kwargs}, details={result.details}")

Authentication

Authentication is straightforward.

Ticket Auth

fromxconn.sessionimportSessionfromxconn.clientimportconnect_ticketsession: Session=connect_ticket("ws://localhost:8080/ws", "realm1", "authid", "ticket")

Challenge Response Auth

fromxconn.sessionimportSessionfromxconn.clientimportconnect_wampcrasession: Session=connect_wampcra("ws://localhost:8080/ws", "realm1", "authid", "secret")

Cryptosign Auth

fromxconn.sessionimportSessionfromxconn.clientimportconnect_cryptosignsession: Session=connect_cryptosign("ws://localhost:8080/ws", "realm1", "authid", "d850fff4ff199875c01d3e652e7205309dba2f053ae813c3d277609150adff13")

Async Client

fromxconnimportrunfromxconn.async_sessionimportAsyncSessionfromxconn.async_clientimportconnect_anonymousasyncdefmain():
session: AsyncSession=awaitconnect_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

fromxconn.typesimportEventfromxconn.async_sessionimportAsyncSessionasyncdefexample_subscribe(session: AsyncSession):
awaitsession.subscribe("io.xconn.example", event_handler)
print("Subscribed to topic 'io.xconn.example'")
asyncdefevent_handler(event: Event):
print(f"Event Received: args={event.args}, kwargs={event.kwargs}, details={event.details}")

Publish to a topic

fromxconn.async_sessionimportAsyncSessionasyncdefexample_publish(session: AsyncSession):
awaitsession.publish("io.xconn.example", ["test"])
print("Published to topic io.xconn.example")

Register a procedure

fromxconn.typesimportInvocationfromxconn.async_sessionimportAsyncSessionasyncdefexample_register(session: AsyncSession):
awaitsession.register("io.xconn.example", invocation_handler)
asyncdefinvocation_handler(invocation: Invocation):
print(f"Received args={invocation.args}, kwargs={invocation.kwargs}, details={invocation.details}")

Call a procedure

fromxconn.async_sessionimportAsyncSessionasyncdefexample_call(session: AsyncSession):
result=awaitsession.call("io.xconn.example", ["1", "2"], {"key": "value"})
print(f"Received args={result.args}, kwargs={result.kwargs}, details={result.details}")

Authentication

Authentication is straightforward.

Ticket Auth

fromxconnimportrunfromxconn.async_sessionimportAsyncSessionfromxconn.async_clientimportconnect_ticketasyncdefconnect():
session: AsyncSession=awaitconnect_ticket("ws://localhost:8080/ws", "realm1", "authid", "ticket")
if__name__=="__main__":
run(connect())

Challenge Response Auth

fromxconnimportrunfromxconn.async_sessionimportAsyncSessionfromxconn.async_clientimportconnect_wampcraasyncdefconnect():
session: AsyncSession=awaitconnect_wampcra("ws://localhost:8080/ws", "realm1", "authid", "secret")
if__name__=="__main__":
run(connect())

Cryptosign Auth

fromxconnimportrunfromxconn.async_sessionimportAsyncSessionfromxconn.async_clientimportconnect_cryptosignasyncdefconnect():
session: AsyncSession=awaitconnect_cryptosign("ws://localhost:8080/ws", "realm1", "authid", "d850fff4ff199875c01d3e652e7205309dba2f053ae813c3d277609150adff13")
if__name__=="__main__":
run(connect())

look at examples directory for more examples