[1fanwang]

Description

Stopping an Anthropic stream could report unfinished tool arguments as a complete call. This marks partial text and tool output incomplete and preserves the latest provider token count, including zero, without changing SDK responses.

Fixes #824.

A response without a stop event or stop reason is incomplete. The wrapper keeps unfinished tool-argument JSON as text and takes token counts from the SDK's accumulated response.

Known gaps: Streaming-thinking and tool recordings remain unavailable. Thinking coverage uses local HTTP/SSE rather than a provider recording.

Type of change

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to not work as expected)
• This change requires a documentation update

How has this been tested?

• Test A

Testing Done

Anthropic 0.51.0 and 1.13.0 produced identical results.

The tool-call input uses a synthetic local HTTP server, not a provider recording.

Before:

SOURCE dc32152ebbbb9738a35a95f6cb2beb03087f9fea REPRO repro_early_exit.py
received_arguments: '{"city": "Chica'
recorded_arguments: {}
recorded_output_tokens: 0
finish_reasons: None
sdk_output_tokens: 0

After:

SOURCE fd3c0a1ecf2c8de864f1c39e78015dfd9f56ae4f REPRO repro_early_exit.py
received_arguments: '{"city": "Chica'
recorded_arguments: '{"city": "Chica'
recorded_output_tokens: 0
finish_reasons: ('error',)
sdk_output_tokens: 0

The usage replay uses unchanged provider recordings.

Before:

SOURCE 0bb7349d943e1477dd5ca04c6c6c79bac5a536d6 REPRO repro_recorded_stream.py
{"mode": "sync-text_stream", "output": [{"finish_reason": "error", "name": null, "parts": [{"content": "Hello!", "type": "text"}], "role": "assistant"}], "received_text": "Hello!", "recorded_output_tokens": null, "sdk_output_tokens": 2, "sdk_unchanged": true}
{"mode": "async-text_stream", "output": [{"finish_reason": "error", "name": null, "parts": [{"content": "Hello.", "type": "text"}], "role": "assistant"}], "received_text": "Hello.", "recorded_output_tokens": null, "sdk_output_tokens": 2, "sdk_unchanged": true}

After:

SOURCE fd3c0a1ecf2c8de864f1c39e78015dfd9f56ae4f REPRO repro_recorded_stream.py
{"mode": "sync-text_stream", "output": [{"finish_reason": "error", "name": null, "parts": [{"content": "Hello!", "type": "text"}], "role": "assistant"}], "received_text": "Hello!", "recorded_output_tokens": 2, "sdk_output_tokens": 2, "sdk_unchanged": true}
{"mode": "async-text_stream", "output": [{"finish_reason": "error", "name": null, "parts": [{"content": "Hello.", "type": "text"}], "role": "assistant"}], "received_text": "Hello.", "recorded_output_tokens": 2, "sdk_output_tokens": 2, "sdk_unchanged": true}

No live provider calls. Save all three sources below in /tmp. Run from this checkout with its tox environments:

.tox/py310-test-instrumentation-genai-anthropic-oldest/bin/python /tmp/compare_stream_repros.py
.tox/py314-test-instrumentation-genai-anthropic-latest/bin/python /tmp/compare_stream_repros.py

Reproducer source: compare_stream_repros.py

"""Run both reproducers against their before/after production source."""

import io
import os
import pathlib
import subprocess
import sys
import tarfile
import tempfile


package = "instrumentation/opentelemetry-instrumentation-genai-anthropic/src"
head = "fd3c0a1ecf2c8de864f1c39e78015dfd9f56ae4f"
cases = [
    ("dc32152ebbbb9738a35a95f6cb2beb03087f9fea", "repro_early_exit.py"),
    (head, "repro_early_exit.py"),
    ("0bb7349d943e1477dd5ca04c6c6c79bac5a536d6", "repro_recorded_stream.py"),
    (head, "repro_recorded_stream.py"),
]
for ref, script in cases:
    with tempfile.TemporaryDirectory(prefix="otel-stream-repro-") as temporary:
        archive = subprocess.check_output(["git", "archive", ref, package])
        with tarfile.open(fileobj=io.BytesIO(archive)) as tar:
            tar.extractall(temporary, filter="data")
        env = os.environ.copy()
        env["PYTHONPATH"] = str(pathlib.Path(temporary) / package)
        print(f"\nSOURCE {ref} REPRO {script}", flush=True)
        subprocess.run(
            [sys.executable, str(pathlib.Path(__file__).with_name(script))],
            env=env,
            check=True,
        )

Reproducer source: repro_early_exit.py

"""Replay an interrupted tool call through the SDK and a local HTTP server."""

import json
import os
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from threading import Thread

import anthropic
from opentelemetry.instrumentation.genai.anthropic import AnthropicInstrumentor
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter

os.environ["OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"] = "SPAN_ONLY"


def event(data):
    return f"event: {data['type']}\ndata: {json.dumps(data)}\n\n".encode()


body = event({
    "type": "message_start",
    "message": {
        "id": "msg_local", "type": "message", "role": "assistant",
        "model": "test-model", "content": [], "stop_reason": None,
        "stop_sequence": None, "usage": {"input_tokens": 3, "output_tokens": 0},
    },
})
body += event({
    "type": "content_block_start", "index": 0,
    "content_block": {
        "type": "tool_use", "id": "toolu_A", "name": "get_weather", "input": {},
    },
})
for character in '{"city": "Chicago", "unit": "celsius"}':
    body += event({
        "type": "content_block_delta", "index": 0,
        "delta": {"type": "input_json_delta", "partial_json": character},
    })
body += event({"type": "content_block_stop", "index": 0})
body += event({
    "type": "message_delta",
    "delta": {"stop_reason": "tool_use", "stop_sequence": None},
    "usage": {"output_tokens": 10},
})
body += event({"type": "message_stop"})


class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        request = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        assert self.path == "/v1/messages" and request["stream"] is True
        self.send_response(200)
        self.send_header("Content-Type", "text/event-stream")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *_args):
        pass


server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
thread = Thread(target=server.serve_forever)
thread.start()
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
instrumentor = AnthropicInstrumentor()
instrumentor.instrument(tracer_provider=provider)
try:
    with anthropic.Anthropic(
        api_key="test_anthropic_api_key",
        base_url=f"http://127.0.0.1:{server.server_port}",
    ) as client:
        received = ""
        with client.messages.stream(
            model="test-model", max_tokens=100,
            messages=[{"role": "user", "content": "What is the weather?"}],
            tools=[{"name": "get_weather", "input_schema": {"type": "object"}}],
        ) as stream:
            for chunk in stream:
                if chunk.type == "input_json":
                    received += chunk.partial_json
                    if len(received) >= 15:
                        break
        attrs = exporter.get_finished_spans()[0].attributes
        messages = json.loads(attrs["gen_ai.output.messages"])
        print("received_arguments:", repr(received))
        print("recorded_arguments:", repr(messages[0]["parts"][0]["arguments"]))
        print("recorded_output_tokens:", attrs.get("gen_ai.usage.output_tokens"))
        print("finish_reasons:", attrs.get("gen_ai.response.finish_reasons"))
        print("sdk_output_tokens:", stream.current_message_snapshot.usage.output_tokens)
finally:
    instrumentor.uninstrument()
    provider.shutdown()
    server.shutdown()
    server.server_close()
    thread.join()

Reproducer source: repro_recorded_stream.py

"""Replay real Anthropic recordings through sync/async SDK stream accessors."""

import asyncio
import json
import os

import anthropic
import vcr
from opentelemetry.instrumentation.genai.anthropic import AnthropicInstrumentor
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter


os.environ["OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"] = "SPAN_ONLY"
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
instrumentor = AnthropicInstrumentor()
instrumentor.instrument(tracer_provider=provider)
replay = vcr.VCR(
    cassette_library_dir="instrumentation/opentelemetry-instrumentation-genai-anthropic/tests/cassettes",
    record_mode="none",
    match_on=["method", "scheme", "host", "port", "path", "query", "body"],
)
request = {
    "max_tokens": 100,
    "messages": [{"role": "user", "content": "Say hello in one word."}],
}


def report(mode, text, snapshot, before_close):
    spans = exporter.get_finished_spans()
    assert len(spans) == 1
    attrs = spans[0].attributes
    print(
        json.dumps(
            {
                "mode": mode,
                "received_text": text,
                "sdk_output_tokens": snapshot.usage.output_tokens,
                "recorded_output_tokens": attrs.get("gen_ai.usage.output_tokens"),
                "output": json.loads(attrs["gen_ai.output.messages"]),
                "sdk_unchanged": snapshot.model_dump() == before_close,
            },
            sort_keys=True,
        )
    )
    exporter.clear()


def run_sync(accessor):
    with replay.use_cassette("test_sync_messages_stream.yaml"):
        with anthropic.Anthropic(api_key="test_anthropic_api_key") as client:
            with client.messages.stream(model="claude-sonnet-4-20250514", **request) as stream:
                if accessor == "text_stream":
                    text = next(stream.text_stream)
                else:
                    for event in stream:
                        if event.type == "text":
                            text = event.text
                            break
                snapshot = stream.current_message_snapshot
                before_close = snapshot.model_dump()
                assert snapshot.stop_reason is None
            report(f"sync-{accessor}", text, snapshot, before_close)


async def run_async(accessor):
    with replay.use_cassette("test_async_messages_stream.yaml"):
        async with anthropic.AsyncAnthropic(api_key="test_anthropic_api_key") as client:
            async with client.messages.stream(model="claude-haiku-4-5", **request) as stream:
                if accessor == "text_stream":
                    text = await anext(stream.text_stream)
                else:
                    async for event in stream:
                        if event.type == "text":
                            text = event.text
                            break
                snapshot = stream.current_message_snapshot
                before_close = snapshot.model_dump()
                assert snapshot.stop_reason is None
            report(f"async-{accessor}", text, snapshot, before_close)


try:
    for accessor in ("events", "text_stream"):
        run_sync(accessor)
        asyncio.run(run_async(accessor))
finally:
    instrumentor.uninstrument()
    provider.shutdown()

Early thinking exit:

Anthropic SDK 1.12.0 received a local HTTP/SSE stream with a thinking block followed by text. The observation below uses the committed local-server fixtures and an in-memory span exporter. No provider request was made; the package has no recorded streaming-thinking cassette.

Save the source below as thinking-sse-observation.py in the Anthropic package directory of the 4e4beb5 checkout. Keep the package's installed dependencies and test fixtures from that checkout while comparing the two wrapper versions.

cd instrumentation/opentelemetry-instrumentation-genai-anthropic
git show 227ea2898ef125cd1a00858c3d4eea7164c30f72:instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/wrappers.py > src/opentelemetry/instrumentation/genai/anthropic/wrappers.py
env -u ANTHROPIC_API_KEY ../../.tox/py314-test-instrumentation-genai-anthropic-latest/bin/python thinking-sse-observation.py

At 227ea28, the process exited 1:

received_text='The answer.'
sdk_snapshot_unchanged=True; output_tokens=0
span_output_messages=[{'role': 'assistant', 'parts': [{'content': 'The answer.', 'type': 'text'}], 'finish_reason': 'error', 'name': None}]
AssertionError

git show 4e4beb5d4a0639cbe610eeee32292754216f5fdc:instrumentation/opentelemetry-instrumentation-genai-anthropic/src/opentelemetry/instrumentation/genai/anthropic/wrappers.py > src/opentelemetry/instrumentation/genai/anthropic/wrappers.py
env -u ANTHROPIC_API_KEY ../../.tox/py314-test-instrumentation-genai-anthropic-latest/bin/python thinking-sse-observation.py

At 4e4beb5, the process exited 0:

received_text='The answer.'
sdk_snapshot_unchanged=True; output_tokens=0
span_output_messages=[{'role': 'assistant', 'parts': [{'content': 'Reasoning before the answer.', 'type': 'reasoning'}, {'content': 'The answer.', 'type': 'text'}], 'finish_reason': 'error', 'name': None}]

Reproducer source: thinking-sse-observation.py

import json
import sys
from pathlib import Path

from anthropic import Anthropic
from opentelemetry.instrumentation.genai.anthropic import AnthropicInstrumentor
from opentelemetry.sdk._logs import LoggerProvider
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import InMemoryMetricReader
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
    InMemorySpanExporter,
)
from opentelemetry.semconv._incubating.attributes import (
    gen_ai_attributes as GenAI,
)
from opentelemetry.test_util_genai.instrumentor import instrument

sys.path.insert(0, str(Path.cwd() / "tests"))
from conftest import (
    incomplete_thinking_text_stream_sse_body,
    local_anthropic_sse_server,
)

thinking = "Reasoning before the answer."
text = "The answer."
exporter = InMemorySpanExporter()
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(exporter))

with instrument(
    AnthropicInstrumentor(),
    tracer_provider=tracer_provider,
    logger_provider=LoggerProvider(),
    meter_provider=MeterProvider(
        metric_readers=[InMemoryMetricReader()]
    ),
    content_capture="SPAN_ONLY",
):
    with local_anthropic_sse_server(
        incomplete_thinking_text_stream_sse_body(
            thinking=thinking,
            text=text,
        )
    ) as base_url:
        client = Anthropic(
            api_key="test_anthropic_api_key",
            base_url=base_url,
        )
        with client.messages.stream(
            model="claude-sonnet-4-20250514",
            max_tokens=100,
            thinking={"type": "enabled", "budget_tokens": 16},
            messages=[{"role": "user", "content": "Calculate 2 + 2."}],
        ) as stream:
            received_text = next(stream.text_stream)
            snapshot = stream.current_message_snapshot
            before_close = snapshot.model_dump()
        client.close()

span = exporter.get_finished_spans()[0]
output_messages = json.loads(
    span.attributes[GenAI.GEN_AI_OUTPUT_MESSAGES]
)
print(f"received_text={received_text!r}")
print(
    "sdk_snapshot_unchanged="
    f"{snapshot.model_dump() == before_close}; "
    f"output_tokens={snapshot.usage.output_tokens}"
)
print(f"span_output_messages={output_messages!r}")
assert received_text == text
assert snapshot.model_dump() == before_close
assert snapshot.usage.output_tokens == 0
assert output_messages[0]["parts"] == [
    {"type": "reasoning", "content": thinking},
    {"type": "text", "content": text},
]

Checklist

See CONTRIBUTING.md
for the style guide, changelog guidance, and more.

• Followed the style guidelines of this project
• Changelog updated if the change requires an entry
• Unit tests added
• Documentation updated (not applicable)

Documentation changes are not applicable.

[opentelemetry-pr-dashboard]

Pull request dashboard status

Waiting on the author · refreshed 2026-10-23 22:10 UTC

Respond to 1 review item (e.g. link a commit, explain why not, ask a follow-up):

• Inline threads: 1

Status above doesn't look right?

• Just replied or pushed? Anything around or after the refresh time above may not be picked up yet — give it a few minutes.
• Should this be with reviewers? Comment /dashboard route:reviewers to route it to them.
• Anything wrong — including the routing? Report it with what you expected; it helps us improve the dashboard.

[Copilot]

Copilot review overview

🟡 Changes recommended

Finalization mutates caller-visible SDK data and drops captured partial text when callers abandon text_stream.

Review effort: Balanced
Findings: 2 · 1

Open (3)

• Preserve SDK snapshot output_tokens on early exit · New
• Preserve partial text from early-exited text streams · New
• Avoid buffering content blocks when capture is disabled · New

What changed in this PR

Updates Anthropic streaming telemetry for #824 to distinguish abandoned responses from completed ones.

Changes:

• Tracks partial content, marks incomplete responses, and omits unknown output-token counts.
• Adds sync and async early-exit regression tests and a changelog entry.

File	Description
instrumentation/​opentelemetry-instrumentation-genai-anthropic/​tests/​test_sync_messages.py	Tests synchronous early exits during tool calls.
instrumentation/​opentelemetry-instrumentation-genai-anthropic/​tests/​test_async_messages.py	Tests asynchronous early exits during tool calls.
instrumentation/​opentelemetry-instrumentation-genai-anthropic/​src/​opentelemetry/​instrumentation/​genai/​anthropic/​wrappers.py	Tracks partial responses and adjusts finalization.
instrumentation/​opentelemetry-instrumentation-genai-anthropic/​.changelog/​838.fixed	Documents the early-exit correction.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[lmolkova]

Please use recorded SDK calls for these regressions instead of transport mocks. Extend the existing test_sync_messages_stream_closed_early_by_caller and test_sync_messages_stream_interrupted_mid_iteration in this file, with matching async coverage.

[lmolkova]

Breaking after the first text chunk of an extended-thinking response now drops the preceding reasoning. text_stream bypasses event tracking, and this fallback only copies text from the SDK snapshot. Preserve thinking blocks too.

Suggested change

	if block.type == "text":
	if block.type in ("text", "thinking"):

Regression test for tests/test_sync_messages.py, using its existing imports and fixtures. This needs a real streaming cassette and matching async coverage. It has not been run; the expected failure is an empty recorded reasoning list despite nonempty reasoning in the SDK snapshot.

@pytest.mark.vcr()
def test_sync_messages_text_stream_early_exit_preserves_thinking(
    span_exporter, anthropic_client, instrument_with_content
):
    with anthropic_client.messages.stream(
        model="claude-sonnet-4-20250514",
        max_tokens=16000,
        thinking={"type": "enabled", "budget_tokens": 10000},
        messages=[
            {"role": "user", "content": "What is 17*19? Think first."}
        ],
    ) as stream:
        text = next(stream.text_stream)
        snapshot = stream.current_message_snapshot
        reasoning = [
            {"type": "reasoning", "content": block.thinking}
            for block in snapshot.content
            if block.type == "thinking"
        ]
        assert reasoning
        assert all(part["content"] for part in reasoning)
        assert text
        assert snapshot.stop_reason is None

    spans = span_exporter.get_finished_spans()
    assert len(spans) == 1
    span = spans[0]
    output = _load_span_messages(
        span, GenAIAttributes.GEN_AI_OUTPUT_MESSAGES
    )
    assert len(output) == 1
    assert [
        part for part in output[0]["parts"]
        if part["type"] == "reasoning"
    ] == reasoning
    assert output[0]["finish_reason"] == "error"

[opentelemetry-pr-dashboard]

Hi @1fanwang — just a friendly reminder that this pull request is waiting on you. The dashboard status comment has the open items and is kept current.

• Replying is enough to hand it off — answer, explain why no change is needed, or ask a follow-up. The dashboard routes it onward once nothing on the list is waiting on you.
• To hand it back for any other reason, including the dashboard getting this wrong, comment /dashboard route:reviewers.