跳转到主要内容
流式传输会逐个 token 显示输出，而不是等待完整响应生成完毕。用户可以立即看到生成进度，这对于较长输出或交互式应用非常重要。

​
一行代码开启流式传输

设置 stream=True，这样用户就能在智能体执行任务时实时看到进度。 
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())

​
搭配工具的流式传输

流式传输同样适用于调用工具的工作流。智能体在调用本地工具、MCP，或同时使用这两种方式时，都可以进行流式传输。 
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())

​
对比：非流式传输 vs 流式传输（同一场景）

下面的两个代码片段使用的是同一场景。唯一的区别在于你是否设置了 stream=True 并对该流进行迭代。
在 Python 中，非流式传输指的是 stream=False，而不是“同步（sync）”。如果你使用 AsyncDedalus，你仍然会编写异步代码并使用 asyncio.run(...)。如果你更偏好完全同步的代码，请使用 Dedalus Client（如下例所示）。

​
Python

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())

​
Python（同步 Client）

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()

​
TypeScript

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();

​
用户体验有何不同

  • 渐进式渲染：可以在文本到达时立即显示（“打字”效果），而不必等待完整响应生成完毕。
  • 可见的工作过程：在 tool/模型上下文协议 (MCP) 工作流中，可以在智能体调用工具期间展示状态更新（例如，“正在搜索 Ticketmaster…”）。
  • 可中断性：如果用户已经获得所需信息，可以在 Client 侧提前停止，而不必为完整补全结果付费。

​
何时使用流式传输

在以下情况使用流式传输：
  • 构建对感知延迟敏感的聊天界面时
  • 生成长篇内容（文章、代码、分析）时
  • 在终端或日志中运行任务、需要进度反馈时
在以下情况不要使用流式传输：
  • 需要在展示前先解析完整响应时
  • 使用结构化输出并调用 .parse() 时
  • 响应时间已经足够快时

​
后续步骤

  • 跨模型路由： Handoffs — 在不同阶段搭配使用快速/强大模型
  • 最后再加入图像： Images & Vision — 在文本工作流稳定后再增加多模态能力
  • 掌握模式： Use Cases — 更多流式传输智能体示例
以编程方式将这些文档连接 到 Claude、VSCode 等工具，通过模型上下文协议（MCP）获取实时回答。