OAuth

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:
    """आवश्यकता होने पर OAuth ब्राउज़र फ़्लो को संभालते हुए async फ़ंक्शन चलाएं।"""
    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("\nOAuth के लिए ब्राउज़र खोल रहे हैं...")
        print("यदि ब्राउज़र नहीं खुलता है, तो इस पर जाएं:\n")
        print(url)

        webbrowser.open(url)
        input("\nOAuth पूर्ण होने के बाद Enter दबाएं...")

        return await fn()

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("\nOAuth पूरा करने के बाद Enter दबाएं...")
        return await fn()

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

    result = await with_oauth_retry(
        lambda: runner.run(
            input="मेरे हाल के ईमेल की सूची बनाएं और उनका सारांश दें",
            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())

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("\nOAuth पूरा करने के बाद Enter दबाएं...")
        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": "मेरे हाल के ईमेल की सूची बनाएं और उनका सारांश दें",
                }
            ],
            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())

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