流式传输

import asyncio
from dedalus_labs import AsyncDedalus, DedalusRunner
from dedalus_labs.utils.stream import stream_async
from dotenv import load_dotenv

load_dotenv()

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

    stream = runner.run(
        input="Find me the nearest basketball games in January in San Francisco (stream your work).",
        model="anthropic/claude-opus-4-5",
        mcp_servers=["windsor/ticketmaster-mcp"],  # 通过 Ticketmaster 发现活动
        stream=True,
    )

    await stream_async(stream)

if __name__ == "__main__":
    asyncio.run(main())

import asyncio

from dedalus_labs import AsyncDedalus, DedalusRunner
from dedalus_labs.utils.stream import stream_async
from dotenv import load_dotenv

load_dotenv()

def summarize_headlines(headlines: list[str]) -> str:
    """Format headlines as a short bullet list."""
    return "\n".join(f"• {h}" for h in headlines[:3])

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

    stream = runner.run(
        input=(
            "Search for AI news. Extract 3 headlines. "
            "Then call summarize_headlines(headlines) and stream your final answer."
        ),
        model="openai/gpt-5.2",
        mcp_servers=["windsor/brave-search-mcp"],  # 通过 Brave Search MCP 进行网页搜索
        tools=[summarize_headlines],
        stream=True,
    )

    await stream_async(stream)

if __name__ == "__main__":
    asyncio.run(main())

import asyncio
from dedalus_labs import AsyncDedalus, DedalusRunner
from dotenv import load_dotenv

load_dotenv()

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

    result = await runner.run(
        input="Find me the nearest basketball games in January in San Francisco.",
        model="anthropic/claude-opus-4-5",
        mcp_servers=["windsor/ticketmaster-mcp"],  # 通过 Ticketmaster 发现活动
    )

    # 只有在整个运行完成后，你才能看到输出。
    print(result.final_output)

if **name** == "**main**":
asyncio.run(main())

from dedalus_labs import Dedalus, DedalusRunner
from dotenv import load_dotenv

load_dotenv()

def main():
client = Dedalus()
runner = DedalusRunner(client)

    result = runner.run(
        input="Find me the nearest basketball games in January in San Francisco.",
        model="anthropic/claude-opus-4-5",
        mcp_servers=["windsor/ticketmaster-mcp"],  # 通过 Ticketmaster 查找活动
    )

    print(result.final_output)

if **name** == "**main**":
main()

import Dedalus from "dedalus-labs";
import { DedalusRunner } from "dedalus-labs";

const client = new Dedalus();
const runner = new DedalusRunner(client, true);

async function main() {
	const result = await runner.run({
		input: "Find me the nearest basketball games in January in San Francisco.",
		model: "anthropic/claude-opus-4-5",
		mcpServers: ["windsor/ticketmaster-mcp"], // 通过 Ticketmaster 查找活动
	});

	console.log((result as any).finalOutput);
}

main();