Function _stream_agent

libs/code/deepagents_code/non_interactive.py:738–773

Consume the full agent stream and update *state* with results. Args: agent: The agent (Pregel or RemoteAgent). stream_input: Either the initial user message dict or a `Command(resume=...)` for HITL continuation. config: LangGraph runnable config (thread ID, m

(
    agent: Any,  # noqa: ANN401
    stream_input: dict[str, Any] | Command,
    config: RunnableConfig,
    state: StreamState,
    console: Console,
    file_op_tracker: FileOpTracker,
    context: CLIContext,
)

async def _stream_agent(
    agent: Any,  # noqa: ANN401
    stream_input: dict[str, Any] | Command,
    config: RunnableConfig,
    state: StreamState,
    console: Console,
    file_op_tracker: FileOpTracker,
    context: CLIContext,
) -> None:
    """Consume the full agent stream and update *state* with results.

    Args:
        agent: The agent (Pregel or RemoteAgent).
        stream_input: Either the initial user message dict or a
            `Command(resume=...)` for HITL continuation.
        config: LangGraph runnable config (thread ID, metadata, etc.).
        state: Shared stream state.
        console: Rich console for formatted output.
        file_op_tracker: Tracker for file-operation diffs.
        context: Runtime context for model-call middleware.
    """
    if state.spinner:
        state.spinner.start()
    try:
        async for chunk in agent.astream(
            stream_input,
            stream_mode=["messages", "updates", "custom"],
            subgraphs=True,
            config=config,
            context=context,
            durability="exit",
        ):
            _process_stream_chunk(chunk, state, console, file_op_tracker)
    finally:
        if state.spinner:
            state.spinner.stop()


async def _run_agent_loop(