How to Build an Agent Auth Handshake (OIDC) with Arcade’s OAuth Toolkit

AI agents require secure access to external services to perform actions on behalf of users. Arcade handles OAuth authorization flows, enabling agents to authenticate with services like Gmail, Slack, and GitHub while keeping tokens isolated from the AI model.

Prerequisites

Arcade API key
Python 3.8+ or Node.js 18+
OAuth credentials from target service provider
Basic OAuth knowledge

Agent Authentication Architecture

The Authorization Problem

AI agents operate without user interfaces. Traditional OAuth flows assume human interaction through a browser. Agents face three critical challenges:

Headless operation without browser access
Multi-user access through single agent instance
Just-in-time authorization when tools execute
Token management separate from AI decision-making

Arcade's authorization system manages OAuth flows, token storage, and refresh cycles while maintaining credential isolation.

OAuth Authorization Code Flow

Arcade implements authorization code grant flow with optional PKCE support. This flow enables:

User-specific scope authorization
Secure token exchange
Automatic token refresh
Per-user permission isolation

Configuring OAuth Providers

Cloud Provider Setup

Configure OAuth providers through the Arcade Dashboard:

Access OAuth section
Select "Add OAuth Provider"
Choose provider type (Google, GitHub, Slack)
Input Client ID and Client Secret
Copy generated redirect URL
Configure redirect URL in provider settings

Cloud redirect URL format: https://cloud.arcade.dev/api/v1/oauth/callback

Self-Hosted Configuration

Create OAuth provider in engine.yaml:

auth:
  providers:
    - id: github-provider
      description: "GitHub OAuth for agent access"
      enabled: true
      type: oauth2
      provider_id: github
      client_id: ${env:GITHUB_CLIENT_ID}
      client_secret: ${env:GITHUB_CLIENT_SECRET}

Set environment variables:

export GITHUB_CLIENT_ID="your_client_id"
export GITHUB_CLIENT_SECRET="your_client_secret"
export ARCADE_API_KEY="your_arcade_api_key"

Custom OAuth Integration

Configure custom OAuth providers for unsupported services:

auth:
  providers:
    - id: custom-oauth-provider
      enabled: true
      type: oauth2
      client_id: ${env:CUSTOM_CLIENT_ID}
      client_secret: ${env:CUSTOM_CLIENT_SECRET}
      oauth2:
        authorization_request:
          endpoint: "https://provider.com/oauth/authorize"
          params:
            response_type: code
            scope: "read write"
        token_request:
          endpoint: "https://provider.com/oauth/token"
          params:
            grant_type: authorization_code
        user_info:
          endpoint: "https://provider.com/oauth/userinfo"

Authorization Handshake Implementation

Core Authorization Flow

The authorization handshake follows this sequence:

Agent executes tool requiring OAuth
Arcade verifies user authorization status
Returns authorization URL if unauthorized
User completes OAuth in browser
Arcade captures callback and stores token
Agent executes tool with valid token

Python implementation:

from arcadepy import Arcade

client = Arcade(api_key="your_api_key")
USER_ID = "user@example.com"

# Initiate authorization
auth_response = client.tools.authorize(
    tool_name="Gmail.SendEmail",
    user_id=USER_ID
)

if auth_response.status != "completed":
    print(f"Authorize at: {auth_response.url}")
    await client.auth.wait_for_completion(auth_response)

# Execute authorized tool
result = client.tools.execute(
    tool_name="Gmail.SendEmail",
    input={
        "to": "recipient@example.com",
        "subject": "Agent Email",
        "body": "Sent via Arcade authorization"
    },
    user_id=USER_ID
)

JavaScript implementation:

import Arcade from "@arcadeai/arcadejs";

const client = new Arcade();
const userId = "user@example.com";

const authResponse = await client.tools.authorize({
    tool_name: "Gmail.SendEmail",
    user_id: userId
});

if (authResponse.status !== "completed") {
    console.log(`Authorization URL: ${authResponse.url}`);
    await client.auth.waitForCompletion(authResponse);
}

const result = await client.tools.execute({
    tool_name: "Gmail.SendEmail",
    input: {
        to: "recipient@example.com",
        subject: "Agent Email",
        body: "Sent via Arcade authorization"
    },
    user_id: userId
});

Agent Framework Integration

Handle authorization with OpenAI Agents:

from agents import Agent, Runner
from arcadepy import AsyncArcade
from agents_arcade import get_arcade_tools
from agents_arcade.errors import AuthorizationError

async def main():
    client = AsyncArcade()
    tools = await get_arcade_tools(client, toolkits=["gmail"])

    gmail_agent = Agent(
        name="Gmail Assistant",
        instructions="Manage Gmail operations",
        model="gpt-4o-mini",
        tools=tools
    )

    try:
        result = await Runner.run(
            starting_agent=gmail_agent,
            input="Send summary email",
            context={"user_id": "user@example.com"}
        )
    except AuthorizationError as e:
        print(f"Authorization required: {e.url}")
        await client.auth.wait_for_completion(e.auth_id)

        result = await Runner.run(
            starting_agent=gmail_agent,
            input="Send summary email",
            context={"user_id": "user@example.com"}
        )

Multi-User Authorization Management

Session Management

Implement user-specific authorization tracking:

from typing import Dict
from datetime import datetime
from arcadepy import Arcade

class MultiUserAuth:
    def __init__(self):
        self.client = Arcade()
        self.sessions: Dict = {}

    async def get_user_tools(self, user_id: str, toolkit: str):
        cache_key = f"{user_id}:{toolkit}"

        if cache_key in self.sessions:
            session = self.sessions[cache_key]
            if self._valid_session(session):
                return session["tools"]

        response = await self.client.tools.list(
            toolkit=toolkit,
            user_id=user_id,
            limit=30
        )

        self.sessions[cache_key] = {
            "tools": response.items,
            "timestamp": datetime.now(),
            "user_id": user_id
        }

        return response.items

    def _valid_session(self, session: Dict) -> bool:
        age = datetime.now() - session["timestamp"]
        return age.total_seconds() < 3600

    async def execute_tool(self, user_id: str, tool_name: str, input_data: Dict):
        try:
            result = await self.client.tools.execute(
                tool_name=tool_name,
                input=input_data,
                user_id=user_id
            )
            return {"success": True, "data": result.output}
        except Exception as e:
            if hasattr(e, "authorization_required"):
                auth_response = await self.client.tools.authorize(
                    tool_name=tool_name,
                    user_id=user_id
                )
                return {
                    "success": False,
                    "authorization_required": True,
                    "auth_url": auth_response.url
                }
            raise

Web Application Integration

Implement OAuth callbacks in web applications:

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, RedirectResponse
from arcadepy import Arcade

app = FastAPI()
arcade = Arcade()
pending_auth = {}

@app.post("/api/tools/execute")
async def execute_tool(request: Request):
    data = await request.json()
    user_id = data.get("user_id")
    tool_name = data.get("tool_name")
    tool_input = data.get("input", {})

    try:
        result = await arcade.tools.execute(
            tool_name=tool_name,
            input=tool_input,
            user_id=user_id
        )
        return JSONResponse({"success": True, "result": result.output})
    except Exception as e:
        if "authorization" in str(e).lower():
            auth_response = await arcade.tools.authorize(
                tool_name=tool_name,
                user_id=user_id
            )

            pending_auth[auth_response.id] = {
                "user_id": user_id,
                "tool_name": tool_name,
                "tool_input": tool_input
            }

            return JSONResponse({
                "success": False,
                "authorization_required": True,
                "auth_url": auth_response.url,
                "auth_id": auth_response.id
            })
        raise

@app.get("/api/oauth/callback")
async def oauth_callback(auth_id: str):
    if auth_id not in pending_auth:
        return JSONResponse({"error": "Invalid authorization"}, status_code=400)

    auth_data = pending_auth[auth_id]

    result = await arcade.tools.execute(
        tool_name=auth_data["tool_name"],
        input=auth_data["tool_input"],
        user_id=auth_data["user_id"]
    )

    del pending_auth[auth_id]
    return RedirectResponse(url="/success")

Custom OAuth Tool Development

Tool Creation with Authorization

Build custom tools with OAuth using Arcade SDK:

from typing import Annotated
from arcade.sdk import ToolContext, tool
from arcade.sdk.auth import Google
import httpx

@tool(
    requires_auth=Google(
        scopes=[
            "https://www.googleapis.com/auth/gmail.send",
            "https://www.googleapis.com/auth/gmail.readonly"
        ]
    )
)
async def send_gmail_with_attachment(
    context: ToolContext,
    to: Annotated[str, "Recipient email"],
    subject: Annotated[str, "Email subject"],
    body: Annotated[str, "Email body"],
    attachment_url: Annotated[str, "File URL"]
) -> Annotated[str, "Email ID"]:
    """Send Gmail with attachment using OAuth token"""

    token = context.authorization.token

    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }

    async with httpx.AsyncClient() as client:
        attachment = await client.get(attachment_url)

        response = await client.post(
            "https://gmail.googleapis.com/gmail/v1/users/me/messages/send",
            headers=headers,
            json={
                "to": to,
                "subject": subject,
                "body": body,
                "attachments": [attachment.content]
            }
        )

        result = response.json()
        return f"Sent: {result['id']}"

Generic OAuth Integration

Integrate services without pre-built support:

from arcade.sdk import tool, ToolContext
from arcade.sdk.auth import OAuth2
import httpx

@tool(
    requires_auth=OAuth2(
        id="custom-api-provider",
        scopes=["read", "write"]
    )
)
async def call_custom_api(
    context: ToolContext,
    endpoint: str,
    method: str = "GET",
    data: dict = None
):
    """Call custom OAuth API"""

    token = context.authorization.token

    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }

    async with httpx.AsyncClient() as client:
        if method == "GET":
            response = await client.get(
                f"https://api.custom.com/{endpoint}",
                headers=headers
            )
        elif method == "POST":
            response = await client.post(
                f"https://api.custom.com/{endpoint}",
                headers=headers,
                json=data
            )

        response.raise_for_status()
        return response.json()

Production Security Implementation

Token Isolation

Never expose tokens to AI models:

class SecureTokenManager:
    def __init__(self):
        self.arcade = Arcade()

    async def execute_secure(self, user_id: str, tool_name: str, input_data: dict):
        result = await self.arcade.tools.execute(
            tool_name=tool_name,
            input=input_data,
            user_id=user_id
        )

        return {
            "output": result.output,
            "metadata": result.metadata
        }

    def log_execution(self, user_id: str, tool_name: str, success: bool):
        import hashlib
        hashed_user = hashlib.sha256(user_id.encode()).hexdigest()[:8]

        log_entry = {
            "user_hash": hashed_user,
            "tool": tool_name,
            "success": success,
            "timestamp": datetime.now().isoformat()
        }

        print(f"Execution: {log_entry}")

Least Privilege Scopes

Request minimum required permissions:

# Correct: Minimal scopes
@tool(
    requires_auth=Google(
        scopes=["https://www.googleapis.com/auth/gmail.send"]
    )
)
async def send_email_only(context: ToolContext):
    pass

# Incorrect: Excessive permissions
@tool(
    requires_auth=Google(
        scopes=[
            "https://www.googleapis.com/auth/gmail.modify",
            "https://www.googleapis.com/auth/drive"
        ]
    )
)
async def send_email_bad(context: ToolContext):
    pass

PKCE Enhancement

Enable PKCE for added security:

auth:
  providers:
    - id: secure-provider
      enabled: true
      type: oauth2
      provider_id: custom
      client_id: ${env:CLIENT_ID}
      client_secret: ${env:CLIENT_SECRET}
      oauth2:
        pkce:
          enabled: true
          code_challenge_method: S256

Error Handling with Retry Logic

Implement exponential backoff:

import asyncio
from typing import Optional

async def execute_with_retry(
    client: Arcade,
    user_id: str,
    tool_name: str,
    input_data: dict,
    max_retries: int = 3
) -> Optional[dict]:

    for attempt in range(max_retries):
        try:
            result = await client.tools.execute(
                tool_name=tool_name,
                input=input_data,
                user_id=user_id
            )
            return result.output
        except Exception as e:
            if "rate_limit" in str(e).lower():
                wait = 2 ** attempt
                await asyncio.sleep(wait)
                continue

            if "authorization" in str(e).lower():
                auth_response = await client.tools.authorize(
                    tool_name=tool_name,
                    user_id=user_id
                )

                if auth_response.status != "completed":
                    await client.auth.wait_for_completion(auth_response)
                continue

            raise

    return None

Authorization Status Verification

Check user authorization before execution:

from arcadepy import Arcade

client = Arcade()

async def check_auth_status(user_id: str, tool_name: str) -> bool:
    auth_response = await client.tools.authorize(
        tool_name=tool_name,
        user_id=user_id
    )

    return auth_response.status == "completed"

# Usage
if await check_auth_status("user@example.com", "Slack.SendMessage"):
    result = await client.tools.execute(
        tool_name="Slack.SendMessage",
        input={"channel": "#general", "message": "Hello"},
        user_id="user@example.com"
    )
else:
    auth_response = await client.tools.authorize(
        tool_name="Slack.SendMessage",
        user_id="user@example.com"
    )
    print(f"Authorize: {auth_response.url}")

Troubleshooting Authorization Issues

Redirect URL Configuration

Fix redirect URL mismatches:

Verify exact URL match in OAuth provider settings
Confirm HTTPS protocol usage
Ensure public accessibility for self-hosted deployments
Match Arcade Dashboard generated URL precisely

Token Refresh Handling

Automatic token refresh implementation:

async def execute_with_refresh(client: Arcade, user_id: str, tool_name: str, input_data: dict):
    try:
        result = await client.tools.execute(
            tool_name=tool_name,
            input=input_data,
            user_id=user_id
        )
        return result
    except Exception as e:
        if "expired" in str(e).lower() or "invalid_token" in str(e).lower():
            result = await client.tools.execute(
                tool_name=tool_name,
                input=input_data,
                user_id=user_id
            )
            return result
        raise

Multiple Provider Management

Specify provider ID when using multiple providers:

from arcade.sdk.auth import Google

@tool(
    requires_auth=Google(
        id="company-google-calendar",
        scopes=["https://www.googleapis.com/auth/calendar.readonly"]
    )
)
async def list_calendar_events(context: ToolContext):
    pass

Additional Resources

Continue with these resources:

Arcade Toolkits - Pre-built integrations
API Reference - Complete API documentation
GitHub Examples - Sample applications
Arcade Documentation - Full platform guides
Tool Development - Custom tool creation

For production deployments, review self-hosted options for complete control over authentication infrastructure.

How to Build an Agent Auth Handshake (OIDC) with Arcade's OAuth Toolkit