Token Authentication

Overview

Token authentication proves a client's identity to a KubeMQ server that has authentication enabled, without embedding a username/password or issuing per-client TLS certs. It's the mechanism you reach for in shared clusters, multi-tenant deployments, or any environment where you need to control and audit which clients are allowed to connect — the token is issued and revoked by your identity provider, not baked into the application.

The token travels as a gRPC metadata header attached to every outgoing request, set once at client construction with the auth_token argument to AsyncPubSubClient. The server validates it against its configured authentication provider before honoring any call, including the very first one. Because a static token eventually expires, KubeMQ clients also support a credential-provider hook that fetches a fresh token before each request instead of forcing you to reconnect on rotation.

Gotchas: an invalid or expired token isn't rejected until the first real call — call ping() right after connecting so failures surface as a KubeMQConnectionError immediately instead of on your first business request; never hardcode a real token in source, read it from an environment variable or secrets manager; and a static auth_token value never refreshes itself, so short-lived JWTs need a rotating credential provider, not a periodically-restarted client.

Prerequisites

• KubeMQ server running on localhost:50000
• Python SDK installed (pip install kubemq)

Code

"""Example: Token authentication — connect with an auth token."""

from __future__ import annotations

import asyncio

from kubemq import AsyncPubSubClient, EventMessage


async def main() -> None:
    async with AsyncPubSubClient(
        address="localhost:50000",
        auth_token="your-authentication-token",
        client_id="python-connection-token-auth-client",
    ) as client:
        info = await client.ping()
        print(f"Authenticated and connected to {info.host}")

        await client.publish_event(
            EventMessage(
                channel="python-connection.token-auth",
                body=b"Authenticated message",
            )
        )
        print("Message sent with authentication")


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

How It Works

auth_token is passed directly to AsyncPubSubClient and is attached as a gRPC metadata header on every request. Replace "your-authentication-token" with the JWT issued by your KubeMQ cluster. The ping() call verifies that the token is accepted before sending real traffic; a rejected token raises KubeMQConnectionError.

Related

• Python SDK Reference
• Connect
• Close