# Token Authentication (/sdks/python/how-to/connection/token-auth)



## Overview [#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 [#prerequisites]

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

## Code [#code]

```python title="token_auth.py"
"""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 [#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 [#related]

* [Python SDK Reference](/sdks/python/reference)
* [Connect](/sdks/python/tutorials/connect)
* [Close](/sdks/python/how-to/connection/close)