OAuth2 Flows for AI Agent Integrations: Connecting to Third-Party Services

Why AI Agents Need OAuth2

AI agents rarely operate in isolation. A customer support agent may need to read emails from Gmail, a DevOps agent may need to create issues in GitHub, and a sales agent may need to update records in Salesforce. Each of these integrations requires the agent to act on behalf of a user — accessing their data without ever seeing their password. OAuth2 is the standard protocol that makes this possible.

OAuth2 defines several grant types (flows), each suited to different scenarios. Choosing the right flow depends on whether a human user is involved, whether the client is public or confidential, and whether the agent acts on its own behalf or on behalf of a user.

The Four Flows That Matter

Authorization Code Flow — the standard for web applications where a user grants consent. The agent redirects the user to the provider, receives a code, and exchanges it for tokens server-side. Best for user-facing AI agent dashboards.

Authorization Code with PKCE — extends the authorization code flow for public clients (SPAs, mobile apps, CLI tools) that cannot securely store a client secret. Uses a code verifier and challenge to prevent interception attacks.

Client Credentials Flow — for service-to-service communication where no user is involved. The agent authenticates with its own credentials and receives a token scoped to its service account. Ideal for background agent processes.

Refresh Token Flow — not a standalone flow but a mechanism to obtain new access tokens without user interaction. Critical for long-running agents that need persistent access to third-party APIs.

Authorization Code Flow with PKCE in FastAPI

This is the most common flow for AI agent platforms where users connect their external accounts. Here is a complete implementation:

# integrations/oauth2.py
import hashlib
import secrets
import base64
from urllib.parse import urlencode
import httpx
from fastapi import APIRouter, HTTPException, Request
from fastapi.responses import RedirectResponse

router = APIRouter(prefix="/integrations/oauth2")

# Configuration per provider
PROVIDERS = {
    "github": {
        "auth_url": "https://github.com/login/oauth/authorize",
        "token_url": "https://github.com/login/oauth/access_token",
        "client_id": "your-client-id",
        "client_secret": "your-client-secret",
        "scopes": "repo read:user",
    },
}

# In production, use Redis or a database
pending_flows: dict[str, dict] = {}


def generate_pkce_pair() -> tuple[str, str]:
    """Generate a PKCE code verifier and challenge."""
    verifier = secrets.token_urlsafe(64)
    digest = hashlib.sha256(verifier.encode()).digest()
    challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode()
    return verifier, challenge


@router.get("/connect/{provider}")
async def start_oauth_flow(provider: str, request: Request):
    if provider not in PROVIDERS:
        raise HTTPException(status_code=400, detail="Unsupported provider")

    config = PROVIDERS[provider]
    state = secrets.token_urlsafe(32)
    verifier, challenge = generate_pkce_pair()

    # Store state for validation in callback
    pending_flows[state] = {
        "provider": provider,
        "code_verifier": verifier,
        "user_id": request.state.user.sub,
    }

    params = {
        "client_id": config["client_id"],
        "redirect_uri": "https://app.example.com/integrations/oauth2/callback",
        "scope": config["scopes"],
        "state": state,
        "response_type": "code",
        "code_challenge": challenge,
        "code_challenge_method": "S256",
    }
    return RedirectResponse(f"{config['auth_url']}?{urlencode(params)}")

Handling the Callback

When the provider redirects back with an authorization code, exchange it for tokens:

@router.get("/callback")
async def oauth_callback(code: str, state: str):
    flow = pending_flows.pop(state, None)
    if not flow:
        raise HTTPException(status_code=400, detail="Invalid or expired state")

    config = PROVIDERS[flow["provider"]]

    async with httpx.AsyncClient() as client:
        response = await client.post(
            config["token_url"],
            data={
                "client_id": config["client_id"],
                "client_secret": config["client_secret"],
                "code": code,
                "redirect_uri": "https://app.example.com/integrations/oauth2/callback",
                "grant_type": "authorization_code",
                "code_verifier": flow["code_verifier"],
            },
            headers={"Accept": "application/json"},
        )

    if response.status_code != 200:
        raise HTTPException(status_code=502, detail="Token exchange failed")

    tokens = response.json()
    # Store tokens securely — encrypted in database
    await store_user_tokens(
        user_id=flow["user_id"],
        provider=flow["provider"],
        access_token=tokens["access_token"],
        refresh_token=tokens.get("refresh_token"),
        expires_in=tokens.get("expires_in"),
    )
    return RedirectResponse("/dashboard?integration=connected")

Client Credentials for Service-to-Service

When an AI agent runs as a background service and needs to access an API without user involvement, the client credentials flow is appropriate:

async def get_service_token(provider_config: dict) -> str:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            provider_config["token_url"],
            data={
                "grant_type": "client_credentials",
                "client_id": provider_config["client_id"],
                "client_secret": provider_config["client_secret"],
                "scope": provider_config["scopes"],
            },
        )
    tokens = response.json()
    return tokens["access_token"]

Secure Token Storage

Never store OAuth tokens in plain text. Encrypt them at rest using a key management service or at minimum a server-side encryption key. Structure your storage to track expiration and support automatic refresh:

from datetime import datetime, timezone, timedelta
from sqlalchemy import Column, String, DateTime, Text
from cryptography.fernet import Fernet

ENCRYPTION_KEY = Fernet(b"your-fernet-key-from-env")


class UserIntegration(Base):
    __tablename__ = "user_integrations"
    user_id = Column(String, primary_key=True)
    provider = Column(String, primary_key=True)
    encrypted_access_token = Column(Text)
    encrypted_refresh_token = Column(Text, nullable=True)
    token_expires_at = Column(DateTime(timezone=True))

    def set_tokens(self, access_token: str, refresh_token: str, expires_in: int):
        self.encrypted_access_token = ENCRYPTION_KEY.encrypt(access_token.encode()).decode()
        if refresh_token:
            self.encrypted_refresh_token = ENCRYPTION_KEY.encrypt(
                refresh_token.encode()
            ).decode()
        self.token_expires_at = datetime.now(timezone.utc) + timedelta(seconds=expires_in)

    def get_access_token(self) -> str:
        return ENCRYPTION_KEY.decrypt(self.encrypted_access_token.encode()).decode()

FAQ

When should I use client credentials vs authorization code flow?

Use client credentials when the AI agent acts on its own behalf without a user context — such as a background processing agent pulling data from an internal API. Use authorization code flow when the agent needs to access resources owned by a specific user, like reading their Gmail or GitHub repos.

How do I handle token expiration in long-running agent tasks?

Implement a token refresh wrapper that checks the token_expires_at timestamp before each API call. If the token is within five minutes of expiry, refresh it proactively. For tasks that take hours, use a middleware that refreshes tokens transparently so the agent logic does not need to handle expiration directly.

Is PKCE necessary if my agent backend is a confidential client?

Strictly speaking, confidential clients that can safely store a client secret do not require PKCE. However, OAuth2.1 recommends PKCE for all clients, including confidential ones. Adding PKCE provides defense in depth against authorization code interception attacks and is considered a best practice in modern implementations.

---

#OAuth2 #APIIntegration #FastAPI #AIAgents #PKCE #ThirdPartyAPIs #AgenticAI #LearnAI #AIEngineering