Skip to main content
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.
1

Request Without Token

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

Protected Resource Discovery

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

AuthenticationError

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

Browser Interaction

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

Token Exchange

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

Retry Request

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

Authenticated Requests

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

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-live-...
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