This is your first hands-on lesson with the Python SDK: create a client, send an event, and receive it. Make sure you have the SDK installed (see the Python SDK overview).

Create a Client

The SDK provides pattern-specific async clients: AsyncPubSubClient for Events/Events Store, AsyncQueuesClient for Queues, and AsyncCQClient for Commands/Queries.

connect.py

import asyncio
from kubemq import AsyncPubSubClient

async def main():
    async with AsyncPubSubClient(address="localhost:50000") as client:
        print("Connected to KubeMQ")

asyncio.run(main())

Or using other pattern-specific clients:

queues_client.py

import asyncio
from kubemq import AsyncQueuesClient

async def main():
    async with AsyncQueuesClient(address="localhost:50000") as client:
        # Use for Queues operations
        pass

asyncio.run(main())

Send Your First Event

send_event.py

import asyncio
from kubemq import AsyncPubSubClient, EventMessage

async def main():
    async with AsyncPubSubClient(address="localhost:50000") as client:
        await client.publish_event(
            EventMessage(channel="notifications", body=b"hello kubemq")
        )
        print("Event sent!")

asyncio.run(main())

Receive Events

receive_events.py

import asyncio
from kubemq import AsyncPubSubClient, AsyncCancellationToken, EventsSubscription

async def main():
    async with AsyncPubSubClient(address="localhost:50000") as client:
        token = AsyncCancellationToken()

        async for event in client.subscribe_to_events(
            subscription=EventsSubscription(
                channel="notifications",
                on_receive_event_callback=lambda e: None,
                on_error_callback=lambda e: print(f"Error: {e}"),
            ),
            cancellation_token=token,
        ):
            print(f"Received: {event.body.decode('utf-8')}")

asyncio.run(main())

Configuration Options

Option	Type	Default	Description
`address`	`str`	`"localhost:50000"`	KubeMQ server gRPC address
`client_id`	`str`	hostname	Unique client identifier
`auth_token`	`str`	`""`	Authentication token
`tls`	`TLSConfig`	disabled	TLS configuration
`max_send_size`	`int`	`4194304`	Max send message size (bytes)
`max_receive_size`	`int`	`4194304`	Max receive message size (bytes)
`auto_reconnect`	`bool`	`True`	Auto-reconnect on connection loss
`reconnect_interval_seconds`	`int`	`1`	Seconds between reconnect attempts
`log_level`	`int`	`None`	Python logging level

Error Handling

All SDK errors extend KubeMQError:

error_handling.py

import asyncio
from kubemq import (
    AsyncCQClient, CommandMessage,
    KubeMQError, KubeMQConnectionError, KubeMQTimeoutError,
)

async def main():
    try:
        async with AsyncCQClient(address="localhost:50000") as client:
            await client.send_command(
                CommandMessage(channel="orders", body=b"data", timeout_in_seconds=5)
            )
    except KubeMQConnectionError as e:
        print(f"Connection failed: {e}")
    except KubeMQTimeoutError as e:
        print(f"Operation timed out: {e}")
    except KubeMQError as e:
        print(f"KubeMQ error: {e}")

asyncio.run(main())

Next Steps

Python SDK Reference — full API documentation
Python SDK Examples — complete examples for all patterns
GitHub Repository — source code and issues

Send Your First Message