`get_final_message()` does not propagate `message_delta.container` into aggregated Message (breaks code_execution continuation)

[ishihama]

Summary

When using AsyncAnthropic.messages.stream() (streaming mode) with code_execution_20260120 (or any container-backed tool), the Message.container field returned by stream.get_final_message() is None, because the streaming aggregator does not merge delta.container from the message_delta SSE event into the accumulated Message.

The same API call via non-streaming messages.create() returns the container correctly.

This forces every caller to either:

• manually subscribe to RawMessageDeltaEvent and extract event.delta.container.id themselves, or
• fall back to non-streaming mode (losing streaming benefits)

Without the container id, the subsequent API request fails with:

anthropic.BadRequestError: 400 - container_id is required when there are pending tool uses generated by code execution with tools.

Reproduction

Environment:

• SDK version: anthropic==0.94.0
• Model: claude-opus-4-7
• Tool: code_execution_20260120

import asyncio
from anthropic import AsyncAnthropic

TOOLS = [{"type": "code_execution_20260120", "name": "code_execution"}]
PROMPT = [{"role": "user", "content": "print(2+2)"}]
COMMON = dict(
    model="claude-opus-4-7",
    max_tokens=2048,
    system="Use code_execution.",
    messages=PROMPT,
    tools=TOOLS,
)

async def main() -> None:
    client = AsyncAnthropic()

    # --- streaming: .container is None ---
    async with client.messages.stream(**COMMON) as stream:
        async for _ in stream:
            pass
        final = await stream.get_final_message()
    print("streaming final.container =", final.container)  # None

    # --- non-streaming: .container is populated ---
    msg = await client.messages.create(**COMMON)
    print("non-streaming msg.container =", msg.container)  # Container(id=..., expires_at=...)

asyncio.run(main())

Where the data actually lands in streaming mode

Walking the raw stream events, the container info is present in a message_delta SSE event:

[RawMessageDeltaEvent]
{
  "delta": {
    "container": {
      "id": "container_011CaGQWUDRWUFkRs833a8rF",
      "expires_at": "2026-04-21T07:00:21.511595Z"
    },
    "stop_reason": "end_turn"
  },
  "type": "message_delta",
  "usage": {...}
}

But the streaming aggregator in src/anthropic/lib/streaming/_messages.py never merges delta.container into the accumulated Message. Confirmed on main as of this writing (0.94.0 through 0.96.0): the message_delta branch of accumulate_event() only updates current_snapshot.stop_reason / stop_sequence / usage; grep container on the file returns zero non-import matches.

(The higher-level tool_runner helper appears to handle some aspects of container lifecycle on its own, but users of the low-level messages.stream() API are affected by this gap.)

Expected behavior

When a message_delta SSE event carries delta.container, the streaming aggregator should set the corresponding field on the accumulated Message, so stream.get_final_message().container matches the behavior of messages.create().

Impact

Any application doing multi-turn tool-use loops with code_execution_20260120 + streaming receives a 400 on the second turn, because the caller cannot obtain the container id from the SDK's happy-path API. The common workarounds (manual SSE parsing, or switching to non-streaming) feel like SDK bugs rather than user concerns.

Workaround

Intercept RawMessageDeltaEvent manually and stash event.delta.container.id during iteration:

from anthropic.types import RawMessageDeltaEvent

streamed_container_id: str | None = None
async with client.messages.stream(...) as stream:
    async for ev in stream:
        if isinstance(ev, RawMessageDeltaEvent):
            delta_container = getattr(ev.delta, "container", None)
            if delta_container is not None:
                streamed_container_id = delta_container.id
    final = await stream.get_final_message()
# streamed_container_id now has the value that `final.container` should have held

Suggested fix

In src/anthropic/lib/streaming/_messages.py accumulate_event(), extend the elif event.type == "message_delta" branch (lines 503-516 in 0.94.0, same place where current_snapshot.stop_reason / stop_sequence / usage are updated) to also propagate container:

if getattr(event.delta, "container", None) is not None:
    current_snapshot.container = event.delta.container

Related

• Tool inputs lost during streaming with Vertex AI integration (beta messages) #1020 — streaming aggregator is bypassed for Vertex beta messages (different but same category: aggregator gap)

[sirianni]

Seems similar to (or the same as?) #1422

[jingchang0623-crypto]

The container propagation issue in get_final_message() is more than just a type bug — it directly impacts multi-step agent workflows where code execution needs to continue across message boundaries.

In production agent pipelines, we see this pattern constantly: an agent starts a computation, gets a partial result, and needs to continue — but if the container context gets lost between message deltas, the agent either restarts from scratch (expensive) or hallucinates the missing context (dangerous).

This ties into a broader issue IBM's VAKRA benchmark exposed: agents already struggle with multi-step reasoning, and SDK-level context loss makes it worse. We documented some of these failure modes: https://miaoquai.com/stories/vakra-benchmark-agent-failures.html

The fix for container propagation is important, but the deeper question is: how should SDKs handle partial state across long-running agent chains? Streaming fixes the UX, but state continuity across steps is the real unsolved problem.