Python SDK

The Kinde Python SDK allows developers to quickly and securely integrate a new or an existing Python application into the Kinde platform. The SDK supports both Flask and FastAPI frameworks through a single unified interface.

What you need

A Kinde account (Sign up for free)
Python version 3.9 or up

If you are using a previous version of Python, you may need to refer to the previous v1 SDK.

For new projects, you can find our Starter Kit on GitHub.

Migration from v1 SDK

If you’re migrating from an older version of the SDK or switching between client types, follow these recommendations:

The current SDK (v2+) provides a more flexible architecture with multiple client types:

Old v1 SDK code

from kinde_sdk.kinde_api_client import KindeApiClient

kinde_client = KindeApiClient(**kinde_api_client_params)
login_url = kinde_client.get_login_url()

New SDK code (OAuth client)

from kinde_sdk.auth.oauth import OAuth
import asyncio

oauth = OAuth(framework="flask", app=app)
login_url = asyncio.get_event_loop().run_until_complete(oauth.login())

Choosing the right client for migration

If you’re using Flask/FastAPI: Continue using OAuth client
If you’re building async-only applications: Migrate to AsyncOAuth
If you need flexibility: Consider SmartOAuth for context-aware behavior

Migration steps

Update imports: Change from kinde_sdk.kinde_api_client to the appropriate client type
Update initialization: Use the new client initialization pattern
Update method calls: Most methods are now async - use await or asyncio.run_until_complete()
Update error handling: Use new exception types from kinde_sdk.exceptions
Test thoroughly: Verify all authentication flows work correctly

Client selection recommendations

Scenario	Recommended Client	Reason
Flask application	`OAuth`	Framework integration, session management
FastAPI application	`OAuth` or `AsyncOAuth`	Both work well, choose based on preference
Pure async app	`AsyncOAuth`	Native async support, better performance
Serverless/Lambda	`AsyncOAuth`	No framework dependencies, async-friendly
Library/Utility	`SmartOAuth`	Works in both sync and async contexts
Mixed codebase	`SmartOAuth`	Context-aware behavior

For more detailed migration instructions, see our GitHub migration guide.

Install

Install PIP and then execute the following command:

pip install kinde-python-sdk

Environment variables

The Kinde Python SDK uses environment variables for configuration. Here are all the supported variables:

Required variables

KINDE_CLIENT_ID - Your application’s client ID from Kinde
KINDE_CLIENT_SECRET - Your application’s client secret from Kinde
KINDE_REDIRECT_URI - The callback URL where Kinde will redirect after authentication
KINDE_HOST - Your Kinde domain (e.g., https://yourdomain.kinde.com)
KINDE_ISSUER_URL - Your Kinde issuer URL (typically same as KINDE_HOST)
GRANT_TYPE - The OAuth grant type to use (e.g., AUTHORIZATION_CODE_WITH_PKCE)

Optional variables

KINDE_AUDIENCE - The intended recipient of the access token (for API access)
KINDE_CALLBACK_URL - Alternative name for KINDE_REDIRECT_URI
LOGOUT_REDIRECT_URL - Where users are redirected after logout
SITE_HOST - Your application’s host (default: 127.0.0.1)
SITE_PORT - Your application’s port (default: 5000)
SITE_URL - Your application’s base URL
CODE_VERIFIER - Required for PKCE flow (auto-generated if not provided)

Session management variables (core SDK features):

SECRET_KEY - Used for session management and token security
SESSION_TYPE - Session storage type (e.g., filesystem)
SESSION_PERMANENT - Whether sessions are permanent (default: False)

Application configuration:

TEMPLATES_AUTO_RELOAD - Whether to auto-reload templates (default: True)

Management API variables (only needed if using Management API features):

MGMT_API_CLIENT_ID - Management API client ID
MGMT_API_CLIENT_SECRET - Management API client secret

Example .env file:

KINDE_CLIENT_ID=your_client_id
KINDE_CLIENT_SECRET=your_client_secret
KINDE_REDIRECT_URI=http://localhost:5000/api/auth/kinde_callback
KINDE_HOST=https://yourdomain.kinde.com
KINDE_ISSUER_URL=https://yourdomain.kinde.com
GRANT_TYPE=AUTHORIZATION_CODE_WITH_PKCE
SITE_HOST=localhost
SITE_PORT=5000
SITE_URL=http://localhost:5000
LOGOUT_REDIRECT_URL=http://localhost:8000
SECRET_KEY=your_secret_key
SESSION_TYPE=filesystem
SESSION_PERMANENT=False
TEMPLATES_AUTO_RELOAD=True

Set callback URLs

In Kinde, go to Settings > Applications > [Your app] > View details.
Add your callback URLs in the relevant fields. For example:
- Allowed callback URLs (also known as redirect URIs) - for example, http://localhost:8000/callback
- Allowed logout redirect URLs - for example, http://localhost:8000
Select Save.

Add environments

Kinde comes with a production environment, but you can set up other environments if you want to. Note that each environment needs to be set up independently, so you need to use the Environment subdomain in the code block above for those new environments.

Choose your OAuth client

The Kinde Python SDK provides three client types to suit different application needs:

Client selection guide

Use OAuth client when:

You’re using Flask or FastAPI frameworks
You want framework-specific integrations (automatic route registration, session management)
Your application follows traditional request-response patterns
You need synchronous operations with async support via framework adapters

Use AsyncOAuth client when:

You’re building a native async application (e.g., pure async Python, async web frameworks)
You want direct async/await support without framework abstractions
You’re using async libraries that don’t fit traditional framework patterns
You need maximum performance with async I/O operations

Use SmartOAuth client when:

Your application uses both sync and async contexts
You want context-aware behavior (automatically sync or async based on usage)
You’re building libraries or utilities that need to work in various contexts
You need flexibility to work with both sync and async code

Comparison table

Feature	OAuth	AsyncOAuth	SmartOAuth
Framework support	Flask, FastAPI	Any async framework	Any context
Sync operations	✅	❌	✅ (context-aware)
Async operations	✅ (via adapters)	✅ (native)	✅ (native)
Route registration	✅ Automatic	❌ Manual	❌ Manual
Session management	✅ Built-in	⚠️ Custom	⚠️ Custom
Best for	Framework apps	Pure async apps	Mixed contexts

Configure your app

Using the OAuth client (Framework-based)

The OAuth client is automatically configured based on the framework you’re using. Simply import the OAuth class from the auth module and create an instance:

from kinde_sdk.auth.oauth import OAuth

# For Flask applications
from flask import Flask
app = Flask(__name__)
oauth = OAuth(
    framework="flask",
    app=app  # optional: pass your Flask app instance
)

# For FastAPI applications
from fastapi import FastAPI
app = FastAPI()
oauth = OAuth(
    framework="fastapi",
    app=app  # optional: pass your FastAPI app instance
)

The SDK will automatically detect and configure the appropriate framework implementation based on the framework parameter and app instance you provide.

Using the AsyncOAuth client (Native async)

The AsyncOAuth client is designed for native async applications. It provides direct async/await support without framework abstractions:

from kinde_sdk.auth.async_oauth import AsyncOAuth

# Initialize AsyncOAuth client
oauth = AsyncOAuth()

# All methods are async
login_url = await oauth.login()
register_url = await oauth.register()

Key features:

Native async/await support
No framework dependencies
Direct control over async operations
Ideal for pure async applications and serverless functions

Using the SmartOAuth client (Context-aware)

The SmartOAuth client automatically adapts to sync or async contexts:

from kinde_sdk.auth.smart_oauth import SmartOAuth

# Initialize SmartOAuth client
oauth = SmartOAuth()

# Works in async context
async def async_function():
    login_url = await oauth.login()  # Returns coroutine

# Works in sync context
def sync_function():
    login_url = oauth.login()  # Returns URL directly

Key features:

Context-aware behavior (sync/async)
Works in both sync and async code
Flexible usage patterns
Ideal for libraries and utilities

Standalone usage (Serverless/Lambda)

For serverless functions (AWS Lambda, Google Cloud Functions, Azure Functions), you can use the SDK in standalone mode without framework dependencies.

AWS Lambda example

import json
from kinde_sdk.auth.async_oauth import AsyncOAuth

# Initialize client (can be outside handler for reuse)
oauth = AsyncOAuth()

async def lambda_handler(event, context):
    """AWS Lambda handler for authentication."""
    try:
        # Handle login
        if event.get('path') == '/login':
            login_url = await oauth.login()
            return {
                'statusCode': 302,
                'headers': {'Location': login_url}
            }

        # Handle callback
        if event.get('path') == '/callback':
            code = event.get('queryStringParameters', {}).get('code')
            state = event.get('queryStringParameters', {}).get('state')

            if not code:
                return {
                    'statusCode': 400,
                    'body': json.dumps({'error': 'No code provided'})
                }

            result = await oauth.handle_redirect(code, state)
            return {
                'statusCode': 200,
                'body': json.dumps({'message': 'Authenticated successfully'})
            }

        # Check authentication
        if event.get('path') == '/user':
            if await oauth.is_authenticated(event):
                user_info = await oauth.get_user_info(event)
                return {
                    'statusCode': 200,
                    'body': json.dumps(user_info)
                }
            else:
                login_url = await oauth.login()
                return {
                    'statusCode': 302,
                    'headers': {'Location': login_url}
                }

    except Exception as e:
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

Google Cloud Functions example

from flask import Request
from kinde_sdk.auth.async_oauth import AsyncOAuth
import asyncio

oauth = AsyncOAuth()

def cloud_function_handler(request: Request):
    """Google Cloud Function handler."""
    async def handle_request():
        if request.path == '/login':
            url = await oauth.login()
            return redirect(url, code=302)

        if request.path == '/callback':
            code = request.args.get('code')
            state = request.args.get('state')
            await oauth.handle_redirect(code, state)
            return {'message': 'Authenticated'}, 200

    return asyncio.run(handle_request())

Token storage for serverless

In serverless environments, you’ll need to implement custom token storage:

from kinde_sdk.auth.async_oauth import AsyncOAuth

class ServerlessTokenStorage:
    """Custom token storage for serverless environments."""

    def __init__(self, storage_backend):  # e.g., Redis, DynamoDB, etc.
        self.storage = storage_backend

    async def store_tokens(self, user_id: str, tokens: dict):
        await self.storage.set(f"tokens:{user_id}", tokens, ttl=3600)

    async def get_tokens(self, user_id: str) -> dict:
        return await self.storage.get(f"tokens:{user_id}")

    async def clear_tokens(self, user_id: str):
        await self.storage.delete(f"tokens:{user_id}")

# Usage
storage = ServerlessTokenStorage(redis_client)
oauth = AsyncOAuth(token_storage=storage)

Authentication flow

The authentication flow consists of several steps. Here’s a comprehensive guide to implementing authentication with the Kinde Python SDK.

Step 1: Initialize the client

Choose the appropriate client based on your application type (see Client Selection Guide).

The Kinde client provides methods for easy sign in and sign up. You can add buttons in your HTML as follows:

<div class="navigation">
  <a href="{{ url_for('login') }}" type="button">Sign in</a>
  <a href="{{ url_for('register') }}" type="button">Sign up</a>
</div>

Automatic Route Registration

The framework wrapper can automatically register all necessary routes. For Flask:

from kinde_sdk.auth.oauth import OAuth
from flask import Flask

app = Flask(__name__)
oauth = OAuth(
    framework="flask",
    app=app
)

For FastAPI:

from kinde_sdk.auth.oauth import OAuth
from fastapi import FastAPI

app = FastAPI()
oauth = OAuth(
    framework="fastapi",
    app=app
)

Framework-specific implementation guides

Flask implementation

Flask applications use the OAuth client with framework support, or AsyncOAuth for async patterns.

Using OAuth client (recommended for Flask):

from flask import Flask, request, session, redirect, url_for
from kinde_sdk.auth.oauth import OAuth
import asyncio

app = Flask(__name__)
app.secret_key = 'your-secret-key'

oauth = OAuth(
    framework="flask",
    app=app
)

@app.route('/login')
def login():
    """Redirect to Kinde login page."""
    loop = asyncio.get_event_loop()
    login_url = loop.run_until_complete(oauth.login())
    return redirect(login_url)

@app.route('/register')
def register():
    """Redirect to Kinde registration page."""
    loop = asyncio.get_event_loop()
    register_url = loop.run_until_complete(oauth.register())
    return redirect(register_url)

@app.route('/callback')
def callback():
    """Handle the OAuth callback from Kinde."""
    try:
        code = request.args.get('code')
        state = request.args.get('state')

        if not code:
            return "Authentication failed: No code provided", 400

        user_id = session.get('user_id') or str(uuid.uuid4())
        loop = asyncio.get_event_loop()
        result = loop.run_until_complete(oauth.handle_redirect(code, user_id, state))
        session['user_id'] = user_id
        return redirect(url_for('index'))
    except Exception as e:
        return f"Authentication failed: {str(e)}", 400

@app.route('/logout')
def logout():
    """Logout the user and redirect to Kinde logout page."""
    user_id = session.get('user_id')
    session.clear()
    loop = asyncio.get_event_loop()
    logout_url = loop.run_until_complete(oauth.logout(user_id))
    return redirect(logout_url)

@app.route('/user')
def get_user():
    """Get the current user's information."""
    try:
        if not oauth.is_authenticated(request):
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
            try:
                login_url = loop.run_until_complete(oauth.login())
                return redirect(login_url)
            finally:
                loop.close()
        return oauth.get_user_info(request)
    except Exception as e:
        return f"Failed to get user info: {str(e)}", 400

FastAPI implementation

FastAPI natively supports async/await, making it ideal for async operations.

Using OAuth client with FastAPI:

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import RedirectResponse
from kinde_sdk.auth.oauth import OAuth

app = FastAPI()

oauth = OAuth(
    framework="fastapi",
    app=app
)

@app.get("/login")
async def login(request: Request):
    """Redirect to Kinde login page."""
    url = await oauth.login()
    return RedirectResponse(url=url)

@app.get("/register")
async def register(request: Request):
    """Redirect to Kinde registration page."""
    url = await oauth.register()
    return RedirectResponse(url=url)

@app.get("/callback")
async def callback(request: Request, code: str, state: str = None):
    """Handle the OAuth callback from Kinde."""
    try:
        result = await oauth.handle_redirect(code, state)
        return RedirectResponse(url="/")
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"Authentication failed: {str(e)}")

@app.get("/logout")
async def logout(request: Request):
    """Logout the user and redirect to Kinde logout page."""
    await oauth.logout()
    return RedirectResponse(url="/")

@app.get("/user")
async def get_user(request: Request):
    """Get the current user's information."""
    if not await oauth.is_authenticated(request):
        url = await oauth.login()
        return RedirectResponse(url=url)
    return await oauth.get_user_info(request)

Using AsyncOAuth client with FastAPI (for more control):

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import RedirectResponse
from kinde_sdk.auth.async_oauth import AsyncOAuth

app = FastAPI()
oauth = AsyncOAuth()

@app.get("/login")
async def login(request: Request):
    url = await oauth.login()
    return RedirectResponse(url=url)

@app.get("/callback")
async def callback(request: Request, code: str, state: str = None):
    try:
        result = await oauth.handle_redirect(code, state)
        return RedirectResponse(url="/")
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

Manual route implementation

If you prefer to implement the routes manually, here’s how you can do it:

For Flask:

import asyncio
from flask import Flask, request, session, redirect
from kinde_sdk.auth.oauth import OAuth

app = Flask(__name__)
oauth = OAuth(
    framework="flask",
    app=app
)

@app.route('/login')
def login():
    """Redirect to Kinde login page.""
    loop = asyncio.get_event_loop()
    login_url = loop.run_until_complete(oauth.login())
    return redirect(login_url)

@app.route('/register')
def register():
    """Redirect to Kinde registration page.""
    loop = asyncio.get_event_loop()
    register_url = loop.run_until_complete(oauth.register())
    return redirect(register_url)

@app.route('/callback')
def callback():
    """Handle the OAuth callback from Kinde.""
    try:
        code = request.args.get('code')
        state = request.args.get('state')

        if not code:
            return "Authentication failed: No code provided", 400

        # Generate a unique user ID for the session
        user_id = session.get('user_id') or str(uuid.uuid4())

        # Use OAuth's handle_redirect method to process the callback
        loop = asyncio.get_event_loop()
        result = loop.run_until_complete(oauth.handle_redirect(code, user_id, state))

        # Store user ID in session
        session['user_id'] = user_id

        return redirect('/')
    except Exception as e:
        return f"Authentication failed: {str(e)}", 400

@app.route('/logout')
def logout():
    """Logout the user and redirect to Kinde logout page.""
    user_id = session.get('user_id')
    session.clear()
    loop = asyncio.get_event_loop()
    logout_url = loop.run_until_complete(oauth.logout(user_id))
    return redirect(logout_url)

@app.route('/user')
def get_user():
    """Get the current user's information.""
    try:
        if not oauth.is_authenticated(request):
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
            try:
                login_url = loop.run_until_complete(oauth.login())
                return redirect(login_url)
            finally:
                loop.close()

        return oauth.get_user_info(request)
    except Exception as e:
        return f"Failed to get user info: {str(e)}", 400

For FastAPI:

from fastapi import FastAPI, Request
from fastapi.responses import RedirectResponse

@app.get("/login")
async def login(request: Request):
    url = await oauth.login()
    return RedirectResponse(url=url)

@app.get("/register")
async def register(request: Request):
    url = await oauth.register()
    return RedirectResponse(url=url)

@app.get("/callback")
async def callback(request: Request, code: str, state: Optional[str] = None):
    try:
        result = await oauth.handle_redirect(code, state)
        return RedirectResponse(url="/")
    except Exception as e:
        return HTMLResponse(f"Authentication failed: {str(e)}")

@app.get("/logout")
async def logout(request: Request):
    request.session.clear()
    return RedirectResponse(url=await oauth.logout())

@app.get("/user")
async def get_user(request: Request):
    if not oauth.is_authenticated(request):
        return RedirectResponse(url=await oauth.login())
    return oauth.get_user_info(request)

The manual implementation gives you more control over the authentication flow and allows you to add custom logic like session management, error handling, and logging. Note that Flask requires special handling of async methods using asyncio since it doesn’t natively support async/await like FastAPI does.

Authentication flow steps

User initiates login/registration: User clicks login/register button
Redirect to Kinde: Application redirects user to Kinde’s authorization server
User authenticates: User signs in/up on Kinde’s hosted pages
Callback handling: Kinde redirects back to your callback URL with authorization code
Token exchange: SDK exchanges authorization code for access tokens
Session creation: SDK stores tokens and creates user session
Protected resources: User can now access protected routes and resources

User permissions

The Kinde Python SDK provides a simple way to check user permissions in your application. The API supports both sync and async patterns depending on your client type.

With OAuth client (Framework-based)

from kinde_sdk.auth.oauth import OAuth
from flask import request
import asyncio

oauth = OAuth(framework="flask", app=app)

# Async pattern (required for OAuth client)
def check_permission_sync():
    loop = asyncio.get_event_loop()
    permission = loop.run_until_complete(
        oauth.get_permission("create:todos", request)
    )
    return permission["isGranted"]

# In FastAPI (native async)
@app.get("/todos")
async def create_todo(request: Request):
    permission = await oauth.get_permission("create:todos", request)
    if not permission["isGranted"]:
        raise HTTPException(status_code=403, detail="Permission denied")
    # Create todo logic...

With AsyncOAuth client (Native async)

from kinde_sdk.auth.async_oauth import AsyncOAuth

oauth = AsyncOAuth()

# Native async pattern
async def check_permission():
    permission = await oauth.get_permission("create:todos")
    if permission["isGranted"]:
        print(f"User has permission in organization: {permission['orgCode']}")
        return True
    return False

# Get all permissions
async def get_all_permissions():
    all_permissions = await oauth.get_permissions()
    print(f"User belongs to organization: {all_permissions['orgCode']}")
    print("User permissions:", all_permissions["permissions"])
    return all_permissions

With SmartOAuth client (Context-aware)

from kinde_sdk.auth.smart_oauth import SmartOAuth

oauth = SmartOAuth()

# Works in async context
async def async_check():
    permission = await oauth.get_permission("create:todos")
    return permission["isGranted"]

# Works in sync context (if available)
def sync_check():
    permission = oauth.get_permission("create:todos")
    return permission["isGranted"]

Checking permissions

Practical examples

Here’s how to use permissions in your application with different client types:

Example 1: Permission check in FastAPI (OAuth client)

from fastapi import FastAPI, Request, HTTPException
from kinde_sdk.auth.oauth import OAuth

app = FastAPI()
oauth = OAuth(framework="fastapi", app=app)

@app.post("/todos")
async def create_todo(request: Request, todo_data: dict):
    permission = await oauth.get_permission("create:todos", request)
    if not permission["isGranted"]:
        raise HTTPException(status_code=403, detail="Permission denied")
    # Create todo logic here...
    return {"message": "Todo created"}

@app.get("/todos")
async def list_todos(request: Request):
    permission = await oauth.get_permission("read:todos", request)
    if not permission["isGranted"]:
        raise HTTPException(status_code=403, detail="Permission denied")
    # List todos logic...
    return {"todos": []}

Example 2: Permission check with AsyncOAuth client

from kinde_sdk.auth.async_oauth import AsyncOAuth

oauth = AsyncOAuth()

async def create_todo_handler(request):
    permission = await oauth.get_permission("create:todos")
    if not permission["isGranted"]:
        return {"error": "Permission denied"}, 403

    org_code = permission.get("orgCode")
    # Create todo with organization context
    return {"message": "Todo created", "org_code": org_code}

async def get_all_user_permissions():
    all_permissions = await oauth.get_permissions()
    return {
        "org_code": all_permissions["orgCode"],
        "permissions": all_permissions["permissions"]
    }

Example 3: Permission-based conditional rendering (Flask)

from flask import Flask, request, render_template
from kinde_sdk.auth.oauth import OAuth
import asyncio

app = Flask(__name__)
oauth = OAuth(framework="flask", app=app)

@app.route("/dashboard")
def dashboard():
    loop = asyncio.get_event_loop()
    can_create = loop.run_until_complete(
        oauth.get_permission("create:todos", request)
    )["isGranted"]

    can_delete = loop.run_until_complete(
        oauth.get_permission("delete:todos", request)
    )["isGranted"]

    return render_template("dashboard.html",
                         can_create=can_create,
                         can_delete=can_delete)

Common permission patterns

Here are some common permission patterns you might use:

# Resource-based permissions
"create:todos
"read:todos
"update:todos
"delete:todos

# Feature-based permissions
"can:export_data
"can:manage_users
"can:view_analytics

# Organization-based permissions
"org:manage_members
"org:view_billing
"org:update_settings

For more information about setting up permissions in Kinde, see User permissions.

Feature flags

The Kinde Python SDK provides a simple way to access feature flags from your application. Feature flags support both sync and async patterns.

With OAuth client (Framework-based)

from kinde_sdk.auth.oauth import OAuth
from flask import request
import asyncio

oauth = OAuth(framework="flask", app=app)

# Async pattern (required for OAuth client)
def get_theme_sync():
    loop = asyncio.get_event_loop()
    theme_flag = loop.run_until_complete(
        oauth.get_flag("theme", request, default_value="light")
    )
    return theme_flag.value

# In FastAPI (native async)
@app.get("/settings")
async def get_settings(request: Request):
    theme = await oauth.get_flag("theme", request, default_value="light")
    dark_mode = await oauth.get_flag("is_dark_mode", request, default_value=False)
    return {
        "theme": theme.value,
        "dark_mode": dark_mode.value
    }

With AsyncOAuth client (Native async)

from kinde_sdk.auth.async_oauth import AsyncOAuth

oauth = AsyncOAuth()

# Native async pattern
async def get_feature_flags():
    # Get a string feature flag
    theme_flag = await oauth.get_flag("theme", default_value="light")
    print(f"Current theme: {theme_flag.value}")

    # Get a boolean feature flag with default value
    dark_mode = await oauth.get_flag("is_dark_mode", default_value=False)
    if dark_mode.value:
        print("Dark mode is enabled")

    # Get a numeric feature flag
    competitions_limit = await oauth.get_flag("competitions_limit", default_value=3)
    print(f"User can create up to {competitions_limit.value} competitions")

    return {
        "theme": theme_flag.value,
        "dark_mode": dark_mode.value,
        "limit": competitions_limit.value
    }

Getting feature flags

To get a specific feature flag value:

Practical examples

Here’s how to use feature flags in your application with different client types:

Example 1: Conditional feature rendering (FastAPI with OAuth)

from fastapi import FastAPI, Request
from kinde_sdk.auth.oauth import OAuth

app = FastAPI()
oauth = OAuth(framework="fastapi", app=app)

@app.get("/competitions/create-button")
async def render_create_button(request: Request):
    can_create = await oauth.get_flag("create_competition", request, default_value=False)
    if can_create.value:
        return {"html": "<button>Create Competition</button>"}
    return {"html": ""}

Example 2: Theme configuration (AsyncOAuth)

from kinde_sdk.auth.async_oauth import AsyncOAuth

oauth = AsyncOAuth()

async def get_user_theme():
    theme = await oauth.get_flag("theme", default_value="light")
    dark_mode = await oauth.get_flag("is_dark_mode", default_value=False)
    return {
        "theme": theme.value,
        "is_dark_mode": dark_mode.value
    }

Example 3: Feature limits with validation (FastAPI)

from fastapi import FastAPI, Request, HTTPException
from kinde_sdk.auth.oauth import OAuth

app = FastAPI()
oauth = OAuth(framework="fastapi", app=app)

@app.post("/competitions")
async def create_competition(request: Request, competition_data: dict):
    limit_flag = await oauth.get_flag("competitions_limit", request, default_value=3)
    current_count = await get_user_competition_count(request)

    if current_count >= limit_flag.value:
        raise HTTPException(
            status_code=403,
            detail=f"Competition limit reached (max: {limit_flag.value})"
        )
    # Create competition logic here...
    return {"message": "Competition created"}

Example 4: Type-safe flag access (AsyncOAuth)

from kinde_sdk.auth.async_oauth import AsyncOAuth

oauth = AsyncOAuth()

async def get_all_flags():
    # Get all feature flags
    all_flags = await oauth.get_all_flags()
    result = {}
    for code, flag in all_flags.items():
        result[code] = {
            "value": flag.value,
            "type": flag.type,
            "is_default": flag.is_default
        }
    return result

# Type-specific getters
async def get_boolean_flag(flag_name: str, default: bool = False):
    return await oauth.get_boolean_flag(flag_name, default_value=default)

async def get_string_flag(flag_name: str, default: str = ""):
    return await oauth.get_string_flag(flag_name, default_value=default)

async def get_integer_flag(flag_name: str, default: int = 0):
    return await oauth.get_integer_flag(flag_name, default_value=default)

Feature flag types

The SDK supports the following feature flag types:

# String flags
{
    "t": "s",
    "v": "pink"
}

# Boolean flags
{
    "t": "b",
    "v": true
}

# Integer flags
{
    "t": "i",
    "v": 5
}

Common use cases

# Feature Toggles
can_use_feature = await feature_flags.get_flag("enable_new_feature", default_value=False)

# User Preferences
theme = await feature_flags.get_flag("theme", default_value="light")
dark_mode = await feature_flags.get_flag("is_dark_mode", default_value=False)

# Usage Limits
max_uploads = await feature_flags.get_flag("max_uploads", default_value=10)

# A/B Testing
test_group = await feature_flags.get_flag("ab_test_group", default_value="control")

Claims

The Kinde Python SDK provides a simple way to access user claims from your application. Claims support both sync and async patterns.

With OAuth client (Framework-based)

from kinde_sdk.auth.oauth import OAuth
from flask import request
import asyncio

oauth = OAuth(framework="flask", app=app)

# Async pattern (required for OAuth client)
def get_user_name_sync():
    loop = asyncio.get_event_loop()
    claim = loop.run_until_complete(
        oauth.get_claim("given_name", request, token_type="id_token")
    )
    return claim["value"]

# In FastAPI (native async)
@app.get("/profile")
async def get_profile(request: Request):
    given_name = await oauth.get_claim("given_name", request, token_type="id_token")
    family_name = await oauth.get_claim("family_name", request, token_type="id_token")
    email = await oauth.get_claim("email", request, token_type="id_token")
    return {
        "name": f"{given_name['value']} {family_name['value']}",
        "email": email["value"]
    }

With AsyncOAuth client (Native async)

from kinde_sdk.auth.async_oauth import AsyncOAuth

oauth = AsyncOAuth()

# Native async pattern
async def get_claims():
    # Get the audience claim from the access token
    aud_claim = await oauth.get_claim("aud")
    print(f"Token audience: {aud_claim['value']}")

    # Get the given_name claim from the ID token
    name_claim = await oauth.get_claim("given_name", token_type="id_token")
    print(f"User's given name: {name_claim['value']}")

    return {
        "audience": aud_claim["value"],
        "name": name_claim["value"]
    }

Getting claims

To get a specific claim from the user’s tokens:

Practical examples

Here’s how to use claims in your application with different client types:

Example 1: Accessing user information (FastAPI with OAuth)

from fastapi import FastAPI, Request
from kinde_sdk.auth.oauth import OAuth

app = FastAPI()
oauth = OAuth(framework="fastapi", app=app)

@app.get("/profile")
async def get_user_profile(request: Request):
    given_name = await oauth.get_claim("given_name", request, token_type="id_token")
    family_name = await oauth.get_claim("family_name", request, token_type="id_token")
    email = await oauth.get_claim("email", request, token_type="id_token")

    if given_name["value"] and family_name["value"]:
        return {
            "name": f"{given_name['value']} {family_name['value']}",
            "email": email["value"]
        }
    return None

Example 2: Token validation (AsyncOAuth)

from fastapi import HTTPException
from kinde_sdk.auth.async_oauth import AsyncOAuth

oauth = AsyncOAuth()

async def validate_token():
    aud_claim = await oauth.get_claim("aud")
    if not aud_claim["value"] or "api.yourapp.com" not in aud_claim["value"]:
        raise HTTPException(status_code=401, detail="Invalid token audience")
    return {"message": "Access granted"}

@app.get("/api/protected")
async def protected_endpoint():
    return await validate_token()

Example 3: Getting all claims (AsyncOAuth)

from kinde_sdk.auth.async_oauth import AsyncOAuth

oauth = AsyncOAuth()

async def get_all_user_claims():
    # Get all claims from the access token
    all_claims = await oauth.get_all_claims()
    access_token_data = {}
    for claim_name, claim_value in all_claims.items():
        access_token_data[claim_name] = claim_value

    # Get all claims from the ID token
    id_token_claims = await oauth.get_all_claims(token_type="id_token")
    id_token_data = {}
    for claim_name, claim_value in id_token_claims.items():
        id_token_data[claim_name] = claim_value

    return {
        "access_token": access_token_data,
        "id_token": id_token_data
    }

Example 4: Organization context (FastAPI)

from fastapi import FastAPI, Request
from kinde_sdk.auth.oauth import OAuth

app = FastAPI()
oauth = OAuth(framework="fastapi", app=app)

@app.get("/organization-info")
async def get_org_info(request: Request):
    org_code = await oauth.get_claim("org_code", request)
    org_name = await oauth.get_claim("org_name", request)

    return {
        "org_code": org_code["value"],
        "org_name": org_name["value"] if org_name else None
    }

Common claims

Here are some common claims you might want to access:

# User Information (ID Token)
"given_name
"family_name
"email
"picture

# Token Information (Access Token)
"aud"           # Audience
"iss"           # Issuer
"exp"           # Expiration time
"iat"           # Issued at time

# Organization Information
"org_code
"org_name
"org_id

Organizations

Create an organization

To create a new organization within your application, you will need to run a similar function to below:

return app.redirect(oauth.create_org())

Kinde has a unique code for every organization. You’ll have to pass this code through when you register a new user or sign in to a particular organization. Example function below:

oauth.get_claim("org_code")

@app.route("/login")
def login():
    return app.redirect(oauth.get_login_url())


@app.route("/register")
def register():
    return app.redirect(oauth.get_register_url())

Following authentication, Kinde provides a json web token (jwt) to your application. Along with the standard information we also include the org_code and the permissions for that organization (this is important as a user can belong to multiple organizations and have different permissions for each).

Example of a returned token:

{
   "aud": [],
   "exp": 1658475930,
   "iat": 1658472329,
   "iss": "https://your_subdomain.kinde.com",
   "jti": "123457890",
   "org_code": "org_1234",
   "permissions": ["read:todos", "create:todos"],
   "scp": [
       "openid",
       "profile",
       "email",
       "offline
   ],
   "sub": "kp:123457890"
}

The id_token will also contain an array of organizations that a user belongs to - this is useful if you wanted to build out an organization switcher for example.

{
    ...
    "org_codes": ["org_1234", "org_4567"],
    ...
};

There are two helper functions you can use to extract information:

oauth.get_organization()

oauth.get_user_organizations()

For more information about how organizations work in Kinde, see Kinde organizations for developers.

Token and session management

The Kinde Python SDK automatically handles token and session management for your application. Once a user has successfully authenticated, the SDK manages:

Token acquisition and storage: Automatically obtains and securely stores access tokens, ID tokens, and refresh tokens
Token refresh: Automatically refreshes tokens when they expire
Session management: Handles user sessions across requests
Framework integration: Works seamlessly with Flask and FastAPI session systems

The SDK uses the session configuration from your environment variables (SECRET_KEY, SESSION_TYPE, SESSION_PERMANENT) to manage sessions appropriately for your chosen framework.

Token types

The SDK supports two types of tokens:

Access Token (token_type="access_token"):
- Contains authorization information
- Used for API access
- Contains permissions and organization context
- Default token type
ID Token (token_type="id_token"):
- Contains user identity information
- Used for user profile data
- Contains name, email, and other user details
- Must be explicitly requested using token_type="id_token"

Session handling

The SDK automatically integrates with your framework’s session system:

Flask: Uses Flask’s built-in session management
FastAPI: Integrates with FastAPI’s session handling

You don’t need to manually manage tokens or sessions - the SDK handles this automatically for you.

Management API

The Kinde Python SDK provides a Management API client for interacting with Kinde’s management endpoints. This allows you to programmatically manage users, organizations, and other resources. The Management API supports both sync and async patterns.

Getting started

With OAuth client (Framework-based)

from kinde_sdk.auth.oauth import OAuth
from flask import Flask
import asyncio

app = Flask(__name__)
oauth = OAuth(framework="flask", app=app)

# Get the management client
management = oauth.get_management()

# Use with asyncio in Flask
def list_users_sync():
    loop = asyncio.get_event_loop()
    users = loop.run_until_complete(management.get_users())
    return users

With AsyncOAuth client (Native async)

from kinde_sdk.auth.async_oauth import AsyncOAuth

oauth = AsyncOAuth()

# Get the management client (native async)
management = await oauth.get_management()

# All methods are async
users = await management.get_users()

With SmartOAuth client (Context-aware)

from kinde_sdk.auth.smart_oauth import SmartOAuth

oauth = SmartOAuth()

# Works in async context
async def async_get_users():
    management = await oauth.get_management()
    return await management.get_users()

# Works in sync context (if supported)
def sync_get_users():
    management = oauth.get_management()
    return management.get_users()

Available endpoints

The Management API provides methods for common operations on resources. All examples use async patterns:

User management:

# List users (async)
users = await management.get_users()

# Get a specific user (async)
user = await management.get_user(user_id="user_123")

# Create a new user (async)
new_user = await management.create_user(
    email="user@example.com",
    given_name="John",
    family_name="Doe"
)

# Update a user (async)
updated_user = await management.update_user(
    user_id="user_123",
    given_name="Johnny"
)

# Delete a user (async)
await management.delete_user(user_id="user_123")

Organization management

Using Management API with FastAPI (OAuth client):

from fastapi import FastAPI, HTTPException
from kinde_sdk.auth.oauth import OAuth

app = FastAPI()
oauth = OAuth(framework="fastapi", app=app)

@app.get("/organizations")
async def list_organizations():
    management = oauth.get_management()
    orgs = await management.get_organizations()
    return orgs

@app.get("/organizations/{org_id}")
async def get_organization(org_id: str):
    management = oauth.get_management()
    org = await management.get_organization(org_id=org_id)
    return org

@app.post("/organizations")
async def create_organization(name: str):
    management = oauth.get_management()
    new_org = await management.create_organization(name=name)
    return new_org

@app.put("/organizations/{org_id}")
async def update_organization(org_id: str, name: str):
    management = oauth.get_management()
    updated_org = await management.update_organization(
        org_id=org_id,
        name=name
    )
    return updated_org

@app.delete("/organizations/{org_id}")
async def delete_organization(org_id: str):
    management = oauth.get_management()
    await management.delete_organization(org_id=org_id)
    return {"message": "Organization deleted"}

Using Management API with AsyncOAuth client:

from kinde_sdk.auth.async_oauth import AsyncOAuth

oauth = AsyncOAuth()

async def manage_organizations():
    # Get management client
    management = await oauth.get_management()

    # List organizations
    orgs = await management.get_organizations()

    # Get a specific organization
    org = await management.get_organization(org_id="org_123")

    # Create a new organization
    new_org = await management.create_organization(
        name="My Organization"
    )

    # Update an organization
    updated_org = await management.update_organization(
        org_id="org_123",
        name="Updated Name"
    )

    # Delete an organization
    await management.delete_organization(org_id="org_123")

    return orgs

Error handling

The Management API methods will raise exceptions for API errors. It’s recommended to handle these appropriately:

Example with OAuth client (FastAPI):

from fastapi import FastAPI, HTTPException
from kinde_sdk.auth.oauth import OAuth
from kinde_sdk.exceptions import KindeAPIException

app = FastAPI()
oauth = OAuth(framework="fastapi", app=app)

@app.get("/users/{user_id}")
async def get_user(user_id: str):
    management = oauth.get_management()
    try:
        user = await management.get_user(user_id=user_id)
        return user
    except KindeAPIException as e:
        raise HTTPException(status_code=e.status_code, detail=str(e))
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Internal error: {str(e)}")

Example with AsyncOAuth client:

from kinde_sdk.auth.async_oauth import AsyncOAuth
from kinde_sdk.exceptions import KindeAPIException

oauth = AsyncOAuth()

async def get_user_safely(user_id: str):
    management = await oauth.get_management()
    try:
        user = await management.get_user(user_id=user_id)
        return user
    except KindeAPIException as e:
        print(f"API Error {e.status_code}: {e.message}")
        return None
    except Exception as e:
        print(f"Unexpected error: {str(e)}")
        return None

Token management

The Management API client has its own token management system for API authentication, which is separate from the core SDK’s user session token management. The Management API client automatically handles:

accessing Kinde Management API endpoints: Obtains tokens for accessing Kinde’s management endpoints
Token refresh: Automatically refreshes management API tokens when they expire
Token storage: Securely stores management API tokens
Thread safety: Ensures thread-safe token handling for concurrent requests

You don’t need to manually manage Management API tokens - the client handles this for you. This is different from the core SDK’s user session token management, which handles user authentication tokens automatically.

Best practices

Always use async/await when calling Management API methods: The Management API is async-native for better performance
Handle API errors appropriately: Use try/except blocks and handle KindeAPIException specifically
Cache results when appropriate: Reduce API calls by caching user data, organizations, and permissions
Use appropriate error handling for production: Implement logging, monitoring, and graceful error recovery
Keep your client credentials secure: Use environment variables, never commit secrets to version control
Use connection pooling: For high-traffic applications, configure HTTP connection pooling
Implement retry logic: Add retry logic with exponential backoff for transient failures
Monitor token expiration: Handle token refresh gracefully to avoid authentication failures

For more information about the Management API endpoints and capabilities, see the Kinde Management API documentation.

Error handling and best practices

Proper error handling is crucial for building robust applications with the Kinde Python SDK. This section covers common error scenarios and best practices.

Common exceptions

The SDK raises specific exception types that you should handle:

from kinde_sdk.exceptions import (
    KindeAPIException,
    KindeAuthenticationException,
    KindeAuthorizationException,
    KindeValidationException,
    KindeConfigurationException
)

Error handling patterns

Pattern 1: Comprehensive error handling (FastAPI)

from fastapi import FastAPI, HTTPException
from kinde_sdk.auth.oauth import OAuth
from kinde_sdk.exceptions import (
    KindeAPIException,
    KindeAuthenticationException,
    KindeAuthorizationException
)

app = FastAPI()
oauth = OAuth(framework="fastapi", app=app)

@app.get("/protected")
async def protected_route(request: Request):
    try:
        # Check authentication
        if not await oauth.is_authenticated(request):
            raise HTTPException(status_code=401, detail="Not authenticated")

        # Get user info
        user_info = await oauth.get_user_info(request)
        return user_info

    except KindeAuthenticationException as e:
        # Handle authentication errors
        raise HTTPException(status_code=401, detail=f"Authentication failed: {str(e)}")

    except KindeAuthorizationException as e:
        # Handle authorization errors
        raise HTTPException(status_code=403, detail=f"Authorization failed: {str(e)}")

    except KindeAPIException as e:
        # Handle API errors
        raise HTTPException(status_code=e.status_code, detail=f"API error: {str(e)}")

    except Exception as e:
        # Handle unexpected errors
        raise HTTPException(status_code=500, detail=f"Internal error: {str(e)}")

Pattern 2: Error handling with AsyncOAuth

from kinde_sdk.auth.async_oauth import AsyncOAuth
from kinde_sdk.exceptions import KindeAPIException
import logging

oauth = AsyncOAuth()
logger = logging.getLogger(__name__)

async def safe_get_user():
    try:
        user_info = await oauth.get_user_info()
        return user_info
    except KindeAPIException as e:
        logger.error(f"Kinde API error: {e.status_code} - {e.message}")
        return None
    except Exception as e:
        logger.exception(f"Unexpected error: {str(e)}")
        return None

Pattern 3: Permission checking with error handling

from fastapi import HTTPException
from kinde_sdk.auth.oauth import OAuth
from kinde_sdk.exceptions import KindeAPIException

oauth = OAuth(framework="fastapi", app=app)

async def check_permission_with_error_handling(request: Request, permission: str):
    try:
        perm_result = await oauth.get_permission(permission, request)
        if not perm_result.get("isGranted"):
            raise HTTPException(
                status_code=403,
                detail=f"Permission '{permission}' not granted"
            )
        return perm_result
    except KindeAPIException as e:
        raise HTTPException(
            status_code=e.status_code,
            detail=f"Error checking permission: {str(e)}"
        )

Best practices summary

Authentication and authorization

Always validate authentication: Check is_authenticated() before accessing protected resources
Handle token expiration: Implement token refresh logic or redirect to login
Validate permissions: Check permissions before allowing operations
Use appropriate HTTP status codes: 401 for authentication, 403 for authorization failures

Configuration

Use environment variables: Never hardcode credentials
Validate configuration: Check required environment variables at startup
Use secure storage: Store tokens securely (encrypted sessions, secure cookies)
Set appropriate session timeouts: Balance security and user experience

Performance

Cache when appropriate: Cache user info, permissions, and feature flags
Use async operations: Prefer async/await for better performance
Implement connection pooling: For high-traffic applications
Batch operations: When possible, batch Management API calls

Security

Protect client secrets: Use environment variables or secret management services
Use HTTPS: Always use HTTPS in production
Validate redirect URLs: Ensure callback URLs are whitelisted
Implement CSRF protection: Use state parameters in OAuth flows
Log security events: Log authentication failures and authorization denials

Error handling

Handle specific exceptions: Catch specific exception types, not just Exception
Log errors appropriately: Log errors with context but don’t expose sensitive information
Provide user-friendly messages: Don’t expose internal error details to users
Implement retry logic: For transient failures, implement exponential backoff
Monitor error rates: Track error rates and patterns in production

Development

Use type hints: Improve code clarity and IDE support
Write tests: Test authentication flows, error cases, and edge cases
Use linting: Catch errors early with tools like mypy and ruff
Document error handling: Document how your application handles errors

Example: Complete error handling setup

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from kinde_sdk.auth.oauth import OAuth
from kinde_sdk.exceptions import (
    KindeAPIException,
    KindeAuthenticationException,
    KindeAuthorizationException,
    KindeValidationException
)
import logging

app = FastAPI()
oauth = OAuth(framework="fastapi", app=app)
logger = logging.getLogger(__name__)

# Global exception handler
@app.exception_handler(KindeAuthenticationException)
async def auth_exception_handler(request: Request, exc: KindeAuthenticationException):
    logger.warning(f"Authentication failed: {str(exc)}")
    return JSONResponse(
        status_code=401,
        content={"error": "Authentication required", "detail": "Please log in"}
    )

@app.exception_handler(KindeAuthorizationException)
async def authz_exception_handler(request: Request, exc: KindeAuthorizationException):
    logger.warning(f"Authorization failed: {str(exc)}")
    return JSONResponse(
        status_code=403,
        content={"error": "Access denied", "detail": "Insufficient permissions"}
    )

@app.exception_handler(KindeAPIException)
async def api_exception_handler(request: Request, exc: KindeAPIException):
    logger.error(f"Kinde API error {exc.status_code}: {str(exc)}")
    return JSONResponse(
        status_code=exc.status_code,
        content={"error": "API error", "detail": "An error occurred with the authentication service"}
    )

@app.get("/api/protected")
async def protected_endpoint(request: Request):
    # This will automatically handle exceptions via the handlers above
    if not await oauth.is_authenticated(request):
        raise KindeAuthenticationException("Not authenticated")

    permission = await oauth.get_permission("read:data", request)
    if not permission.get("isGranted"):
        raise KindeAuthorizationException("Permission denied")

    return {"message": "Access granted"}

Join the community

Get support from the Kinde team and expert users in the Kinde community

Join via Slack or Discord
Contact support

Chat to support through the Kinde help widget, or send a question via email

Send an email

Python SDK

What you need

Migration from v1 SDK

Choosing the right client for migration

Migration steps

Client selection recommendations

Install

Environment variables

Required variables

Optional variables

Set callback URLs

Add environments

Choose your OAuth client

Client selection guide

Comparison table

Configure your app

Using the OAuth client (Framework-based)

Using the AsyncOAuth client (Native async)

Using the SmartOAuth client (Context-aware)

Standalone usage (Serverless/Lambda)

AWS Lambda example

Google Cloud Functions example

Token storage for serverless

Authentication flow

Step 1: Initialize the client

Step 2: Sign in and sign up

Automatic Route Registration

Framework-specific implementation guides

Flask implementation

FastAPI implementation

Manual route implementation

Authentication flow steps

User permissions

With OAuth client (Framework-based)

With AsyncOAuth client (Native async)

With SmartOAuth client (Context-aware)

Checking permissions

Practical examples

Common permission patterns

Feature flags

With OAuth client (Framework-based)

With AsyncOAuth client (Native async)

Getting feature flags

Practical examples

Feature flag types

Common use cases

Claims

With OAuth client (Framework-based)

With AsyncOAuth client (Native async)

Getting claims

Practical examples

Common claims

Organizations

Create an organization

Sign up and sign in to organizations

Token and session management

Token types

Session handling

Management API

Getting started

With OAuth client (Framework-based)

With AsyncOAuth client (Native async)

With SmartOAuth client (Context-aware)

Available endpoints

Organization management

Error handling

Token management

Best practices

Error handling and best practices

Common exceptions

Error handling patterns

Best practices summary

Authentication and authorization

Configuration

Performance

Security

Error handling

Development

Example: Complete error handling setup

Useful links