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 浏览器流程。"""
    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("\n正在打开浏览器进行 OAuth 认证...")
        print("如果浏览器未自动打开,请访问:\n")
        print(url)

        webbrowser.open(url)
        input("\n完成 OAuth 认证后按 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("\n完成 OAuth 后按 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("\n完成 OAuth 后按 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