How to Authenticate with Client Credentials (OAuth 2.0)

TL;DR

Configure OAuth 2.0 client credentials authentication for your REST API connection using the oaa-runner framework’s built-in token management.

Note: For simpler API key authentication, see How to Authenticate with API Keys.

Prerequisites

  • OAuth 2.0 client credentials from your API provider (client_id and client_secret)

  • Token URL endpoint (OAuth 2.0 token endpoint)

  • Understanding of OAuth 2.0 client credentials flow

  • A REST API source already configured (see How to Add a REST API Source)

When to Use Client Credentials vs API Keys

Feature

Client Credentials (OAuth 2.0)

API Keys

Complexity

More complex setup

Simpler

Token Expiration

Auto-refreshes tokens

Static key (manual rotation)

Security

Better for production

Suitable for simpler APIs

Framework Support

Built-in via oaa.modules.oauth_creds

Manual header/param setup

Best For

Enterprise APIs

Simple REST APIs

Steps

Step 1: Store Credentials Securely

Create or update .env file in your integration directory:

# .env (never commit this file!)
OAUTH2_CLIENT_ID=your_client_id_here
OAUTH2_CLIENT_SECRET=your_client_secret_here
OAUTH2_TOKEN_URL=https://api.example.com/oauth/token

Add .env to .gitignore:

echo ".env" >> .gitignore

Step 2: Create Your API Client Module

Create modules/my_api_client.py and import get_token from oaa-runner:

import os
import requests
from dotenv import load_dotenv
from oaa.modules.oauth_creds import get_token
from oaa.modules.lazy_dicts import lazy_dict_table

load_dotenv()

class MyAPIClient:
    def __init__(self, client_id=None, client_secret=None, token_url=None, **kwargs):
        """Initialize with OAuth 2.0 client credentials"""
        self.client_id = client_id or os.getenv('OAUTH2_CLIENT_ID')
        self.client_secret = client_secret or os.getenv('OAUTH2_CLIENT_SECRET')
        self.token_url = token_url or os.getenv('OAUTH2_TOKEN_URL')
        self.base_url = kwargs.get('base_url')
        
        if not all([self.client_id, self.client_secret, self.token_url]):
            raise ValueError("Client ID, secret, and token URL are required")
        
        self._session = None
    
    @property
    def session(self):
        """
        Lazy-load session with OAuth2 token management.
        
        The get_token() function handles:
        - Token caching in memory
        - Automatic token refresh when expired
        - Bearer token header setup
        """
        if self._session is None:
            self._session = get_token(
                client_id=self.client_id,
                client_secret=self.client_secret,
                token_url=self.token_url
            )
        
        return self._session
    
    def fetch(self, resource, **params):
        """
        Fetch records from API endpoint.
        Session automatically includes Bearer token.
        """
        url = f"{self.base_url}{resource}"
        
        try:
            response = self.session.get(url, params=params, timeout=30)
            response.raise_for_status()
            
            data = response.json()
            items = data.get('items', data.get('data', data.get('results', [])))
            
            # Yield one at a time for memory efficiency
            yield from items
            
        except requests.RequestException as e:
            raise RuntimeError(f"API request failed: {e}")


def fetch(connection=None, source=None, **kwargs):
    """
    Fetch the data. This is the function that is called by oaa-runner.
    
    Returns lazy_dict_table for memory-efficient processing.
    """
    api = MyAPIClient(**connection.params)
    resource = source.params.get('resource', '')
    records = api.fetch(resource=resource)
    
    return lazy_dict_table(records)

Step 3: Configure in config.yaml

Add the connection with your token URL:

connections:
  my_api:
    connection_type: custom
    custom_module: my_api_client
    params:
      base_url: https://api.example.com
      OAUTH2_TOKEN_URL: https://api.example.com/oauth/token
      # Client ID and secret come from environment variables

The framework will automatically use OAUTH2_CLIENT_ID and OAUTH2_CLIENT_SECRET from your environment.

Step 4: Configure Your Source

Add a source that uses the OAuth2 connection:

sources:
  users:
    destination: LocalUser
    source_type: custom
    connection: my_api
    params:
      resource: /v1/users
    field_mapping:
      user_id:
        destination_field: unique_id
      email:
        destination_field: email
      name:
        destination_field: name

Step 5: Test Authentication

Validate your configuration:

uv tool run --from oaa-runner oaa app validate --config=config.yaml

Run the integration to test the connection:

uv tool run --from oaa-runner oaa app run --config=config.yaml --save-json

Check logs for any authentication errors. Successful output should show Bearer token being automatically included in requests.

Understanding the get_token Decorator

The get_token() function from oaa.modules.oauth_creds includes a @get_or_refresh decorator that:

  1. Caches tokens in memory - Reduces token requests

  2. Auto-refreshes when expired - Adds 60-second buffer before expiration

  3. Returns configured session - Includes Bearer token header automatically

  4. Handles errors gracefully - Logs and raises exceptions for debugging

This means your API client doesn’t need to manually manage token lifecycle.

Production Example: BeyondTrust

From exeter/beyondtrust/modules/beyondtrust.py:

@property
def session(self, **kwargs):
    if self._session == None:
        self._session = get_token(self._client_id, self._client_secret,
                                  token_url=self._token_url)
        
        # Perform initial authentication
        res = self._session.post(self.make_url('/Auth/SignAppIn'))
        self._session_called += 1
    
    return self._session

This pattern shows:

  • Lazy-loading session only when needed (@property)

  • Passing credentials to get_token()

  • Session is automatically configured with Bearer token

  • Can immediately use session for API calls

Troubleshooting

Error: 401 Unauthorized

  • Verify client_id and client_secret are correct

  • Check that token_url is correct

  • Ensure credentials haven’t been revoked by the API provider

Error: OAUTH2_CLIENT_ID not found in environment

  • Verify .env file exists in integration directory

  • Check variable names match exactly (case-sensitive)

  • Consider using required_env in your connection config for automatic validation — see How to Use required_env

Error: Connection refused to token URL

  • Verify token URL is correct and accessible

  • Check network connectivity to token endpoint

  • Confirm firewall isn’t blocking requests

Token expiration errors

  • The framework handles this automatically via @get_or_refresh

  • If still seeing issues, ensure expires_in is in token response

  • Check logs for token refresh attempts

Next Steps

See Also

  • Different Grant Types: For password-based OAuth flows, a separate guide covers get_token_password()

  • API Key Authentication: See How to Authenticate with API Keys for simpler use cases