OAuth

Use OAuth for MCP servers that require user consent, like Gmail, Google Calendar, or other services with delegated access.

OAuth Flow

The OAuth flow is triggered when you call an MCP server that requires user authentication.

Request Without Token

The SDK calls the MCP server. If no valid token exists, the server returns 401 with a WWW-Authenticate header.

Protected Resource Discovery

The SDK fetches /.well-known/oauth-protected-resource (RFC 9728) to discover the authorization server and supported scopes.

AuthenticationError

The SDK raises AuthenticationError containing a connect_url—the full OAuth authorization URL.

Browser Interaction

Your app opens the user’s browser to the connect_url. The user logs in and grants (or denies) the requested scopes.

Token Exchange

Upon approval, the authorization server exchanges the authorization code for tokens using PKCE. DAuth stores the tokens server-side.

Retry Request

The user returns to your app and triggers a retry. The SDK re-sends the request, now with valid credentials.

Authenticated Requests

The access token is automatically included for subsequent requests to the MCP server.

Token Refresh

If the access token expires, DAuth automatically uses the refresh token to obtain a new access token.

How It Works

OAuth Retry Helper

Handle the OAuth flow with a retry wrapper:

import webbrowser
from collections.abc import Awaitable, Callable
from typing import TypeVar

from dedalus_labs import AuthenticationError

T = TypeVar("T")

async def with_oauth_retry(fn: Callable[[], Awaitable[T]]) -> T:
    """Run async function, handling OAuth browser flow if needed."""
    try:
        return await fn()
    except AuthenticationError as e:
        body = e.body if isinstance(e.body, dict) else {}
        url = body.get("connect_url") or body.get("detail", {}).get("connect_url")
        if not url:
            raise

        print("\nOpening browser for OAuth...")
        print("If the browser does not open, visit:\n")
        print(url)

        webbrowser.open(url)
        input("\nPress Enter after completing OAuth...")

        return await fn()

Full Example: DedalusRunner

import asyncio
import webbrowser
from collections.abc import Awaitable, Callable
from typing import TypeVar

from dotenv import load_dotenv

load_dotenv()

from dedalus_labs import AsyncDedalus, AuthenticationError, DedalusRunner

T = TypeVar("T")

async def with_oauth_retry(fn: Callable[[], Awaitable[T]]) -> T:
    try:
        return await fn()
    except AuthenticationError as e:
        body = e.body if isinstance(e.body, dict) else {}
        url = body.get("connect_url") or body.get("detail", {}).get("connect_url")
        if not url:
            raise
        webbrowser.open(url)
        input("\nPress Enter after completing OAuth...")
        return await fn()

async def main():
    client = AsyncDedalus()
    runner = DedalusRunner(client)

    result = await with_oauth_retry(
        lambda: runner.run(
            input="List my recent emails and summarize them",
            model="openai/gpt-4.1",
            mcp_servers=["anny_personal/gmail-mcp"],
        )
    )

    print(result.output)

    if result.mcp_results:
        for r in result.mcp_results:
            print(f"{r.tool_name} ({r.duration_ms}ms): {r.result}")

asyncio.run(main())

Full Example: Raw Client

For single requests with full control over API response:

import asyncio
import webbrowser
from collections.abc import Awaitable, Callable
from typing import TypeVar

from dotenv import load_dotenv

load_dotenv()

from dedalus_labs import AsyncDedalus, AuthenticationError

T = TypeVar("T")

async def with_oauth_retry(fn: Callable[[], Awaitable[T]]) -> T:
    try:
        return await fn()
    except AuthenticationError as e:
        body = e.body if isinstance(e.body, dict) else {}
        url = body.get("connect_url") or body.get("detail", {}).get("connect_url")
        if not url:
            raise
        webbrowser.open(url)
        input("\nPress Enter after completing OAuth...")
        return await fn()

async def main():
    client = AsyncDedalus()

    async def do_request():
        return await client.chat.completions.create(
            model="openai/gpt-4.1",
            messages=[
                {
                    "role": "user",
                    "content": "List my recent emails and summarize them",
                }
            ],
            mcp_servers=["anny_personal/gmail-mcp"],
        )

    resp = await with_oauth_retry(do_request)

    print(resp.choices[0].message.content)

    if resp.mcp_tool_results:
        for r in resp.mcp_tool_results:
            print(f"{r.tool_name} ({r.duration_ms}ms): {r.result}")

asyncio.run(main())

Environment

# .env
DEDALUS_API_KEY=dsk-...
DEDALUS_API_URL=https://api.dedaluslabs.ai
DEDALUS_AS_URL=https://as.dedaluslabs.ai

No OAuth credentials needed client-side. The MCP server handles OAuth configuration, and DAuth manages token storage.

When to Use

OAuth works for:

User-facing applications
Delegated access (acting on behalf of users)
Services like Gmail, Google Calendar, Linear, GitHub

Use Bearer Auth instead for:

API keys and service tokens
Backend integrations without user context
Service-to-service calls

Python

TypeScript

OAuth Flow

How It Works

OAuth Retry Helper

Full Example: DedalusRunner

Full Example: Raw Client

Environment

When to Use

​OAuth Flow

​How It Works

​OAuth Retry Helper

​Full Example: DedalusRunner

​Full Example: Raw Client

​Environment

​When to Use

OAuth Flow

How It Works

OAuth Retry Helper

Full Example: DedalusRunner

Full Example: Raw Client

Environment

When to Use