How the OAuth Flow Works for Third-Party Integrations in Omi

The Omi platform implements a standardized OAuth 2.0 authorization code flow across all third-party integrations, using CSRF state tokens stored in Redis (with JSON fallback) and automatic refresh mechanisms to maintain persistent API access.

The basedhardware/omi repository powers an AI wearable ecosystem that connects with external services like Whoop, Twitter/X, Slack, and Notion. Understanding how the OAuth flow works for third-party integrations reveals a consistent architectural pattern that balances security with developer ergonomics, enabling seamless user authentication across diverse provider APIs.

The Six-Step OAuth 2.0 Authorization Code Flow

Every integration in the Omi codebase follows an identical high-level sequence, implemented in provider-specific FastAPI routers. The flow protects against CSRF attacks while ensuring long-lived access through token refresh capabilities.

1. Generate and store state token – A unique CSRF token is created and mapped to the user ID in Redis or a local JSON file.
2. Redirect to provider – The user is sent to the provider’s /authorize endpoint with client_id, redirect_uri, scope, and the state parameter.
3. Callback verification – Upon return, the state parameter is validated against stored values before proceeding.
4. Exchange code for tokens – The authorization code is traded for access and refresh tokens via POST request to the provider’s token endpoint.
5. Persist credentials – Token data, expiry timestamps, and refresh tokens are stored for subsequent API calls.
6. Automatic refresh – Before each API request, the system checks token validity and refreshes expired credentials silently.

This pattern appears in plugins/omi-whoop-app/main.py, plugins/omi-twitter-chat-tools-app/main.py, and every other integration plugin, differing only in provider-specific URLs and scopes.

Implementation Deep Dive: Whoop Integration Example

The Whoop integration demonstrates the complete OAuth implementation. Located in plugins/omi-whoop-app/main.py, this module handles the full lifecycle from initial redirect to token refresh.

Initiating the Authentication Request

When a user initiates a connection, the system generates an 8-character state token and redirects the browser to Whoop’s authorization endpoint:

# plugins/omi-whoop-app/main.py

@app.get("/auth/whoop")
async def whoop_login(uid: str = Query(...)):
    # Generate short random state string (Whoop requires exactly 8 chars)

    state = secrets.token_urlsafe(8)[:8]
    
    # Store mapping between state and user ID for CSRF protection

    store_oauth_state(state, uid)
    
    # Compose provider authorization URL

    params = {
        "client_id": WHOOP_CLIENT_ID,
        "redirect_uri": WHOOP_REDIRECT_URI,
        "response_type": "code",
        "scope": " ".join(WHOOP_SCOPES),
        "state": state,
    }
    auth_url = f"{WHOOP_AUTH_URL}?{urlencode(params)}"
    
    return RedirectResponse(url=auth_url)

The store_oauth_state function in plugins/omi-whoop-app/db.py handles persistence, writing to Redis with a 10-minute TTL or falling back to a JSON file if Redis is unavailable.

Handling the Callback and Token Exchange

After user consent, Whoop redirects to /auth/whoop/callback with the authorization code and original state. The endpoint validates the state parameter to prevent CSRF attacks before exchanging the code:

# plugins/omi-whoop-app/main.py

@app.get("/auth/whoop/callback")
async def whoop_callback(
    code: str = Query(None),
    state: str = Query(None),
    error: str = Query(None)
):
    # CSRF protection: verify state exists and matches user

    uid = get_uid_from_oauth_state(state)
    if not uid:
        return HTMLResponse("Invalid state parameter", status_code=400)
    
    # One-time use: remove state to prevent replay attacks

    delete_oauth_state(state)
    
    # Exchange authorization code for tokens

    response = requests.post(
        WHOOP_TOKEN_URL,
        data={
            "client_id": WHOOP_CLIENT_ID,
            "client_secret": WHOOP_CLIENT_SECRET,
            "code": code,
            "grant_type": "authorization_code",
            "redirect_uri": WHOOP_REDIRECT_URI,
        },
    )
    
    token_data = response.json()
    expires_at = (datetime.utcnow() + timedelta(
        seconds=token_data.get("expires_in", 3600)
    )).isoformat() + "Z"
    
    # Persist tokens for future API calls

    store_whoop_tokens(
        uid, 
        token_data["access_token"], 
        token_data.get("refresh_token", ""), 
        expires_at
    )
    
    return HTMLResponse("Connected to Whoop!", status_code=200)

The get_uid_from_oauth_state function looks up the state token in Redis or the JSON fallback, returning None if expired or missing, which immediately aborts the authentication attempt.

Automatic Token Refresh

Before making API calls, the integration checks token validity and refreshes automatically if within expiry window:

# plugins/omi-whoop-app/main.py

def get_valid_access_token(uid: str) -> Optional[str]:
    tokens = get_whoop_tokens(uid)
    
    # Check if token needs refresh

    if is_token_expired(tokens["expires_at"]):
        # Request new access token using refresh token

        response = requests.post(
            WHOOP_TOKEN_URL,
            data={
                "client_id": WHOOP_CLIENT_ID,
                "client_secret": WHOOP_CLIENT_SECRET,
                "grant_type": "refresh_token",
                "refresh_token": tokens["refresh_token"],
            },
        )
        
        new_token = response.json()
        new_expires = (datetime.utcnow() + timedelta(
            seconds=new_token["expires_in"]
        )).isoformat() + "Z"
        
        # Update stored credentials

        update_whoop_tokens(
            uid,
            new_token["access_token"],
            new_token.get("refresh_token", tokens["refresh_token"]),
            new_expires
        )
        
        return new_token["access_token"]
    
    return tokens["access_token"]

This pattern ensures API calls never fail due to expired credentials, refreshing tokens transparently before requests to Whoop endpoints.

Security Mechanisms and State Management

The Omi OAuth implementation prioritizes security through several defensive measures implemented consistently across all providers.

CSRF Protection via State Tokens

Each authentication attempt generates a cryptographically random state token using secrets.token_urlsafe(). This value acts as a nonce that binds the authorization request to the callback, preventing cross-site request forgery attacks. The state-to-user mapping persists for 10 minutes in Redis (ex=600) or indefinitely in the JSON fallback, though entries are immediately deleted upon validation.

Dual-Layer Storage Architecture

Token persistence uses a Redis-first approach with automatic JSON file fallback:

# plugins/omi-whoop-app/db.py

def store_oauth_state(state: str, uid: str):
    r = _get_redis()
    if r:
        # Primary: Redis with 10-minute expiration

        r.set(f"whoop:oauth_state:{state}", uid, ex=600)
    else:
        # Fallback: Local JSON file for development/edge cases

        states = _load_json(OAUTH_STATES_FILE)
        states[state] = {
            "uid": uid, 
            "created_at": datetime.utcnow().isoformat()
        }
        _save_json(OAUTH_STATES_FILE, states)

This architecture ensures OAuth flows function in production environments with Redis while maintaining portability for local development.

Provider-Specific Variations

While the core flow remains consistent, certain providers require protocol extensions. The Twitter/X integration in plugins/omi-twitter-chat-tools-app/main.py implements PKCE (Proof Key for Code Exchange) for enhanced security:

# plugins/omi-twitter-chat-tools-app/main.py

def generate_code_verifier() -> str:
    return secrets.token_urlsafe(64)[:128]

def generate_code_challenge(verifier: str) -> str:
    digest = hashlib.sha256(verifier.encode()).digest()
    return base64.urlsafe_b64encode(digest).decode().rstrip("=")

# Usage during authorization

code_verifier = generate_code_verifier()
code_challenge = generate_code_challenge(code_verifier)

# Store verifier for callback verification

store_oauth_state(state, uid, code_verifier=code_verifier)

The code verifier is stored alongside the state parameter and submitted during the token exchange to prevent authorization code interception attacks. This same pattern appears in integrations requiring elevated security postures.

Supported Integration Patterns

The identical OAuth architecture powers connections to:

• Twitter/X – OAuth 2.0 with PKCE
• Slack – Standard authorization code flow
• Google Calendar – Offline access with refresh tokens
• Notion – Scoped workspace authentication
• Shopify – Per-shop OAuth grants
• ShipBob, Dropbox, Linear – Standard OAuth 2.0 implementations

Each plugin contains a main.py defining the FastAPI routes and a db.py handling token storage, maintaining architectural consistency while accommodating provider-specific endpoints and scopes.

Key Files and Module Architecture

File	Purpose	Critical Functions
plugins/omi-whoop-app/main.py	Whoop OAuth endpoints and token refresh	whoop_login, whoop_callback, get_valid_access_token
plugins/omi-whoop-app/db.py	Token and state persistence	store_oauth_state, get_uid_from_oauth_state, store_whoop_tokens
plugins/omi-twitter-chat-tools-app/main.py	Twitter OAuth with PKCE extension	generate_code_verifier, generate_code_challenge
plugins/omi-slack-app/main.py	Slack-specific OAuth implementation	Similar pattern: slack_login, slack_callback
plugins/omi-google-calendar-app/main.py	Google OAuth with offline access	Token refresh for background sync capabilities

These files demonstrate that adding a new third-party integration requires only implementing the two FastAPI endpoints (/auth/<provider> and /auth/<provider>/callback) and the storage helpers in a corresponding db.py.

Summary

• OAuth 2.0 authorization code flow provides the foundation for all third-party integrations in the Omi ecosystem, implemented consistently across every plugin.
• CSRF protection is enforced via cryptographically random state tokens stored in Redis with 10-minute TTL or JSON file fallback.
• Token persistence includes access tokens, refresh tokens, and calculated expiry timestamps, enabling automatic background refresh before API calls.
• Provider variations like Twitter's PKCE implementation extend the base pattern without altering the core architecture.
• Dual-layer storage (Redis primary, JSON fallback) ensures the OAuth flow operates reliably across development and production environments.

Frequently Asked Questions

How does Omi prevent CSRF attacks during the OAuth flow?

Omi generates a unique state token using secrets.token_urlsafe() before redirecting users to the provider. This token is mapped to the user ID and stored in Redis with a 10-minute expiration. When the provider redirects back to the callback endpoint, the system validates that the returned state matches the stored value using get_uid_from_oauth_state() in db.py. If the state is missing, expired, or mismatched, the request is immediately rejected with a 400 status code, preventing cross-site request forgery attacks.

What happens if the access token expires during API usage?

The get_valid_access_token() function in each plugin's main.py automatically checks token expiry before every API call. If the token is near expiration or expired, the system uses the stored refresh token to request new credentials from the provider's token endpoint. The new access token, refresh token (if rotated), and updated expiry timestamp are immediately persisted to storage, ensuring seamless API access without user intervention.

Why does the Whoop integration use an 8-character state token while others use longer tokens?

The Whoop API specifically requires the state parameter to be exactly 8 characters, implemented via secrets.token_urlsafe(8)[:8] in plugins/omi-whoop-app/main.py. Other providers like Twitter or Google accept longer state values (typically 32+ characters). The codebase adapts to provider-specific constraints while maintaining the same CSRF protection semantics across all integrations.

Can the OAuth implementation work without Redis?

Yes, the system includes a JSON file fallback for environments without Redis connectivity. The store_oauth_state() and related functions in each db.py module check for Redis availability first; if unavailable, they read from and write to a local JSON file (oauth_states.json or similar). This dual-layer approach ensures developers can test OAuth flows locally without infrastructure dependencies while production deployments benefit from Redis's performance and TTL capabilities.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}