Skip to main content

Documentation Index

Fetch the complete documentation index at: https://mintlify.com/dvlpjrs/guMCP/llms.txt

Use this file to discover all available pages before exploring further.

​
Overview

The OAuth utilities module provides generic, reusable functions for handling OAuth 2.0 authentication flows, token refresh, and PKCE (Proof Key for Code Exchange) operations. Source: src/utils/oauth/util.py

​
Core Functions

​
run_oauth_flow

Generic OAuth flow handler that can be used by different services. Source: src/utils/oauth/util.py:84 
def run_oauth_flow(
    service_name: str,
    user_id: str,
    scopes: List[str],
    auth_url_base: str,
    token_url: str,
    auth_params_builder: Callable[[Dict[str, Any], str, List[str]], Dict[str, str]],
    token_data_builder: Callable[[Dict[str, Any], str, List[str], str], Dict[str, str]],
    process_token_response: Callable[[Dict[str, Any]], Dict[str, Any]] = None,
    token_header_builder: Callable[[Dict[str, Any]], Dict[str, str]] = None,
    port: int = 8080,
) -> Dict[str, Any]:
​
service_name
str
required
Name of the service (e.g., “microsoft”, “slack”)
​
user_id
str
required
ID of the user authenticating
​
scopes
List[str]
required
List of OAuth scopes to request
​
auth_url_base
str
required
Base authorization URL for the service
​
token_url
str
required
Token exchange URL for the service
​
auth_params_builder
Callable
required
Function to build auth URL parameters. Signature: (oauth_config, redirect_uri, scopes) -> Dict[str, str]
​
token_data_builder
Callable
required
Function to build token request data. Signature: (oauth_config, redirect_uri, scopes, auth_code) -> Dict[str, str]
​
process_token_response
Callable
Optional function to process token response. Signature: (token_response) -> Dict[str, Any]
​
token_header_builder
Callable
Optional function to build token request headers. Signature: (oauth_config) -> Dict[str, str]
​
port
int
default:"8080"
Port to use for local callback server
​
return
Dict[str, Any]
Dictionary containing the OAuth credentials including access_token, refresh_token, and expires_at
Behavior:
  1. Gets OAuth config from auth client
  2. Starts local HTTP server for callback
  3. Opens browser for user authorization
  4. Waits for callback (120 second timeout)
  5. Exchanges authorization code for tokens
  6. Saves credentials using auth client

​
refresh_token_if_needed

Checks if token needs refresh and handles the refresh process. Source: src/utils/oauth/util.py:210 
async def refresh_token_if_needed(
    user_id: str,
    service_name: str,
    token_url: str,
    token_data_builder: Callable[[Dict[str, Any], str, Dict[str, Any]], Dict[str, str]],
    process_token_response: Callable[[Dict[str, Any]], Dict[str, Any]] = None,
    token_header_builder: Callable[[Dict[str, Any]], Dict[str, str]] = None,
    api_key: Optional[str] = None,
    return_full_credentials: Optional[bool] = False,
) -> str:
​
user_id
str
required
ID of the user
​
service_name
str
required
Name of the service
​
token_url
str
required
URL for token refresh
​
token_data_builder
Callable
required
Function to build token refresh request data. Signature: (oauth_config, refresh_token, credentials_data) -> Dict[str, str]
​
process_token_response
Callable
Optional function to process token response
​
token_header_builder
Callable
Optional function to build token request headers
​
api_key
Optional[str]
Optional API key for auth client
​
return_full_credentials
bool
default:"False"
If True, returns full credentials dict; if False, returns only access_token
​
return
str | Dict[str, Any]
Access token string if return_full_credentials=False, otherwise full credentials dictionary
Behavior:
  1. Gets credentials from auth client
  2. Checks if token has expired or will expire in 5 minutes
  3. Refreshes token if needed (only in local environment)
  4. Saves updated credentials
  5. Returns access token or full credentials

​
generate_code_verifier

Generates a code verifier for PKCE. Source: src/utils/oauth/util.py:319 
def generate_code_verifier() -> str:
​
return
str
Cryptographically secure random string of 64 characters using a-z, A-Z, 0-9, ”.”, ”-”, and ”_“

​
generate_code_challenge

Generates a code challenge from the code verifier. Source: src/utils/oauth/util.py:331 
def generate_code_challenge(code_verifier: str) -> str:
​
code_verifier
str
required
The code verifier string
​
return
str
Base64 URL-encoded SHA-256 hash of the code verifier with padding removed

​
OAuthCallbackHandler

HTTP request handler for OAuth callback. Source: src/utils/oauth/util.py:22 
class OAuthCallbackHandler(BaseHTTPRequestHandler):
    """HTTP request handler for OAuth callback."""

​
do_GET

Handles GET requests with OAuth callback. Source: src/utils/oauth/util.py:25 
def do_GET(self):
Behavior:
  • Parses query parameters from callback URL
  • Extracts authorization code or error
  • Parses state parameter for code_verifier (PKCE)
  • Stores additional parameters from callback
  • Returns HTML page with success/error message

​
Example Usage

​
Basic OAuth Flow

from src.utils.oauth.util import run_oauth_flow

def build_auth_params(oauth_config, redirect_uri, scopes):
    return {
        "client_id": oauth_config["client_id"],
        "redirect_uri": redirect_uri,
        "scope": " ".join(scopes),
        "response_type": "code",
    }

def build_token_data(oauth_config, redirect_uri, scopes, auth_code):
    return {
        "client_id": oauth_config["client_id"],
        "client_secret": oauth_config["client_secret"],
        "code": auth_code,
        "redirect_uri": redirect_uri,
        "grant_type": "authorization_code",
    }

credentials = run_oauth_flow(
    service_name="myservice",
    user_id="user123",
    scopes=["read", "write"],
    auth_url_base="https://auth.example.com/oauth/authorize",
    token_url="https://auth.example.com/oauth/token",
    auth_params_builder=build_auth_params,
    token_data_builder=build_token_data,
)

​
Token Refresh

from src.utils.oauth.util import refresh_token_if_needed

def build_refresh_data(oauth_config, refresh_token, credentials_data):
    return {
        "client_id": oauth_config["client_id"],
        "client_secret": oauth_config["client_secret"],
        "refresh_token": refresh_token,
        "grant_type": "refresh_token",
    }

access_token = await refresh_token_if_needed(
    user_id="user123",
    service_name="myservice",
    token_url="https://auth.example.com/oauth/token",
    token_data_builder=build_refresh_data,
)

​
PKCE Flow

from src.utils.oauth.util import (
    generate_code_verifier,
    generate_code_challenge,
    run_oauth_flow
)
import json

def build_auth_params_with_pkce(oauth_config, redirect_uri, scopes):
    code_verifier = generate_code_verifier()
    code_challenge = generate_code_challenge(code_verifier)
    
    # Store code_verifier in state
    state = json.dumps({"code_verifier": code_verifier})
    
    return {
        "client_id": oauth_config["client_id"],
        "redirect_uri": redirect_uri,
        "scope": " ".join(scopes),
        "response_type": "code",
        "code_challenge": code_challenge,
        "code_challenge_method": "S256",
        "state": state,
    }

def build_token_data_with_pkce(oauth_config, redirect_uri, scopes, auth_code):
    return {
        "client_id": oauth_config["client_id"],
        "client_secret": oauth_config["client_secret"],
        "code": auth_code,
        "redirect_uri": redirect_uri,
        "grant_type": "authorization_code",
        "code_verifier": oauth_config.get("code_verifier"),
    }

credentials = run_oauth_flow(
    service_name="myservice",
    user_id="user123",
    scopes=["read", "write"],
    auth_url_base="https://auth.example.com/oauth/authorize",
    token_url="https://auth.example.com/oauth/token",
    auth_params_builder=build_auth_params_with_pkce,
    token_data_builder=build_token_data_with_pkce,
)

​
Error Handling

from src.utils.oauth.util import run_oauth_flow

try:
    credentials = run_oauth_flow(
        # ... parameters
    )
except ValueError as e:
    if "timed out" in str(e):
        print("User didn't complete authentication in time")
    elif "failed" in str(e):
        print(f"Authentication error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

​
See Also

Build docs developers (and LLMs) love

Get started for freeTalk to us