# Configure Python Clients for OAuth/OIDC on Confluent Cloud Configure Python Kafka and Schema Registry clients for OAuth/OIDC by using SASL/OAUTHBEARER and a token refresh callback that requests JWT tokens from your identity provider. Install `confluent-kafka` 2.0.0 or later and implement the `oauth_token_refresh_cb` callback function to handle token requests. ## Prerequisites Before you begin, ensure you have the following prerequisites: - Python 3.8 or later - [confluent-kafka-python](https://pypi.org/project/confluent-kafka/): 2.0.0 or later - OAuth authentication is configured - Client configuration details - **Client ID**: Identifier for requesting OAuth tokens - **Client secret**: Secret for requesting OAuth tokens - **Token endpoint URL**: Location of access tokens - **Scopes**: The permissions your application requires as defined in your identity provider. For example, `kafka:read kafka:write` for Kafka, `schema-registry` for Schema Registry. - **Cluster ID**: Your cluster ID based on cluster type - **Kafka cluster ID** (`lkc-xxxxx`): Your Kafka cluster identifier - **Schema Registry logical cluster ID** (`lsrc-xxxxx`): For Schema Registry operations - **Identity pool ID** (`pool-xxxxx`): Optional Install the latest version with OAuth support: ```bash pip install confluent-kafka ``` ## Configure Kafka Python clients Python Kafka clients authenticate to Confluent Cloud clusters using the OAuth 2.0 protocol with a callback function approach. The client passes OAuth configuration parameters to your callback function, which handles the token request and returns the access token. ### OAuth callback function The Python client requires an OAuth callback function with robust error handling: ```python import requests import json from confluent_kafka import KafkaException def oauth_token_refresh_cb(oauth_config): """ OAuth callback function for Python Kafka client Args: oauth_config: OAuth configuration dictionary Returns: tuple: (access_token, expiration_time_ms) """ try: payload = { 'grant_type': 'client_credentials', 'client_id': oauth_config.get('sasl.oauthbearer.client.id'), 'client_secret': oauth_config.get('sasl.oauthbearer.client.secret'), 'scope': oauth_config.get('sasl.oauthbearer.scope') } response = requests.post( oauth_config.get('sasl.oauthbearer.token.endpoint.url'), data=payload ) response.raise_for_status() token = response.json() # Return the token and its expiration time return token['access_token'], int(token['expires_in'] * 1000) except Exception as e: # Handle exceptions and signal failure raise KafkaException(f"OAuth token refresh failed: {e}") ``` ### Configuration example Define a single, complete configuration dictionary for your Kafka client: ```python from confluent_kafka import Producer, Consumer # Define a single, complete configuration dictionary producer_config = { 'bootstrap.servers': 'your-bootstrap-server:9092', 'security.protocol': 'SASL_SSL', 'sasl.mechanism': 'OAUTHBEARER', 'sasl.oauthbearer.token.endpoint.url': 'https:///oauth2/token', 'sasl.oauthbearer.client.id': '', 'sasl.oauthbearer.client.secret': '', 'sasl.oauthbearer.scope': 'kafka:read kafka:write', # identityPoolId may be omitted as Kafka can automatically identify the pool ID based on the OAuth token claims 'sasl.oauthbearer.extensions': 'logicalCluster=,identityPoolId=', 'oauth_cb': oauth_token_refresh_cb, 'debug': 'security,protocol,broker' # For troubleshooting } # Create producer producer = Producer(producer_config) # For consumer, use the same OAuth configuration consumer_config = { 'bootstrap.servers': 'your-bootstrap-server:9092', 'group.id': 'your-consumer-group', 'security.protocol': 'SASL_SSL', 'sasl.mechanism': 'OAUTHBEARER', 'sasl.oauthbearer.token.endpoint.url': 'https:///oauth2/token', 'sasl.oauthbearer.client.id': '', 'sasl.oauthbearer.client.secret': '', 'sasl.oauthbearer.scope': 'kafka:read kafka:write', 'sasl.oauthbearer.extensions': 'logicalCluster=,identityPoolId=', 'oauth_cb': oauth_token_refresh_cb, 'debug': 'security,protocol,broker' } # Create consumer consumer = Consumer(consumer_config) ``` ### Client assertion configuration For client assertions (KIP-1258), configure the Python client to use JWT-based authentication: ```python from confluent_kafka import Producer, Consumer # Producer configuration with client assertions (KIP-1258) producer_config = { 'bootstrap.servers': 'your-bootstrap-server:9092', 'security.protocol': 'SASL_SSL', 'sasl.mechanism': 'OAUTHBEARER', 'sasl.oauthbearer.method': 'oidc', 'sasl.oauthbearer.token.endpoint.url': 'https:///oauth/token', 'sasl.oauthbearer.client.id': '', 'sasl.oauthbearer.grant.type': 'urn:ietf:params:oauth:grant-type:jwt-bearer', # Client assertion configuration - points to the private key used to sign the assertion 'sasl.oauthbearer.assertion.private.key.file': '/path/to/private_key.pem', 'sasl.oauthbearer.assertion.claim.iss': '', 'sasl.oauthbearer.assertion.claim.aud': 'https:///oauth/token', 'sasl.oauthbearer.extensions': 'logicalCluster=,identityPoolId=', } # Create producer producer = Producer(producer_config) # Consumer configuration with client assertions (KIP-1258) consumer_config = { 'bootstrap.servers': 'your-bootstrap-server:9092', 'group.id': 'your-consumer-group', 'security.protocol': 'SASL_SSL', 'sasl.mechanism': 'OAUTHBEARER', 'sasl.oauthbearer.method': 'oidc', 'sasl.oauthbearer.token.endpoint.url': 'https:///oauth/token', 'sasl.oauthbearer.client.id': '', 'sasl.oauthbearer.grant.type': 'urn:ietf:params:oauth:grant-type:jwt-bearer', # Client assertion configuration 'sasl.oauthbearer.assertion.private.key.file': '/path/to/private_key.pem', 'sasl.oauthbearer.assertion.claim.iss': '', 'sasl.oauthbearer.assertion.claim.aud': 'https:///oauth/token', 'sasl.oauthbearer.extensions': 'logicalCluster=,identityPoolId=', } # Create consumer consumer = Consumer(consumer_config) ``` ### Test your configuration 1. **Test with a simple producer**: ```python # Test producer producer = Producer(producer_config) def delivery_report(err, msg): if err is not None: print(f'Message delivery failed: {err}') else: print(f'Message delivered to {msg.topic()} [{msg.partition()}] - OAuth is working!') producer.produce('test-topic', 'test-message', callback=delivery_report) producer.flush() ``` 2. **Check for common errors**: - **“SASL authentication failed”** - Check your OAuth credentials and endpoint - **“Invalid token”** - Verify your callback function is returning a valid token - **“Connection timeout”** - Check your bootstrap servers and network connectivity 3. **Verify token refresh** - The client should automatically refresh tokens when they expire ## Configure Schema Registry Python clients The `SchemaRegistryClient` authenticates using a built-in OAuth mechanism. You only need to provide the correct configuration parameters, and the client handles the token request automatically. It **does not** use the `oauth_cb` callback. ### Required parameters The following parameters must be included in the `schema.registry.config` dictionary when creating a client. | Parameter | Description | |-----------------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | `url` | The endpoint for your Schema Registry instance. | | `basic.auth.user.info` | The API key and secret for Schema Registry, in the format `:`. | | `bearer.auth.credentials.source` | **Must be set to \`\`OAUTHBEARER\`\`**. | | `bearer.auth.client.id` | The client ID from your identity provider. | | `bearer.auth.client.secret` | The client secret from your identity provider. | | `bearer.auth.issuer.endpoint.url` | The token endpoint URL of your identity provider. | | `bearer.auth.scope` | The required permissions scope (for example, `schema:read`). | | `bearer.auth.logical.cluster` | The Schema Registry’s logical cluster ID (for example, `lsrc-xxxxx`). | | `bearer.auth.identity.pool.id` | The identity pool ID.

For `confluent-kafka` version 2.15.0 or later,
you can specify a comma-separated list of pool IDs
or omit it entirely to use auto pool mapping. If omitted,
Confluent Cloud automatically maps the token to all identity
pools whose filters match the token claims.
For details, see [Use auto pool mapping with OAuth identity pools](../identity-pools.md#oauth-auto-pool-mapping). | ### Configuration example ```python from confluent_kafka.schema_registry import SchemaRegistryClient from confluent_kafka.serialization import StringSerializer from confluent_kafka.avro import AvroSerializer schema_registry_conf = { 'url': 'https://', # OAuth-specific configuration for Schema Registry client 'bearer.auth.credentials.source': 'OAUTHBEARER', 'bearer.auth.issuer.endpoint.url': 'https:///oauth2/token', 'bearer.auth.client.id': '', 'bearer.auth.client.secret': '', 'bearer.auth.scope': 'schema:read', 'bearer.auth.logical.cluster': '', 'bearer.auth.identity.pool.id': '' } schema_registry_client = SchemaRegistryClient(schema_registry_conf) avro_serializer = AvroSerializer(schema_registry_client, user_schema_string, conf={'auto.register.schemas': False}) ``` ## Google OIDC integration For Google OIDC integration with Python clients: ```python # Google OIDC configuration for Kafka google_oauth_config = { 'bootstrap.servers': 'your-bootstrap-server:9092', 'security.protocol': 'SASL_SSL', 'sasl.mechanism': 'OAUTHBEARER', 'sasl.oauthbearer.token.endpoint.url': 'https://oauth2.googleapis.com/token', 'sasl.oauthbearer.client.id': 'your-google-client-id', 'sasl.oauthbearer.client.secret': 'your-google-client-secret', 'sasl.oauthbearer.scope': 'https://www.googleapis.com/auth/cloud-platform', 'sasl.oauthbearer.extensions': 'logicalCluster=,identityPoolId=', 'oauth_cb': oauth_token_refresh_cb, 'debug': 'security,protocol,broker' } # Create producer with Google OIDC configuration producer = Producer(google_oauth_config) ``` ## Troubleshoot Python OAuth clients Common issues and solutions for Python OAuth clients: ### Authentication failures - Verify client ID and secret are correct - Check token endpoint URL is accessible - Ensure logical cluster ID is valid - Validate identity pool ID if used ### Network issues - Confirm network connectivity to OAuth provider - Check firewall rules allow OAuth traffic - Verify SSL certificate validation ### Configuration issues - Ensure all required parameters are provided - Validate OAuth callback function signature - Check timeout values are reasonable ### Debug logging Enable debug logging for OAuth troubleshooting by adding the `debug` parameter to your configuration: ```python # Add to your configuration config = { # ... your OAuth config 'debug': 'security,protocol,broker' } ``` This provides detailed librdkafka logs for authentication issues. ## Related content - [Configure Kafka Clients for OAuth 2.0 Authentication in Confluent Cloud](overview.md#oauth-client-configuration-overview) - [Use OAuth/OIDC to Authenticate to Confluent Cloud](../overview.md#oauth-overview) - [OAuth Configuration Reference for Confluent Cloud Clients](configuration-reference.md#oauth-configuration-reference) - [Troubleshoot OAuth/OIDC Issues on Confluent Cloud](../troubleshooting.md#oauth-troubleshooting)