From ae9d608afeb8a1ed203d63fefe62665b1caf8722 Mon Sep 17 00:00:00 2001 From: Nikhil Pareek Date: Sun, 1 Nov 2026 15:01:40 -0800 Subject: [PATCH 1/5] fix(anthropic): trace each call once when the SDK already traces it natively Anthropic SDK 1.14.0 and later emit their own span for every API call. When that span reaches the tracer provider AnthropicInstrumentor exports to (the client's open_telemetry tracer_provider, or a shared global provider), the instrumentor added a second span for the same call, with a different input token count (3 vs the SDK's cache-inclusive 12). The Messages create (sync and async) and count_tokens wrappers now leave the call to the SDK only in that case, read from the client's resolved public open_telemetry option at call time. When native tracing is disabled, goes to another provider, or has no provider, the instrumentor keeps its span, so its provider never loses the call. Sampling decisions are never used as a fallback signal. A one-time warning explains that TraceConfig and FI_HIDE_* do not apply to the SDK's spans and how to keep traceAI spans (open_telemetry=False). Adds offline ownership and loopback OTLP tests, and a CI matrix for anthropic 0.41.0 (declared floor), 1.14.0 (first native release) and 2.3.0 (latest). Co-Authored-By: Claude Opus 5.5 --- .github/workflows/anthropic-python-tests.yml | 68 ++++ python/frameworks/anthropic/README.md | 11 + .../anthropic/tests/test_native_otlp.py | 136 +++++++ .../anthropic/tests/test_native_ownership.py | 356 ++++++++++++++++++ .../anthropic/traceai_anthropic/__init__.py | 16 +- .../anthropic/traceai_anthropic/_native.py | 68 ++++ .../anthropic/traceai_anthropic/_wrappers.py | 25 +- 7 files changed, 673 insertions(+), 7 deletions(-) create mode 100644 .github/workflows/anthropic-python-tests.yml create mode 100644 python/frameworks/anthropic/tests/test_native_otlp.py create mode 100644 python/frameworks/anthropic/tests/test_native_ownership.py create mode 100644 python/frameworks/anthropic/traceai_anthropic/_native.py diff --git a/.github/workflows/anthropic-python-tests.yml b/.github/workflows/anthropic-python-tests.yml new file mode 100644 index 000000000..e93f588c6 --- /dev/null +++ b/.github/workflows/anthropic-python-tests.yml @@ -0,0 +1,68 @@ +name: Anthropic Python compatibility + +# traceAI-anthropic on the oldest SDK it supports, the first SDK with native +# OpenTelemetry tracing, and the latest SDK. python-tests.yml covers the locked SDK. + +on: + pull_request: + paths: + - 'python/frameworks/anthropic/**' + - 'python/fi_instrumentation/**' + - 'python/tests/harness/**' + - '.github/workflows/anthropic-python-tests.yml' + push: + branches: [dev] + paths: + - 'python/frameworks/anthropic/**' + - 'python/fi_instrumentation/**' + - 'python/tests/harness/**' + - '.github/workflows/anthropic-python-tests.yml' + +permissions: + contents: read + +jobs: + contract: + name: Anthropic Python (anthropic ${{ matrix.sdk }}) + runs-on: ubuntu-24.04 + timeout-minutes: 15 + strategy: + fail-fast: false + matrix: + include: + # Declared floor (pyproject.toml), before native tracing. + - sdk: '0.41.0' + python: '3.10' + extra: 'httpx<0.28' + # First SDK with native OpenTelemetry tracing. + - sdk: '1.14.0' + python: '3.10' + extra: '' + # Latest release. + - sdk: '2.3.0' + python: '3.12' + extra: '' + steps: + - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2 + with: + persist-credentials: false + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5.6.0 + with: + python-version: ${{ matrix.python }} + - name: Install source integration and the selected SDK + env: + SDK: ${{ matrix.sdk }} + EXTRA: ${{ matrix.extra }} + run: | + python -m venv .venv + .venv/bin/pip install -e python -e python/frameworks/anthropic "anthropic==$SDK" opentelemetry-sdk pytest pytest-asyncio ${EXTRA:+"$EXTRA"} + .venv/bin/python -c 'import anthropic; print("anthropic", anthropic.__version__)' + - name: Run with outbound network disabled + run: | + sudo unshare --net -- bash -e -c ' + ip link set lo up + cd "$2/python/frameworks/anthropic" + "$1" -m pytest tests -q -p no:cacheprovider -o addopts="" + cd "$2/python" + FI_BASE_URL=http://127.0.0.1:9 FI_GRPC_URL=http://127.0.0.1:9 "$1" -m pytest tests/test_framework_anthropic.py -q -p no:cacheprovider + ' _ "$PWD/.venv/bin/python" "$PWD" diff --git a/python/frameworks/anthropic/README.md b/python/frameworks/anthropic/README.md index 72134010b..489107607 100644 --- a/python/frameworks/anthropic/README.md +++ b/python/frameworks/anthropic/README.md @@ -47,6 +47,17 @@ from traceai_anthropic import AnthropicInstrumentor AnthropicInstrumentor().instrument(tracer_provider=trace_provider) ``` +### Anthropic SDK native tracing +Anthropic SDK 1.14.0 and later can trace Messages calls themselves. To send those spans to Future AGI without this instrumentor, pass the provider to the client: + +```python +client = anthropic.Anthropic(open_telemetry={"tracer_provider": trace_provider}) +``` + +The SDK's spans include model, timing and token usage, and include prompts and responses only with `"traces": {"content_mode": "content"}`. `TraceConfig` and `FI_HIDE_*` settings don't apply to them. + +When the SDK's spans reach the same tracer provider as `AnthropicInstrumentor` (because you pass it to the client, or because it is the global provider), the instrumentor leaves `messages.create` and `messages.count_tokens` calls to the SDK, so each call is recorded once. To keep this instrumentor's spans and privacy controls instead, create the client with `open_telemetry=False`. + ### Create Anthropic Components Set up your Anthropic client with built-in observability. diff --git a/python/frameworks/anthropic/tests/test_native_otlp.py b/python/frameworks/anthropic/tests/test_native_otlp.py new file mode 100644 index 000000000..5e157028c --- /dev/null +++ b/python/frameworks/anthropic/tests/test_native_otlp.py @@ -0,0 +1,136 @@ +"""The SDK's native spans reach the Future AGI collector path with what Observe reads. + +Real path: Anthropic SDK native tracing -> fi_instrumentation register() provider -> +OTLP/HTTP exporter -> loopback harness Receiver. Synthetic transport and credentials; +nothing leaves 127.0.0.1. +""" + +import inspect +import json +import sys +from pathlib import Path + +import anthropic +import pytest + +try: + import httpx2 as httpx +except ImportError: + import httpx + +_TESTS_DIR = Path(__file__).resolve().parents[3] / "tests" +if not (_TESTS_DIR / "harness" / "__init__.py").exists(): + pytest.skip("needs the repository's OTLP test harness", allow_module_level=True) +sys.path.insert(0, str(_TESTS_DIR)) + +from harness import Receiver # noqa: E402 + +NATIVE = "open_telemetry" in inspect.signature(anthropic.Anthropic).parameters +pytestmark = pytest.mark.skipif(not NATIVE, reason="SDK predates native OpenTelemetry") + +PROMPT = "synthetic prompt" +ANSWER = "synthetic answer" +PARAMS = dict(model="local-test", max_tokens=16, messages=[{"role": "user", "content": PROMPT}]) +MESSAGE = { + "id": "msg_local", + "type": "message", + "role": "assistant", + "content": [{"type": "text", "text": ANSWER}], + "model": "local-test", + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": { + "input_tokens": 3, + "output_tokens": 2, + "cache_creation_input_tokens": 4, + "cache_read_input_tokens": 5, + }, +} + + +def respond(request): + return httpx.Response(200, json=MESSAGE) + + +def attributes(span): + values = {} + for item in span.get("attributes", []): + value = item["value"] + values[item["key"]] = next(iter(value.values())) if value else None + return values + + +@pytest.fixture +def collector(monkeypatch): + with Receiver() as receiver: + monkeypatch.setenv("FI_BASE_URL", receiver.origin) + monkeypatch.setenv("FI_API_KEY", "synthetic-api-key") + monkeypatch.setenv("FI_SECRET_KEY", "synthetic-secret-key") + yield receiver + + +def register_provider(): + from fi_instrumentation import register + from fi_instrumentation.fi_types import ProjectType + + return register(project_type=ProjectType.OBSERVE, project_name="anthropic_native_contract", verbose=False) + + +def send(provider, content_mode=None): + traces = {"content_mode": content_mode} if content_mode else {} + with anthropic.Anthropic( + api_key="synthetic", + base_url="http://localhost", + http_client=httpx.Client(transport=httpx.MockTransport(respond)), + open_telemetry={"tracer_provider": provider, "traces": traces}, + ) as client: + assert client.messages.create(**PARAMS).content[0].text == ANSWER + + +@pytest.mark.parametrize("instrumented", [False, True]) +def test_native_span_reaches_the_collector_once_with_model_usage_and_content(collector, instrumented): + provider = register_provider() + instrumentor = None + if instrumented: + from traceai_anthropic import AnthropicInstrumentor + + instrumentor = AnthropicInstrumentor() + instrumentor.instrument(tracer_provider=provider) + try: + send(provider, content_mode="content") + provider.force_flush() + finally: + if instrumentor is not None: + instrumentor.uninstrument() + provider.shutdown() + + requests = collector.requests() + assert requests and {r["path"] for r in requests} == {"/tracer/v1/traces"} + assert requests[0]["headers"]["x-api-key"] == "synthetic-api-key" + assert requests[0]["headers"]["x-secret-key"] == "synthetic-secret-key" + spans = collector.spans() + assert [span["name"] for span in spans] == ["anthropic.messages.create"] + values = attributes(spans[0]) + assert values["gen_ai.provider.name"] == "anthropic" + assert values["gen_ai.operation.name"] == "chat" + assert values["gen_ai.request.model"] == "local-test" + assert values["gen_ai.response.model"] == "local-test" + # Native input usage already includes cache reads and writes. + assert int(values["gen_ai.usage.input_tokens"]) == 12 + assert int(values["gen_ai.usage.output_tokens"]) == 2 + assert PROMPT in json.loads(values["gen_ai.input.messages"])[0]["parts"][0]["content"] + assert ANSWER in json.loads(values["gen_ai.output.messages"])[0]["parts"][0]["content"] + + +def test_native_default_exports_metadata_without_content(collector): + provider = register_provider() + try: + send(provider) + provider.force_flush() + finally: + provider.shutdown() + spans = collector.spans() + assert [span["name"] for span in spans] == ["anthropic.messages.create"] + payload = json.dumps(spans) + assert PROMPT not in payload and ANSWER not in payload + assert attributes(spans[0])["gen_ai.request.model"] == "local-test" diff --git a/python/frameworks/anthropic/tests/test_native_ownership.py b/python/frameworks/anthropic/tests/test_native_ownership.py new file mode 100644 index 000000000..188d45798 --- /dev/null +++ b/python/frameworks/anthropic/tests/test_native_ownership.py @@ -0,0 +1,356 @@ +"""One span per Messages call in each tracer provider, with SDKs that trace natively. + +Anthropic SDK 1.14.0 and later emit their own OpenTelemetry span for every API call. +When that span reaches the provider this instrumentor exports to, the instrumentor +leaves the call to the SDK. When it does not, the instrumentor keeps its own span, +so the instrumentor's provider never loses the call. Real SDK, synthetic transport. +""" + +import asyncio +import inspect +import json +import logging + +import anthropic +import pytest +from opentelemetry import trace +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.sdk.trace.sampling import Decision, Sampler, SamplingResult +from traceai_anthropic import AnthropicInstrumentor + +try: + import httpx2 as httpx +except ImportError: + import httpx + +NATIVE = "open_telemetry" in inspect.signature(anthropic.Anthropic).parameters +pytestmark = pytest.mark.skipif(not NATIVE, reason="SDK predates native OpenTelemetry") + +NATIVE_SCOPE = "com.anthropic.sdk.python" +PARAMS = dict( + model="local-test", + max_tokens=16, + messages=[{"role": "user", "content": "synthetic prompt"}], +) +MESSAGE = { + "id": "msg_local", + "type": "message", + "role": "assistant", + "content": [{"type": "text", "text": "synthetic answer"}], + "model": "local-test", + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": { + "input_tokens": 3, + "output_tokens": 2, + "cache_creation_input_tokens": 4, + "cache_read_input_tokens": 5, + }, +} +EVENTS = [ + {"type": "message_start", "message": {**MESSAGE, "content": [], "stop_reason": None}}, + {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}, + {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "synthetic answer"}}, + {"type": "content_block_stop", "index": 0}, + { + "type": "message_delta", + "delta": {"stop_reason": "end_turn", "stop_sequence": None}, + "usage": {"output_tokens": 2}, + }, + {"type": "message_stop"}, +] + + +def respond(request): + if request.url.path.endswith("count_tokens"): + return httpx.Response(200, json={"input_tokens": 12}) + if json.loads(request.content).get("stream"): + body = "".join(f"event: {e['type']}\ndata: {json.dumps(e)}\n\n" for e in EVENTS) + return httpx.Response(200, headers={"content-type": "text/event-stream"}, content=body) + return httpx.Response(200, json=MESSAGE) + + +def fail(request): + error = {"type": "error", "error": {"type": "invalid_request_error", "message": "synthetic error"}} + return httpx.Response(400, json=error) + + +def make_provider(sampler=None): + exporter = InMemorySpanExporter() + provider = TracerProvider(sampler=sampler) if sampler else TracerProvider() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + return provider, exporter + + +def scopes(exporter): + return [span.instrumentation_scope.name for span in exporter.get_finished_spans()] + + +@pytest.fixture +def providers(): + fi, fi_exporter = make_provider() + other, other_exporter = make_provider() + yield fi, fi_exporter, other, other_exporter + fi.shutdown() + other.shutdown() + + +@pytest.fixture +def global_provider(monkeypatch): + """Stand in for trace.set_tracer_provider, which a process may call only once. + + Both the SDK and this instrumentor look the global provider up through + opentelemetry.trace.get_tracer_provider at call time. + """ + + def install(provider): + monkeypatch.setattr(trace, "get_tracer_provider", lambda: provider) + + return install + + +@pytest.fixture +def instrument(): + instrumentors = [] + + def start(**kwargs): + instrumentor = AnthropicInstrumentor() + instrumentor.instrument(**kwargs) + instrumentors.append(instrumentor) + return instrumentor + + yield start + for instrumentor in instrumentors: + instrumentor.uninstrument() + + +def call(operation, asynchronous, handler=respond, **client_options): + client_options.setdefault("max_retries", 0) + options = dict(api_key="synthetic", base_url="http://localhost", **client_options) + if asynchronous: + + async def run(): + transport = httpx.MockTransport(handler) + async with anthropic.AsyncAnthropic( + **options, http_client=httpx.AsyncClient(transport=transport) + ) as client: + if operation == "create": + message = await client.messages.create(**PARAMS) + assert message.content[0].text == "synthetic answer" + elif operation == "raw_stream": + async with await client.messages.create(**PARAMS, stream=True) as stream: + assert [event.type async for event in stream] == [e["type"] for e in EVENTS] + elif operation == "helper_stream": + async with client.messages.stream(**PARAMS) as stream: + assert (await stream.get_final_message()).content[0].text == "synthetic answer" + else: + raise AssertionError(operation) + + asyncio.run(run()) + return + with anthropic.Anthropic( + **options, http_client=httpx.Client(transport=httpx.MockTransport(handler)) + ) as client: + if operation == "create": + assert client.messages.create(**PARAMS).content[0].text == "synthetic answer" + elif operation == "raw_stream": + with client.messages.create(**PARAMS, stream=True) as stream: + assert [event.type for event in stream] == [e["type"] for e in EVENTS] + elif operation == "helper_stream": + with client.messages.stream(**PARAMS) as stream: + assert stream.get_final_message().content[0].text == "synthetic answer" + elif operation == "count_tokens": + counted = client.messages.count_tokens(model=PARAMS["model"], messages=PARAMS["messages"]) + assert counted.input_tokens == 12 + else: + raise AssertionError(operation) + + +OPERATIONS = [ + ("create", False), + ("create", True), + ("raw_stream", False), + ("raw_stream", True), + ("count_tokens", False), +] + + +@pytest.mark.parametrize("operation,asynchronous", OPERATIONS) +def test_native_span_in_the_instrumentors_provider_is_the_only_span( + providers, instrument, operation, asynchronous +): + fi, fi_exporter, _, _ = providers + instrument(tracer_provider=fi) + call(operation, asynchronous, open_telemetry={"tracer_provider": fi}) + assert scopes(fi_exporter) == [NATIVE_SCOPE] + + +@pytest.mark.parametrize("operation,asynchronous", OPERATIONS) +@pytest.mark.parametrize("explicit", [True, False]) +def test_global_provider_shared_with_the_sdk_gets_one_span( + providers, global_provider, instrument, operation, asynchronous, explicit +): + # register(set_global_tracer_provider=True) followed by instrument(), with or + # without passing the same provider explicitly. + fi, fi_exporter, _, _ = providers + global_provider(fi) + instrument(**({"tracer_provider": fi} if explicit else {})) + call(operation, asynchronous) + assert scopes(fi_exporter) == [NATIVE_SCOPE] + + +@pytest.mark.parametrize("operation,asynchronous", OPERATIONS) +@pytest.mark.parametrize("native_destination", ["global", "client"]) +def test_instrumentors_provider_keeps_the_call_when_native_spans_go_elsewhere( + providers, global_provider, instrument, operation, asynchronous, native_destination +): + # Future AGI registered without becoming global, while another vendor owns the + # global provider (or the client names another provider). + fi, fi_exporter, other, other_exporter = providers + if native_destination == "global": + global_provider(other) + call_options = {} + else: + call_options = {"open_telemetry": {"tracer_provider": other}} + instrument(tracer_provider=fi) + call(operation, asynchronous, **call_options) + assert scopes(fi_exporter) == ["traceai_anthropic"] + assert scopes(other_exporter) == [NATIVE_SCOPE] + + +@pytest.mark.parametrize("operation,asynchronous", OPERATIONS) +def test_non_global_provider_keeps_the_call_when_sdk_has_no_provider( + providers, instrument, operation, asynchronous +): + # The documented quickstart: register() without a global provider. + fi, fi_exporter, _, _ = providers + assert isinstance(trace.get_tracer_provider(), trace.ProxyTracerProvider) + instrument(tracer_provider=fi) + call(operation, asynchronous) + assert scopes(fi_exporter) == ["traceai_anthropic"] + + +@pytest.mark.parametrize("operation,asynchronous", OPERATIONS) +@pytest.mark.parametrize("how", ["option", "environment"]) +def test_disabled_native_tracing_keeps_the_instrumentor( + providers, global_provider, instrument, monkeypatch, operation, asynchronous, how +): + fi, fi_exporter, _, _ = providers + global_provider(fi) + instrument(tracer_provider=fi) + if how == "environment": + monkeypatch.setenv("ANTHROPIC_OPEN_TELEMETRY", "false") + call(operation, asynchronous) + else: + call(operation, asynchronous, open_telemetry=False) + assert scopes(fi_exporter) == ["traceai_anthropic"] + + +@pytest.mark.parametrize("asynchronous", [False, True]) +def test_helper_stream_is_traced_once_by_the_sdk(providers, instrument, asynchronous): + fi, fi_exporter, _, _ = providers + instrument(tracer_provider=fi) + call("helper_stream", asynchronous, open_telemetry={"tracer_provider": fi}) + assert scopes(fi_exporter) == [NATIVE_SCOPE] + + +def test_native_sampling_decision_is_not_overridden(instrument): + # A sampler that drops the SDK's span must not make the instrumentor fall back. + class DropNative(Sampler): + def should_sample(self, parent_context, trace_id, name, *args, **kwargs): + decision = Decision.DROP if name.startswith("anthropic.") else Decision.RECORD_AND_SAMPLE + return SamplingResult(decision) + + def get_description(self): + return "DropNative" + + fi, fi_exporter = make_provider(sampler=DropNative()) + try: + instrument(tracer_provider=fi) + call("create", False, open_telemetry={"tracer_provider": fi}) + assert scopes(fi_exporter) == [] + finally: + fi.shutdown() + + +def test_client_copies_keep_their_own_choice(providers, instrument, monkeypatch): + fi, fi_exporter, _, _ = providers + instrument(tracer_provider=fi) + monkeypatch.setenv("ANTHROPIC_OPEN_TELEMETRY", "false") + transport = httpx.MockTransport(respond) + with anthropic.Anthropic( + api_key="synthetic", base_url="http://localhost", http_client=httpx.Client(transport=transport) + ) as original: + traced = original.with_options(open_telemetry={"tracer_provider": fi}) + options_before = traced.open_telemetry + with fi.get_tracer("application").start_as_current_span("parent") as parent: + traced.messages.create(**PARAMS) + original.messages.create(**PARAMS) + assert traced.open_telemetry == options_before + assert original.open_telemetry["traces"]["enabled"] is False + spans = {span.instrumentation_scope.name: span for span in fi_exporter.get_finished_spans()} + assert sorted(spans) == ["application", NATIVE_SCOPE, "traceai_anthropic"] + for scope in (NATIVE_SCOPE, "traceai_anthropic"): + assert spans[scope].parent.span_id == parent.get_span_context().span_id + + +@pytest.mark.parametrize("asynchronous", [False, True]) +def test_native_partial_stream_is_returned_unwrapped(providers, instrument, asynchronous): + fi, fi_exporter, _, _ = providers + instrument(tracer_provider=fi) + options = dict(api_key="synthetic", base_url="http://localhost", open_telemetry={"tracer_provider": fi}) + transport = httpx.MockTransport(respond) + if asynchronous: + + async def run(): + async with anthropic.AsyncAnthropic(**options, http_client=httpx.AsyncClient(transport=transport)) as client: + async with await client.messages.create(**PARAMS, stream=True) as stream: + assert type(stream) is anthropic.AsyncStream + assert (await stream.__anext__()).type == "message_start" + assert not fi_exporter.get_finished_spans() + + asyncio.run(run()) + else: + with anthropic.Anthropic(**options, http_client=httpx.Client(transport=transport)) as client: + with client.messages.create(**PARAMS, stream=True) as stream: + assert type(stream) is anthropic.Stream + assert next(stream).type == "message_start" + assert not fi_exporter.get_finished_spans() + assert scopes(fi_exporter) == [NATIVE_SCOPE] + + +@pytest.mark.parametrize("asynchronous", [False, True]) +def test_native_error_is_recorded_once(providers, instrument, asynchronous): + fi, fi_exporter, _, _ = providers + instrument(tracer_provider=fi) + with pytest.raises(anthropic.BadRequestError): + call("create", asynchronous, handler=fail, open_telemetry={"tracer_provider": fi}) + spans = fi_exporter.get_finished_spans() + assert [span.instrumentation_scope.name for span in spans] == [NATIVE_SCOPE] + assert spans[0].status.status_code == trace.StatusCode.ERROR + + +def test_deferral_is_logged_once_per_instrumentor(providers, instrument, caplog): + fi, _, _, _ = providers + instrument(tracer_provider=fi) + with caplog.at_level(logging.WARNING, logger="traceai_anthropic"): + call("create", False, open_telemetry={"tracer_provider": fi}) + call("create", True, open_telemetry={"tracer_provider": fi}) + notices = [r for r in caplog.records if "native OpenTelemetry" in r.getMessage()] + assert len(notices) == 1 + assert "open_telemetry=False" in notices[0].getMessage() + + +def test_unreadable_client_options_keep_the_instrumentor(providers, instrument, monkeypatch): + fi, fi_exporter, _, _ = providers + instrument(tracer_provider=fi) + + def broken(self): + raise RuntimeError("synthetic") + + monkeypatch.setattr(type(anthropic.Anthropic(api_key="synthetic")), "open_telemetry", property(broken)) + # Losing the call is worse than a second span: an unreadable option keeps the instrumentor. + call("create", False, open_telemetry={"tracer_provider": fi}) + assert "traceai_anthropic" in scopes(fi_exporter) diff --git a/python/frameworks/anthropic/traceai_anthropic/__init__.py b/python/frameworks/anthropic/traceai_anthropic/__init__.py index e60118782..66c680ea0 100644 --- a/python/frameworks/anthropic/traceai_anthropic/__init__.py +++ b/python/frameworks/anthropic/traceai_anthropic/__init__.py @@ -8,6 +8,7 @@ from opentelemetry.instrumentation.instrumentor import ( BaseInstrumentor, ) # type: ignore[attr-defined] +from traceai_anthropic._native import _NativeTracing from traceai_anthropic._wrappers import ( _AsyncCompletionsWrapper, _AsyncMessagesWrapper, @@ -57,6 +58,9 @@ def _instrument(self, **kwargs: Any) -> None: logger.warning("ai-evaluation is not installed, please install it to trace protect") Protect = None + # The SDK's own spans (anthropic>=1.14) replace this instrumentor's Messages + # spans only when they reach the same provider; see _native.py. + native_tracing = _NativeTracing(kwargs.get("tracer_provider")) if not (tracer_provider := kwargs.get("tracer_provider")): tracer_provider = trace_api.get_tracer_provider() if not (config := kwargs.get("config")): @@ -93,21 +97,27 @@ def _instrument(self, **kwargs: Any) -> None: wrap_function_wrapper( module="anthropic.resources.messages", name="Messages.create", - wrapper=_MessagesWrapper(tracer=self._tracer), + wrapper=_MessagesWrapper( + tracer=self._tracer, native_tracing=native_tracing + ), ) self._original_async_messages_create = AsyncMessages.create wrap_function_wrapper( module="anthropic.resources.messages", name="AsyncMessages.create", - wrapper=_AsyncMessagesWrapper(tracer=self._tracer), + wrapper=_AsyncMessagesWrapper( + tracer=self._tracer, native_tracing=native_tracing + ), ) self._original_messages_count_tokens = Messages.count_tokens wrap_function_wrapper( module="anthropic.resources.messages", name="Messages.count_tokens", - wrapper=_MessagesCountTokensWrapper(tracer=self._tracer), + wrapper=_MessagesCountTokensWrapper( + tracer=self._tracer, native_tracing=native_tracing + ), ) if Protect is not None: diff --git a/python/frameworks/anthropic/traceai_anthropic/_native.py b/python/frameworks/anthropic/traceai_anthropic/_native.py new file mode 100644 index 000000000..c1d1a8875 --- /dev/null +++ b/python/frameworks/anthropic/traceai_anthropic/_native.py @@ -0,0 +1,68 @@ +"""Who traces a Messages call when the Anthropic SDK has its own OpenTelemetry spans. + +Anthropic SDK 1.14.0 and later give every API call a span of their own. When that +span goes to the tracer provider this instrumentor exports to, a second span from +the instrumentor would count the same call twice, so the instrumentor leaves the call +to the SDK. When the SDK's span goes elsewhere (another provider, or nowhere), the +instrumentor keeps tracing the call, so its provider never loses it. +""" + +import logging +from collections.abc import Mapping +from typing import Any, Optional + +from opentelemetry import trace as trace_api + +logger = logging.getLogger(__name__) + +_NOTICE = ( + "Anthropic SDK native OpenTelemetry tracing already sends Messages calls to the " + "tracer provider of AnthropicInstrumentor, so traceAI leaves those calls to the SDK " + "and emits no second span. TraceConfig and FI_HIDE_* do not apply to the SDK's spans: " + "the client's open_telemetry option controls their content (metadata only unless " + 'traces content_mode is "content"). To keep traceAI spans instead, create the ' + "client with open_telemetry=False." +) + + +class _NativeTracing: + """Decides, for each call, whether the SDK's own span reaches this instrumentor's provider.""" + + def __init__(self, tracer_provider: Optional[trace_api.TracerProvider]) -> None: + # Without an explicit provider, the instrumentor's tracer follows the global + # provider, as the SDK's does, so both are resolved at call time. + if isinstance(tracer_provider, trace_api.ProxyTracerProvider): + tracer_provider = None + self._tracer_provider = tracer_provider + self._notified = False + + def owns(self, resource: Any) -> bool: + """Whether the SDK traces this call of `resource` (a Messages resource) into our provider. + + Reads the client's resolved public `open_telemetry` option, never a span's + recording state: the SDK's sampling decision must not trigger a second span. + Older SDKs, and any option this cannot read, keep the instrumentor tracing. + """ + try: + options = getattr(getattr(resource, "_client", None), "open_telemetry", None) + if not isinstance(options, Mapping): + return False + traces = options.get("traces") + if not isinstance(traces, Mapping) or not traces.get("enabled", False): + return False + native_provider = options.get("tracer_provider") + if native_provider is None: + native_provider = trace_api.get_tracer_provider() + own_provider = self._tracer_provider + if own_provider is None: + own_provider = trace_api.get_tracer_provider() + owns = native_provider is own_provider and not isinstance( + native_provider, trace_api.ProxyTracerProvider + ) + except Exception: + logger.debug("Could not read the Anthropic client's OpenTelemetry option", exc_info=True) + return False + if owns and not self._notified: + self._notified = True + logger.warning(_NOTICE) + return owns diff --git a/python/frameworks/anthropic/traceai_anthropic/_wrappers.py b/python/frameworks/anthropic/traceai_anthropic/_wrappers.py index c21251a9f..89bf43918 100644 --- a/python/frameworks/anthropic/traceai_anthropic/_wrappers.py +++ b/python/frameworks/anthropic/traceai_anthropic/_wrappers.py @@ -32,6 +32,7 @@ ) from opentelemetry import trace as trace_api from opentelemetry.trace import INVALID_SPAN +from traceai_anthropic._native import _NativeTracing from traceai_anthropic._stream import _MessagesStream, _Stream from traceai_anthropic._utils import _finish_request_error from traceai_anthropic._with_span import _WithSpan @@ -48,10 +49,20 @@ class _WithTracer(ABC): Base class for wrappers that need a tracer. """ - def __init__(self, tracer: trace_api.Tracer, *args: Any, **kwargs: Any) -> None: + def __init__( + self, + tracer: trace_api.Tracer, + *args: Any, + native_tracing: Optional[_NativeTracing] = None, + **kwargs: Any, + ) -> None: super().__init__(*args, **kwargs) self._tracer = tracer + self._native_tracing = native_tracing + + def _native_tracing_owns(self, instance: Any) -> bool: + return self._native_tracing is not None and self._native_tracing.owns(instance) @contextmanager def _start_as_current_span( @@ -243,7 +254,9 @@ def __call__( args: Tuple[Any, ...], kwargs: Mapping[str, Any], ) -> Any: - if context_api.get_value(context_api._SUPPRESS_INSTRUMENTATION_KEY): + if context_api.get_value( + context_api._SUPPRESS_INSTRUMENTATION_KEY + ) or self._native_tracing_owns(instance): return wrapped(*args, **kwargs) arguments = kwargs @@ -335,7 +348,9 @@ async def __call__( args: Tuple[Any, ...], kwargs: Mapping[str, Any], ) -> Any: - if context_api.get_value(context_api._SUPPRESS_INSTRUMENTATION_KEY): + if context_api.get_value( + context_api._SUPPRESS_INSTRUMENTATION_KEY + ) or self._native_tracing_owns(instance): return await wrapped(*args, **kwargs) arguments = kwargs @@ -407,7 +422,9 @@ def __call__( args, kwargs, ): - if context_api.get_value(context_api._SUPPRESS_INSTRUMENTATION_KEY): + if context_api.get_value( + context_api._SUPPRESS_INSTRUMENTATION_KEY + ) or self._native_tracing_owns(instance): return wrapped(*args, **kwargs) arguments = kwargs From de9771b73c4d26c33fa54238b2e732244eab33e7 Mon Sep 17 00:00:00 2001 From: Nikhil Pareek Date: Sun, 1 Nov 2026 15:35:08 -0800 Subject: [PATCH 2/5] fix(anthropic): keep Future AGI context by owning shared-provider calls When the SDK's native span would reach the instrumentor's provider, the instrumentor deferred to it, and calls lost session.id, user.id, metadata, tag.tags, gen_ai.span.kind, input/output values and TraceConfig hiding. The instrumentor now keeps its span and runs that one call on a with_options(open_telemetry=False) copy of the client, so each call is still recorded once. The application's client is not changed. If the copy cannot be made, the instrumentor still traces the call and warns once. Co-Authored-By: Claude Opus 5.5 --- python/frameworks/anthropic/README.md | 4 +- .../anthropic/tests/test_native_otlp.py | 57 +++++--- .../anthropic/tests/test_native_ownership.py | 132 ++++++++++++++---- .../anthropic/traceai_anthropic/__init__.py | 5 +- .../anthropic/traceai_anthropic/_native.py | 76 +++++++--- .../anthropic/traceai_anthropic/_wrappers.py | 22 +-- 6 files changed, 212 insertions(+), 84 deletions(-) diff --git a/python/frameworks/anthropic/README.md b/python/frameworks/anthropic/README.md index 489107607..f6d96f145 100644 --- a/python/frameworks/anthropic/README.md +++ b/python/frameworks/anthropic/README.md @@ -54,9 +54,9 @@ Anthropic SDK 1.14.0 and later can trace Messages calls themselves. To send thos client = anthropic.Anthropic(open_telemetry={"tracer_provider": trace_provider}) ``` -The SDK's spans include model, timing and token usage, and include prompts and responses only with `"traces": {"content_mode": "content"}`. `TraceConfig` and `FI_HIDE_*` settings don't apply to them. +The SDK's spans include model, timing and token usage, and include prompts and responses only with `"traces": {"content_mode": "content"}`. `TraceConfig`, `FI_HIDE_*` settings and the `using_session` / `using_attributes` context (session, user, metadata, tags) don't apply to them. -When the SDK's spans reach the same tracer provider as `AnthropicInstrumentor` (because you pass it to the client, or because it is the global provider), the instrumentor leaves `messages.create` and `messages.count_tokens` calls to the SDK, so each call is recorded once. To keep this instrumentor's spans and privacy controls instead, create the client with `open_telemetry=False`. +When `AnthropicInstrumentor` is installed and the SDK's spans would reach the same tracer provider (because you pass it to the client, or because it is the global provider), the instrumentor records `messages.create` and `messages.count_tokens` calls itself and turns the SDK's span off for each of those calls, so each call is recorded once with the Future AGI context and privacy controls. Your client keeps its own `open_telemetry` option. To skip this per-call step, create the client with `open_telemetry=False`. `messages.stream()` is not wrapped by the instrumentor, so it keeps the SDK's span. ### Create Anthropic Components Set up your Anthropic client with built-in observability. diff --git a/python/frameworks/anthropic/tests/test_native_otlp.py b/python/frameworks/anthropic/tests/test_native_otlp.py index 5e157028c..5a1945976 100644 --- a/python/frameworks/anthropic/tests/test_native_otlp.py +++ b/python/frameworks/anthropic/tests/test_native_otlp.py @@ -1,8 +1,8 @@ -"""The SDK's native spans reach the Future AGI collector path with what Observe reads. +"""Each Messages call reaches the Future AGI collector path once, with what Observe reads. -Real path: Anthropic SDK native tracing -> fi_instrumentation register() provider -> -OTLP/HTTP exporter -> loopback harness Receiver. Synthetic transport and credentials; -nothing leaves 127.0.0.1. +Real path: Anthropic SDK native tracing (or AnthropicInstrumentor) -> fi_instrumentation +register() provider -> OTLP/HTTP exporter -> loopback harness Receiver. Synthetic +transport and credentials; nothing leaves 127.0.0.1. """ import inspect @@ -87,27 +87,22 @@ def send(provider, content_mode=None): assert client.messages.create(**PARAMS).content[0].text == ANSWER -@pytest.mark.parametrize("instrumented", [False, True]) -def test_native_span_reaches_the_collector_once_with_model_usage_and_content(collector, instrumented): - provider = register_provider() - instrumentor = None - if instrumented: - from traceai_anthropic import AnthropicInstrumentor +def assert_collector_auth(collector): + requests = collector.requests() + assert requests and {r["path"] for r in requests} == {"/tracer/v1/traces"} + assert requests[0]["headers"]["x-api-key"] == "synthetic-api-key" + assert requests[0]["headers"]["x-secret-key"] == "synthetic-secret-key" + - instrumentor = AnthropicInstrumentor() - instrumentor.instrument(tracer_provider=provider) +def test_native_span_reaches_the_collector_once_with_model_usage_and_content(collector): + provider = register_provider() try: send(provider, content_mode="content") provider.force_flush() finally: - if instrumentor is not None: - instrumentor.uninstrument() provider.shutdown() - requests = collector.requests() - assert requests and {r["path"] for r in requests} == {"/tracer/v1/traces"} - assert requests[0]["headers"]["x-api-key"] == "synthetic-api-key" - assert requests[0]["headers"]["x-secret-key"] == "synthetic-secret-key" + assert_collector_auth(collector) spans = collector.spans() assert [span["name"] for span in spans] == ["anthropic.messages.create"] values = attributes(spans[0]) @@ -122,6 +117,32 @@ def test_native_span_reaches_the_collector_once_with_model_usage_and_content(col assert ANSWER in json.loads(values["gen_ai.output.messages"])[0]["parts"][0]["content"] +def test_instrumented_client_reaches_the_collector_once_with_future_agi_context(collector): + from fi_instrumentation import using_attributes + from traceai_anthropic import AnthropicInstrumentor + + provider = register_provider() + instrumentor = AnthropicInstrumentor() + instrumentor.instrument(tracer_provider=provider) + try: + with using_attributes(session_id="sess-1", user_id="user-1"): + send(provider, content_mode="content") + provider.force_flush() + finally: + instrumentor.uninstrument() + provider.shutdown() + + assert_collector_auth(collector) + spans = collector.spans() + assert [span["name"] for span in spans] == ["Messages"] + values = attributes(spans[0]) + assert values["session.id"] == "sess-1" + assert values["user.id"] == "user-1" + assert values["gen_ai.span.kind"] == "LLM" + assert PROMPT in values["input.value"] + assert ANSWER in values["output.value"] + + def test_native_default_exports_metadata_without_content(collector): provider = register_provider() try: diff --git a/python/frameworks/anthropic/tests/test_native_ownership.py b/python/frameworks/anthropic/tests/test_native_ownership.py index 188d45798..9d58b80db 100644 --- a/python/frameworks/anthropic/tests/test_native_ownership.py +++ b/python/frameworks/anthropic/tests/test_native_ownership.py @@ -1,9 +1,10 @@ """One span per Messages call in each tracer provider, with SDKs that trace natively. Anthropic SDK 1.14.0 and later emit their own OpenTelemetry span for every API call. -When that span reaches the provider this instrumentor exports to, the instrumentor -leaves the call to the SDK. When it does not, the instrumentor keeps its own span, -so the instrumentor's provider never loses the call. Real SDK, synthetic transport. +When that span would reach the provider this instrumentor exports to, the instrumentor +keeps its own span (with the Future AGI context, input/output and TraceConfig) and +turns the SDK's span off for that call. When the SDK's span goes elsewhere, both are +left as they are, so each provider gets the call once. Real SDK, synthetic transport. """ import asyncio @@ -18,6 +19,7 @@ from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter from opentelemetry.sdk.trace.sampling import Decision, Sampler, SamplingResult +from fi_instrumentation import using_attributes from traceai_anthropic import AnthropicInstrumentor try: @@ -29,6 +31,7 @@ pytestmark = pytest.mark.skipif(not NATIVE, reason="SDK predates native OpenTelemetry") NATIVE_SCOPE = "com.anthropic.sdk.python" +FI_SCOPE = "traceai_anthropic" PARAMS = dict( model="local-test", max_tokens=16, @@ -178,13 +181,13 @@ async def run(): @pytest.mark.parametrize("operation,asynchronous", OPERATIONS) -def test_native_span_in_the_instrumentors_provider_is_the_only_span( +def test_client_passed_the_instrumentors_provider_gets_only_the_instrumentor_span( providers, instrument, operation, asynchronous ): fi, fi_exporter, _, _ = providers instrument(tracer_provider=fi) call(operation, asynchronous, open_telemetry={"tracer_provider": fi}) - assert scopes(fi_exporter) == [NATIVE_SCOPE] + assert scopes(fi_exporter) == [FI_SCOPE] @pytest.mark.parametrize("operation,asynchronous", OPERATIONS) @@ -198,7 +201,7 @@ def test_global_provider_shared_with_the_sdk_gets_one_span( global_provider(fi) instrument(**({"tracer_provider": fi} if explicit else {})) call(operation, asynchronous) - assert scopes(fi_exporter) == [NATIVE_SCOPE] + assert scopes(fi_exporter) == [FI_SCOPE] @pytest.mark.parametrize("operation,asynchronous", OPERATIONS) @@ -216,7 +219,7 @@ def test_instrumentors_provider_keeps_the_call_when_native_spans_go_elsewhere( call_options = {"open_telemetry": {"tracer_provider": other}} instrument(tracer_provider=fi) call(operation, asynchronous, **call_options) - assert scopes(fi_exporter) == ["traceai_anthropic"] + assert scopes(fi_exporter) == [FI_SCOPE] assert scopes(other_exporter) == [NATIVE_SCOPE] @@ -229,7 +232,7 @@ def test_non_global_provider_keeps_the_call_when_sdk_has_no_provider( assert isinstance(trace.get_tracer_provider(), trace.ProxyTracerProvider) instrument(tracer_provider=fi) call(operation, asynchronous) - assert scopes(fi_exporter) == ["traceai_anthropic"] + assert scopes(fi_exporter) == [FI_SCOPE] @pytest.mark.parametrize("operation,asynchronous", OPERATIONS) @@ -245,28 +248,30 @@ def test_disabled_native_tracing_keeps_the_instrumentor( call(operation, asynchronous) else: call(operation, asynchronous, open_telemetry=False) - assert scopes(fi_exporter) == ["traceai_anthropic"] + assert scopes(fi_exporter) == [FI_SCOPE] @pytest.mark.parametrize("asynchronous", [False, True]) def test_helper_stream_is_traced_once_by_the_sdk(providers, instrument, asynchronous): + # messages.stream() does not go through Messages.create, so the instrumentor + # never wraps it and the SDK's span is the only one; the client is not changed. fi, fi_exporter, _, _ = providers instrument(tracer_provider=fi) call("helper_stream", asynchronous, open_telemetry={"tracer_provider": fi}) assert scopes(fi_exporter) == [NATIVE_SCOPE] -def test_native_sampling_decision_is_not_overridden(instrument): - # A sampler that drops the SDK's span must not make the instrumentor fall back. - class DropNative(Sampler): +def test_sampling_out_the_instrumentor_span_does_not_bring_back_the_native_span(instrument): + # The SDK's span stays off whatever the sampler decides for the instrumentor's. + class DropInstrumentor(Sampler): def should_sample(self, parent_context, trace_id, name, *args, **kwargs): - decision = Decision.DROP if name.startswith("anthropic.") else Decision.RECORD_AND_SAMPLE + decision = Decision.DROP if name == "Messages" else Decision.RECORD_AND_SAMPLE return SamplingResult(decision) def get_description(self): - return "DropNative" + return "DropInstrumentor" - fi, fi_exporter = make_provider(sampler=DropNative()) + fi, fi_exporter = make_provider(sampler=DropInstrumentor()) try: instrument(tracer_provider=fi) call("create", False, open_telemetry={"tracer_provider": fi}) @@ -275,7 +280,7 @@ def get_description(self): fi.shutdown() -def test_client_copies_keep_their_own_choice(providers, instrument, monkeypatch): +def test_application_clients_keep_their_option_and_parent_context(providers, instrument, monkeypatch): fi, fi_exporter, _, _ = providers instrument(tracer_provider=fi) monkeypatch.setenv("ANTHROPIC_OPEN_TELEMETRY", "false") @@ -288,16 +293,19 @@ def test_client_copies_keep_their_own_choice(providers, instrument, monkeypatch) with fi.get_tracer("application").start_as_current_span("parent") as parent: traced.messages.create(**PARAMS) original.messages.create(**PARAMS) + # The per-call copy never changes the application's clients. assert traced.open_telemetry == options_before + assert traced.open_telemetry["traces"]["enabled"] is True assert original.open_telemetry["traces"]["enabled"] is False - spans = {span.instrumentation_scope.name: span for span in fi_exporter.get_finished_spans()} - assert sorted(spans) == ["application", NATIVE_SCOPE, "traceai_anthropic"] - for scope in (NATIVE_SCOPE, "traceai_anthropic"): - assert spans[scope].parent.span_id == parent.get_span_context().span_id + spans = fi_exporter.get_finished_spans() + assert sorted(span.instrumentation_scope.name for span in spans) == ["application", FI_SCOPE, FI_SCOPE] + for span in spans: + if span.instrumentation_scope.name == FI_SCOPE: + assert span.parent.span_id == parent.get_span_context().span_id @pytest.mark.parametrize("asynchronous", [False, True]) -def test_native_partial_stream_is_returned_unwrapped(providers, instrument, asynchronous): +def test_partial_stream_is_recorded_once(providers, instrument, asynchronous): fi, fi_exporter, _, _ = providers instrument(tracer_provider=fi) options = dict(api_key="synthetic", base_url="http://localhost", open_telemetry={"tracer_provider": fi}) @@ -307,39 +315,36 @@ def test_native_partial_stream_is_returned_unwrapped(providers, instrument, asyn async def run(): async with anthropic.AsyncAnthropic(**options, http_client=httpx.AsyncClient(transport=transport)) as client: async with await client.messages.create(**PARAMS, stream=True) as stream: - assert type(stream) is anthropic.AsyncStream assert (await stream.__anext__()).type == "message_start" - assert not fi_exporter.get_finished_spans() asyncio.run(run()) else: with anthropic.Anthropic(**options, http_client=httpx.Client(transport=transport)) as client: with client.messages.create(**PARAMS, stream=True) as stream: - assert type(stream) is anthropic.Stream assert next(stream).type == "message_start" - assert not fi_exporter.get_finished_spans() - assert scopes(fi_exporter) == [NATIVE_SCOPE] + assert scopes(fi_exporter) == [FI_SCOPE] @pytest.mark.parametrize("asynchronous", [False, True]) -def test_native_error_is_recorded_once(providers, instrument, asynchronous): +def test_error_is_recorded_once(providers, instrument, asynchronous): fi, fi_exporter, _, _ = providers instrument(tracer_provider=fi) with pytest.raises(anthropic.BadRequestError): call("create", asynchronous, handler=fail, open_telemetry={"tracer_provider": fi}) spans = fi_exporter.get_finished_spans() - assert [span.instrumentation_scope.name for span in spans] == [NATIVE_SCOPE] + assert [span.instrumentation_scope.name for span in spans] == [FI_SCOPE] assert spans[0].status.status_code == trace.StatusCode.ERROR -def test_deferral_is_logged_once_per_instrumentor(providers, instrument, caplog): +def test_turning_off_the_native_span_is_logged_once_per_instrumentor(providers, instrument, caplog): fi, _, _, _ = providers instrument(tracer_provider=fi) - with caplog.at_level(logging.WARNING, logger="traceai_anthropic"): + with caplog.at_level(logging.INFO, logger="traceai_anthropic"): call("create", False, open_telemetry={"tracer_provider": fi}) call("create", True, open_telemetry={"tracer_provider": fi}) notices = [r for r in caplog.records if "native OpenTelemetry" in r.getMessage()] assert len(notices) == 1 + assert notices[0].levelno == logging.INFO assert "open_telemetry=False" in notices[0].getMessage() @@ -353,4 +358,69 @@ def broken(self): monkeypatch.setattr(type(anthropic.Anthropic(api_key="synthetic")), "open_telemetry", property(broken)) # Losing the call is worse than a second span: an unreadable option keeps the instrumentor. call("create", False, open_telemetry={"tracer_provider": fi}) - assert "traceai_anthropic" in scopes(fi_exporter) + assert FI_SCOPE in scopes(fi_exporter) + + +@pytest.mark.parametrize("asynchronous", [False, True]) +def test_client_copy_failure_keeps_the_instrumentor_and_warns_once( + providers, instrument, monkeypatch, caplog, asynchronous +): + fi, fi_exporter, _, _ = providers + instrument(tracer_provider=fi) + + def broken(self, **kwargs): + raise RuntimeError("synthetic") + + client_class = anthropic.AsyncAnthropic if asynchronous else anthropic.Anthropic + monkeypatch.setattr(client_class, "with_options", broken) + with caplog.at_level(logging.WARNING, logger="traceai_anthropic"): + call("create", asynchronous, open_telemetry={"tracer_provider": fi}) + call("create", asynchronous, open_telemetry={"tracer_provider": fi}) + # The instrumentor's span is never lost; the SDK's span may then be a second one. + assert scopes(fi_exporter).count(FI_SCOPE) == 2 + warnings = [r for r in caplog.records if "recorded twice" in r.getMessage()] + assert len(warnings) == 1 + + +CONTEXT = dict(session_id="sess-1", user_id="user-1", metadata={"team": "synthetic"}, tags=["t1"]) + + +@pytest.mark.parametrize("operation,asynchronous", OPERATIONS) +@pytest.mark.parametrize("native_destination", ["global", "client"]) +def test_future_agi_context_reaches_the_one_span( + providers, global_provider, instrument, operation, asynchronous, native_destination +): + # register(set_global_tracer_provider=True), or the provider passed to the client, + # with using_attributes: the single span must still carry session, user, metadata, + # tags, span kind and input. + fi, fi_exporter, _, _ = providers + if native_destination == "global": + global_provider(fi) + call_options = {} + else: + call_options = {"open_telemetry": {"tracer_provider": fi}} + instrument(tracer_provider=fi) + with using_attributes(**CONTEXT): + call(operation, asynchronous, **call_options) + spans = fi_exporter.get_finished_spans() + assert [span.instrumentation_scope.name for span in spans] == [FI_SCOPE] + attributes = dict(spans[0].attributes) + assert attributes["session.id"] == "sess-1" + assert attributes["user.id"] == "user-1" + assert json.loads(attributes["metadata"]) == {"team": "synthetic"} + assert list(attributes["tag.tags"]) == ["t1"] + assert attributes["gen_ai.span.kind"] == "LLM" + assert "input.value" in attributes + + +def test_trace_config_applies_to_the_one_span(providers, global_provider, monkeypatch, instrument): + fi, fi_exporter, _, _ = providers + global_provider(fi) + monkeypatch.setenv("FI_HIDE_INPUTS", "true") + monkeypatch.setenv("FI_HIDE_OUTPUTS", "true") + instrument(tracer_provider=fi) + call("create", False) + spans = fi_exporter.get_finished_spans() + assert [span.instrumentation_scope.name for span in spans] == [FI_SCOPE] + payload = json.dumps(dict(spans[0].attributes), default=str) + assert "synthetic prompt" not in payload and "synthetic answer" not in payload diff --git a/python/frameworks/anthropic/traceai_anthropic/__init__.py b/python/frameworks/anthropic/traceai_anthropic/__init__.py index 66c680ea0..c931bf29b 100644 --- a/python/frameworks/anthropic/traceai_anthropic/__init__.py +++ b/python/frameworks/anthropic/traceai_anthropic/__init__.py @@ -58,8 +58,9 @@ def _instrument(self, **kwargs: Any) -> None: logger.warning("ai-evaluation is not installed, please install it to trace protect") Protect = None - # The SDK's own spans (anthropic>=1.14) replace this instrumentor's Messages - # spans only when they reach the same provider; see _native.py. + # When the SDK's own spans (anthropic>=1.14) would reach the same provider, + # the instrumentor keeps its Messages spans and turns the SDK's off per call; + # see _native.py. native_tracing = _NativeTracing(kwargs.get("tracer_provider")) if not (tracer_provider := kwargs.get("tracer_provider")): tracer_provider = trace_api.get_tracer_provider() diff --git a/python/frameworks/anthropic/traceai_anthropic/_native.py b/python/frameworks/anthropic/traceai_anthropic/_native.py index c1d1a8875..23df27d02 100644 --- a/python/frameworks/anthropic/traceai_anthropic/_native.py +++ b/python/frameworks/anthropic/traceai_anthropic/_native.py @@ -1,32 +1,42 @@ """Who traces a Messages call when the Anthropic SDK has its own OpenTelemetry spans. Anthropic SDK 1.14.0 and later give every API call a span of their own. When that -span goes to the tracer provider this instrumentor exports to, a second span from -the instrumentor would count the same call twice, so the instrumentor leaves the call -to the SDK. When the SDK's span goes elsewhere (another provider, or nowhere), the -instrumentor keeps tracing the call, so its provider never loses it. +span would go to the tracer provider this instrumentor exports to, the call would be +recorded twice. The instrumentor's span is the one that carries what Future AGI reads: +the context of `using_session` / `using_attributes` (session, user, metadata, tags, +prompt template), the span kind, input and output values, and the `TraceConfig` / +`FI_HIDE_*` controls. So the instrumentor keeps its span and runs that one call on a +copy of the client whose native tracing is off (`with_options(open_telemetry=False)`, +the SDK's public per-request option). The application's client is never changed. + +When the SDK's span goes elsewhere (another provider, or nowhere), both are left as +they are, so each provider gets the call once. """ import logging from collections.abc import Mapping -from typing import Any, Optional +from typing import Any, Callable, Optional from opentelemetry import trace as trace_api logger = logging.getLogger(__name__) _NOTICE = ( - "Anthropic SDK native OpenTelemetry tracing already sends Messages calls to the " - "tracer provider of AnthropicInstrumentor, so traceAI leaves those calls to the SDK " - "and emits no second span. TraceConfig and FI_HIDE_* do not apply to the SDK's spans: " - "the client's open_telemetry option controls their content (metadata only unless " - 'traces content_mode is "content"). To keep traceAI spans instead, create the ' - "client with open_telemetry=False." + "Anthropic SDK native OpenTelemetry tracing sends Messages calls to the tracer " + "provider of AnthropicInstrumentor. traceAI records those calls itself and turns " + "the SDK's span off for each of them, so every call is recorded once. To skip this " + "per-call step, create the client with open_telemetry=False." +) + +_COPY_FAILED = ( + "Could not turn off Anthropic SDK native tracing for a Messages call, so the call " + "may be recorded twice in the same tracer provider. Create the client with " + "open_telemetry=False to keep only traceAI spans." ) class _NativeTracing: - """Decides, for each call, whether the SDK's own span reaches this instrumentor's provider.""" + """Keeps each Messages call to one span in this instrumentor's provider.""" def __init__(self, tracer_provider: Optional[trace_api.TracerProvider]) -> None: # Without an explicit provider, the instrumentor's tracer follows the global @@ -35,13 +45,13 @@ def __init__(self, tracer_provider: Optional[trace_api.TracerProvider]) -> None: tracer_provider = None self._tracer_provider = tracer_provider self._notified = False + self._copy_failure_notified = False - def owns(self, resource: Any) -> bool: + def shares_provider(self, resource: Any) -> bool: """Whether the SDK traces this call of `resource` (a Messages resource) into our provider. - Reads the client's resolved public `open_telemetry` option, never a span's - recording state: the SDK's sampling decision must not trigger a second span. - Older SDKs, and any option this cannot read, keep the instrumentor tracing. + Reads the client's resolved public `open_telemetry` option. Older SDKs, and any + option this cannot read, count as not shared: the call is left as it is. """ try: options = getattr(getattr(resource, "_client", None), "open_telemetry", None) @@ -56,13 +66,39 @@ def owns(self, resource: Any) -> bool: own_provider = self._tracer_provider if own_provider is None: own_provider = trace_api.get_tracer_provider() - owns = native_provider is own_provider and not isinstance( + return native_provider is own_provider and not isinstance( native_provider, trace_api.ProxyTracerProvider ) except Exception: logger.debug("Could not read the Anthropic client's OpenTelemetry option", exc_info=True) return False - if owns and not self._notified: + + def without_native_span(self, wrapped: Callable[..., Any], instance: Any) -> Callable[..., Any]: + """The original method to call for this call, with the SDK's own span off when it would duplicate ours. + + `wrapped` is the original method bound to `instance`. When the SDK's span would + reach our provider, the same method is bound to a resource of a client copy + with native tracing disabled. If that copy can't be made, `wrapped` is returned + unchanged: a second span is better than a call without the instrumentor's span. + """ + if not self.shares_provider(instance): + return wrapped + try: + function = getattr(wrapped, "__func__", None) + if function is None or getattr(wrapped, "__self__", None) is not instance: + raise TypeError("the original method is not bound to the resource") + client = instance._client.with_options(open_telemetry=False) + options = client.open_telemetry + if not isinstance(options, Mapping) or options["traces"]["enabled"]: + raise ValueError("the client copy still has native tracing") + resource = type(instance)(client) + except Exception: + logger.debug("Could not disable Anthropic SDK native tracing for one call", exc_info=True) + if not self._copy_failure_notified: + self._copy_failure_notified = True + logger.warning(_COPY_FAILED) + return wrapped + if not self._notified: self._notified = True - logger.warning(_NOTICE) - return owns + logger.info(_NOTICE) + return function.__get__(resource, type(resource)) diff --git a/python/frameworks/anthropic/traceai_anthropic/_wrappers.py b/python/frameworks/anthropic/traceai_anthropic/_wrappers.py index 89bf43918..e70437ff8 100644 --- a/python/frameworks/anthropic/traceai_anthropic/_wrappers.py +++ b/python/frameworks/anthropic/traceai_anthropic/_wrappers.py @@ -61,8 +61,11 @@ def __init__( self._tracer = tracer self._native_tracing = native_tracing - def _native_tracing_owns(self, instance: Any) -> bool: - return self._native_tracing is not None and self._native_tracing.owns(instance) + def _without_native_span(self, wrapped: Callable[..., Any], instance: Any) -> Callable[..., Any]: + """The original method for this call, with the SDK's duplicate native span turned off.""" + if self._native_tracing is None: + return wrapped + return self._native_tracing.without_native_span(wrapped, instance) @contextmanager def _start_as_current_span( @@ -254,10 +257,9 @@ def __call__( args: Tuple[Any, ...], kwargs: Mapping[str, Any], ) -> Any: - if context_api.get_value( - context_api._SUPPRESS_INSTRUMENTATION_KEY - ) or self._native_tracing_owns(instance): + if context_api.get_value(context_api._SUPPRESS_INSTRUMENTATION_KEY): return wrapped(*args, **kwargs) + wrapped = self._without_native_span(wrapped, instance) arguments = kwargs llm_input_messages = dict(arguments).pop("messages", None) @@ -348,10 +350,9 @@ async def __call__( args: Tuple[Any, ...], kwargs: Mapping[str, Any], ) -> Any: - if context_api.get_value( - context_api._SUPPRESS_INSTRUMENTATION_KEY - ) or self._native_tracing_owns(instance): + if context_api.get_value(context_api._SUPPRESS_INSTRUMENTATION_KEY): return await wrapped(*args, **kwargs) + wrapped = self._without_native_span(wrapped, instance) arguments = kwargs llm_prompt = dict(arguments).pop("system", None) @@ -422,10 +423,9 @@ def __call__( args, kwargs, ): - if context_api.get_value( - context_api._SUPPRESS_INSTRUMENTATION_KEY - ) or self._native_tracing_owns(instance): + if context_api.get_value(context_api._SUPPRESS_INSTRUMENTATION_KEY): return wrapped(*args, **kwargs) + wrapped = self._without_native_span(wrapped, instance) arguments = kwargs llm_input_messages = arguments.get("messages", None) From 4b886dd4c7f10cd2f9748bcfc798ddc83dcff94a Mon Sep 17 00:00:00 2001 From: Nikhil Pareek Date: Sun, 1 Nov 2026 15:40:03 -0800 Subject: [PATCH 3/5] test(anthropic): check the per-call client copy sends the same request Compares method, URL, headers and body sent by a client given the instrumentor's provider, with and without the instrumentor, sync and async. Notes in the README that async count_tokens keeps the SDK's span. Co-Authored-By: Claude Opus 5.5 --- python/frameworks/anthropic/README.md | 2 +- .../anthropic/tests/test_native_ownership.py | 52 +++++++++++++++++++ 2 files changed, 53 insertions(+), 1 deletion(-) diff --git a/python/frameworks/anthropic/README.md b/python/frameworks/anthropic/README.md index f6d96f145..3b30a34ec 100644 --- a/python/frameworks/anthropic/README.md +++ b/python/frameworks/anthropic/README.md @@ -56,7 +56,7 @@ client = anthropic.Anthropic(open_telemetry={"tracer_provider": trace_provider}) The SDK's spans include model, timing and token usage, and include prompts and responses only with `"traces": {"content_mode": "content"}`. `TraceConfig`, `FI_HIDE_*` settings and the `using_session` / `using_attributes` context (session, user, metadata, tags) don't apply to them. -When `AnthropicInstrumentor` is installed and the SDK's spans would reach the same tracer provider (because you pass it to the client, or because it is the global provider), the instrumentor records `messages.create` and `messages.count_tokens` calls itself and turns the SDK's span off for each of those calls, so each call is recorded once with the Future AGI context and privacy controls. Your client keeps its own `open_telemetry` option. To skip this per-call step, create the client with `open_telemetry=False`. `messages.stream()` is not wrapped by the instrumentor, so it keeps the SDK's span. +When `AnthropicInstrumentor` is installed and the SDK's spans would reach the same tracer provider (because you pass it to the client, or because it is the global provider), the instrumentor records `messages.create` (sync and async) and sync `messages.count_tokens` calls itself and turns the SDK's span off for each of those calls, so each call is recorded once with the Future AGI context and privacy controls. Your client keeps its own `open_telemetry` option. To skip this per-call step, create the client with `open_telemetry=False`. `messages.stream()` and async `messages.count_tokens` are not wrapped by the instrumentor, so they keep the SDK's span. ### Create Anthropic Components Set up your Anthropic client with built-in observability. diff --git a/python/frameworks/anthropic/tests/test_native_ownership.py b/python/frameworks/anthropic/tests/test_native_ownership.py index 9d58b80db..628074a73 100644 --- a/python/frameworks/anthropic/tests/test_native_ownership.py +++ b/python/frameworks/anthropic/tests/test_native_ownership.py @@ -424,3 +424,55 @@ def test_trace_config_applies_to_the_one_span(providers, global_provider, monkey assert [span.instrumentation_scope.name for span in spans] == [FI_SCOPE] payload = json.dumps(dict(spans[0].attributes), default=str) assert "synthetic prompt" not in payload and "synthetic answer" not in payload + + +def sent_requests(asynchronous, instrumented, instrument, provider): + """The requests a client given `provider` sends, with or without the instrumentor.""" + seen = [] + + def capture(request): + headers = {k: v for k, v in request.headers.items() if k not in ("traceparent", "tracestate")} + seen.append((request.method, str(request.url), headers, request.content)) + return respond(request) + + instrumentor = instrument(tracer_provider=provider) if instrumented else None + try: + options = dict( + api_key="sk-ant-synthetic", + base_url="http://localhost", + max_retries=0, + default_headers={"x-custom": "kept"}, + open_telemetry={"tracer_provider": provider}, + ) + extra = dict(extra_headers={"x-extra": "kept"}, metadata={"user_id": "synthetic"}) + if asynchronous: + + async def run(): + transport = httpx.MockTransport(capture) + async with anthropic.AsyncAnthropic(**options, http_client=httpx.AsyncClient(transport=transport)) as client: + await client.with_options(timeout=5).messages.create(**PARAMS, **extra) + + asyncio.run(run()) + else: + with anthropic.Anthropic(**options, http_client=httpx.Client(transport=httpx.MockTransport(capture))) as client: + client.with_options(timeout=5).messages.create(**PARAMS, **extra) + client.messages.count_tokens(model=PARAMS["model"], messages=PARAMS["messages"]) + finally: + if instrumentor is not None: + instrumentor.uninstrument() + for _, _, headers, _ in seen: + # Generated per request; everything else must match. + headers.pop("x-stainless-retry-count", None) + headers.pop("idempotency-key", None) + return seen + + +@pytest.mark.parametrize("asynchronous", [False, True]) +def test_the_client_copy_sends_the_same_request(providers, instrument, asynchronous): + fi, fi_exporter, _, _ = providers + untraced = sent_requests(asynchronous, False, instrument, fi) + traced = sent_requests(asynchronous, True, instrument, fi) + assert traced == untraced + assert all(headers.get("x-custom") == "kept" for _, _, headers, _ in traced) + assert traced[0][2].get("x-extra") == "kept" + assert any("sk-ant-synthetic" in value for value in traced[0][2].values()) From da208081bfef326bcc225c75a40cb802a971fd6a Mon Sep 17 00:00:00 2001 From: Nikhil Pareek Date: Sun, 1 Nov 2026 16:09:37 -0800 Subject: [PATCH 4/5] fix(anthropic): turn off the SDK span without copying the client The previous revision ran each shared-provider Messages call on a per-call client copy with open_telemetry=False and rebound the SDK function to it. That made AnthropicVertex load and refresh Google credentials on every call (the copy, not the application's client, received them) and skipped any wrapt patch installed on Messages.create before the instrumentor. Now the original method runs unchanged on the application's client, through every other wrapper. While it runs, the SDK's span start (anthropic._base_client.start_api_call_span, which returns None for "no span") returns None for that client's tracer only, so calls on other clients inside it keep their spans. The hook is installed by instrument() and removed by uninstrument(); an SDK without it keeps the instrumentor span, may record a second span, and warns once. Co-Authored-By: Claude Opus 5.5 --- .../anthropic/tests/test_native_ownership.py | 160 ++++++++++++++++-- .../anthropic/traceai_anthropic/__init__.py | 8 + .../anthropic/traceai_anthropic/_native.py | 137 +++++++++++---- .../anthropic/traceai_anthropic/_wrappers.py | 8 +- 4 files changed, 269 insertions(+), 44 deletions(-) diff --git a/python/frameworks/anthropic/tests/test_native_ownership.py b/python/frameworks/anthropic/tests/test_native_ownership.py index 628074a73..4f05ccebb 100644 --- a/python/frameworks/anthropic/tests/test_native_ownership.py +++ b/python/frameworks/anthropic/tests/test_native_ownership.py @@ -293,7 +293,7 @@ def test_application_clients_keep_their_option_and_parent_context(providers, ins with fi.get_tracer("application").start_as_current_span("parent") as parent: traced.messages.create(**PARAMS) original.messages.create(**PARAMS) - # The per-call copy never changes the application's clients. + # Turning the SDK's span off never changes the application's clients. assert traced.open_telemetry == options_before assert traced.open_telemetry["traces"]["enabled"] is True assert original.open_telemetry["traces"]["enabled"] is False @@ -362,26 +362,38 @@ def broken(self): @pytest.mark.parametrize("asynchronous", [False, True]) -def test_client_copy_failure_keeps_the_instrumentor_and_warns_once( +def test_sdk_without_the_span_hook_keeps_the_instrumentor_and_warns_once( providers, instrument, monkeypatch, caplog, asynchronous ): + # An SDK that traces natively but starts its spans some other way: the call is + # left as it is, so the instrumentor's span is kept and the SDK's may be a second. + from traceai_anthropic import _native + + monkeypatch.setattr(_native, "_SPAN_START_NAME", "synthetic_missing_hook") fi, fi_exporter, _, _ = providers instrument(tracer_provider=fi) - - def broken(self, **kwargs): - raise RuntimeError("synthetic") - - client_class = anthropic.AsyncAnthropic if asynchronous else anthropic.Anthropic - monkeypatch.setattr(client_class, "with_options", broken) with caplog.at_level(logging.WARNING, logger="traceai_anthropic"): call("create", asynchronous, open_telemetry={"tracer_provider": fi}) call("create", asynchronous, open_telemetry={"tracer_provider": fi}) - # The instrumentor's span is never lost; the SDK's span may then be a second one. assert scopes(fi_exporter).count(FI_SCOPE) == 2 + assert scopes(fi_exporter).count(NATIVE_SCOPE) == 2 warnings = [r for r in caplog.records if "recorded twice" in r.getMessage()] assert len(warnings) == 1 +def test_uninstrument_restores_the_sdk_span_start(providers, instrument): + import anthropic._base_client as base_client + import anthropic._tracing as tracing + + fi, fi_exporter, _, _ = providers + instrumentor = instrument(tracer_provider=fi) + assert base_client.start_api_call_span is not tracing.start_api_call_span + instrumentor.uninstrument() + assert base_client.start_api_call_span is tracing.start_api_call_span + call("create", False, open_telemetry={"tracer_provider": fi}) + assert scopes(fi_exporter) == [NATIVE_SCOPE] + + CONTEXT = dict(session_id="sess-1", user_id="user-1", metadata={"team": "synthetic"}, tags=["t1"]) @@ -468,7 +480,7 @@ async def run(): @pytest.mark.parametrize("asynchronous", [False, True]) -def test_the_client_copy_sends_the_same_request(providers, instrument, asynchronous): +def test_the_instrumented_call_sends_the_same_request(providers, instrument, asynchronous): fi, fi_exporter, _, _ = providers untraced = sent_requests(asynchronous, False, instrument, fi) traced = sent_requests(asynchronous, True, instrument, fi) @@ -476,3 +488,131 @@ def test_the_client_copy_sends_the_same_request(providers, instrument, asynchron assert all(headers.get("x-custom") == "kept" for _, _, headers, _ in traced) assert traced[0][2].get("x-extra") == "kept" assert any("sk-ant-synthetic" in value for value in traced[0][2].values()) + + +class _GoogleCredentials: + token = None + expired = False + + +@pytest.mark.parametrize("asynchronous", [False, True]) +def test_vertex_client_loads_google_credentials_once( + providers, global_provider, instrument, monkeypatch, asynchronous +): + # AnthropicVertex loads and refreshes Google credentials lazily, into the client + # that sends the request. Turning the SDK's span off must not send the call through + # another client, or every call would load and refresh them again. + from anthropic.lib.vertex import _client as vertex + + loads, refreshes = [], [] + + def load_auth(project_id=None): + loads.append(project_id) + return _GoogleCredentials(), "synthetic-project" + + def refresh_auth(credentials): + refreshes.append(credentials) + credentials.token = "synthetic-token" + + monkeypatch.setattr(vertex, "load_auth", load_auth) + monkeypatch.setattr(vertex, "refresh_auth", refresh_auth) + fi, fi_exporter, _, _ = providers + global_provider(fi) + instrument(tracer_provider=fi) + seen = [] + + def capture(request): + seen.append(request.headers["authorization"]) + return respond(request) + + options = dict(region="us-east5", project_id="synthetic-project", max_retries=0) + if asynchronous: + + async def run(): + transport = httpx.MockTransport(capture) + client = anthropic.AsyncAnthropicVertex(**options, http_client=httpx.AsyncClient(transport=transport)) + for _ in range(3): + await client.messages.create(**PARAMS) + return client + + client = asyncio.run(run()) + else: + client = anthropic.AnthropicVertex(**options, http_client=httpx.Client(transport=httpx.MockTransport(capture))) + for _ in range(3): + client.messages.create(**PARAMS) + assert (len(loads), len(refreshes)) == (1, 1) + assert client.credentials is not None + assert seen == ["Bearer synthetic-token"] * 3 + assert scopes(fi_exporter) == [FI_SCOPE] * 3 + + +WRAPPED_METHODS = { + ("create", False): ("Messages", "create"), + ("create", True): ("AsyncMessages", "create"), + ("raw_stream", False): ("Messages", "create"), + ("raw_stream", True): ("AsyncMessages", "create"), + ("count_tokens", False): ("Messages", "count_tokens"), +} + + +@pytest.mark.parametrize("operation,asynchronous", OPERATIONS) +@pytest.mark.parametrize("shared", [True, False]) +def test_wrapt_patch_installed_before_the_instrumentor_still_runs( + providers, global_provider, monkeypatch, instrument, operation, asynchronous, shared +): + # Other instrumentations, APM agents and guardrails wrap Messages.create with wrapt + # too. One installed before this instrumentor must run for every call. + import wrapt + from anthropic.resources import messages as messages_module + + class_name, method = WRAPPED_METHODS[(operation, asynchronous)] + owner = getattr(messages_module, class_name) + seen = [] + + def other(wrapped, instance, args, kwargs): + seen.append(type(instance).__name__) + return wrapped(*args, **kwargs) + + monkeypatch.setattr(owner, method, wrapt.FunctionWrapper(vars(owner)[method], other)) + fi, fi_exporter, _, _ = providers + if shared: + global_provider(fi) + instrument(tracer_provider=fi) + call(operation, asynchronous) + assert seen == [class_name] + assert scopes(fi_exporter) == [FI_SCOPE] + + +def test_call_made_inside_an_owned_call_keeps_its_own_native_span( + providers, global_provider, monkeypatch, instrument +): + # A wrapper that runs inside the instrumented call and makes its own call on a + # client whose SDK spans go to another provider: that call keeps its native span. + import wrapt + from anthropic.resources.messages import Messages + + fi, fi_exporter, other_provider, other_exporter = providers + + def list_models(request): + return httpx.Response(200, json={"data": [], "has_more": False, "first_id": None, "last_id": None}) + + inner = anthropic.Anthropic( + api_key="synthetic", + base_url="http://localhost", + max_retries=0, + open_telemetry={"tracer_provider": other_provider}, + http_client=httpx.Client(transport=httpx.MockTransport(list_models)), + ) + + def guard(wrapped, instance, args, kwargs): + if instance._client is not inner: + inner.models.list() + return wrapped(*args, **kwargs) + + monkeypatch.setattr(Messages, "create", wrapt.FunctionWrapper(vars(Messages)["create"], guard)) + global_provider(fi) + instrument(tracer_provider=fi) + call("create", False) + inner.close() + assert scopes(fi_exporter) == [FI_SCOPE] + assert scopes(other_exporter) == [NATIVE_SCOPE] diff --git a/python/frameworks/anthropic/traceai_anthropic/__init__.py b/python/frameworks/anthropic/traceai_anthropic/__init__.py index c931bf29b..bf81c28aa 100644 --- a/python/frameworks/anthropic/traceai_anthropic/__init__.py +++ b/python/frameworks/anthropic/traceai_anthropic/__init__.py @@ -34,6 +34,7 @@ class AnthropicInstrumentor(BaseInstrumentor): # type: ignore[misc] "_original_messages_count_tokens", "_original_async_messages_create", "_original_protect", + "_native_tracing", "_instruments", "_tracer", ) @@ -62,6 +63,8 @@ def _instrument(self, **kwargs: Any) -> None: # the instrumentor keeps its Messages spans and turns the SDK's off per call; # see _native.py. native_tracing = _NativeTracing(kwargs.get("tracer_provider")) + native_tracing.install() + self._native_tracing = native_tracing if not (tracer_provider := kwargs.get("tracer_provider")): tracer_provider = trace_api.get_tracer_provider() if not (config := kwargs.get("config")): @@ -161,3 +164,8 @@ def _uninstrument(self, **kwargs: Any) -> None: if self._original_protect is not None: Protect.protect = self._original_protect + + native_tracing = getattr(self, "_native_tracing", None) + if native_tracing is not None: + native_tracing.uninstall() + self._native_tracing = None diff --git a/python/frameworks/anthropic/traceai_anthropic/_native.py b/python/frameworks/anthropic/traceai_anthropic/_native.py index 23df27d02..998e778c1 100644 --- a/python/frameworks/anthropic/traceai_anthropic/_native.py +++ b/python/frameworks/anthropic/traceai_anthropic/_native.py @@ -5,19 +5,27 @@ recorded twice. The instrumentor's span is the one that carries what Future AGI reads: the context of `using_session` / `using_attributes` (session, user, metadata, tags, prompt template), the span kind, input and output values, and the `TraceConfig` / -`FI_HIDE_*` controls. So the instrumentor keeps its span and runs that one call on a -copy of the client whose native tracing is off (`with_options(open_telemetry=False)`, -the SDK's public per-request option). The application's client is never changed. +`FI_HIDE_*` controls. So the instrumentor keeps its span and the SDK skips its own for +that one call. + +The call itself is not changed: the original method runs on the application's client, +through any other wrappers on it, so client state such as lazily loaded cloud +credentials stays on that client. While it runs, the SDK's span start +(`start_api_call_span`, which may return `None` for "no span") returns `None` for this +client's tracer only; other clients' calls inside it keep their spans. When the SDK's span goes elsewhere (another provider, or nowhere), both are left as they are, so each provider gets the call once. """ +import importlib import logging from collections.abc import Mapping +from contextvars import ContextVar from typing import Any, Callable, Optional from opentelemetry import trace as trace_api +from wrapt import FunctionWrapper logger = logging.getLogger(__name__) @@ -28,12 +36,40 @@ "per-call step, create the client with open_telemetry=False." ) -_COPY_FAILED = ( - "Could not turn off Anthropic SDK native tracing for a Messages call, so the call " - "may be recorded twice in the same tracer provider. Create the client with " - "open_telemetry=False to keep only traceAI spans." +_NO_HOOK = ( + "Could not turn off Anthropic SDK native tracing for Messages calls with this " + "anthropic version, so they may be recorded twice in the same tracer provider. " + "Create the client with open_telemetry=False to keep only traceAI spans." ) +# Where the SDK starts the span of each API call (anthropic>=1.14), sync and async. +_SPAN_START_MODULE = "anthropic._base_client" +_SPAN_START_NAME = "start_api_call_span" + +# The client whose SDK span the running Messages call turns off, if any. +_owned_client: ContextVar[Any] = ContextVar("traceai_anthropic_owned_client", default=None) + + +def _skip_owned_span(wrapped: Callable[..., Any], instance: Any, args: Any, kwargs: Any) -> Any: + """`start_api_call_span`, returning `None` (no span) for the owned client's calls.""" + client = _owned_client.get() + if client is not None: + try: + tracer = args[0] if args else kwargs.get("tracer") + if tracer is not None and tracer is client._tracer: + return None + except Exception: + logger.debug("Could not match the Anthropic SDK span to the traced call", exc_info=True) + return wrapped(*args, **kwargs) + + +def _span_start_module() -> Optional[Any]: + try: + module = importlib.import_module(_SPAN_START_MODULE) + except Exception: + return None + return module if callable(vars(module).get(_SPAN_START_NAME)) else None + class _NativeTracing: """Keeps each Messages call to one span in this instrumentor's provider.""" @@ -44,8 +80,28 @@ def __init__(self, tracer_provider: Optional[trace_api.TracerProvider]) -> None: if isinstance(tracer_provider, trace_api.ProxyTracerProvider): tracer_provider = None self._tracer_provider = tracer_provider + self._hook: Optional[FunctionWrapper] = None + self._original: Any = None self._notified = False - self._copy_failure_notified = False + self._no_hook_notified = False + + def install(self) -> None: + """Hooks the SDK's span start. SDKs without one (before 1.14) are left alone.""" + module = _span_start_module() + if module is None or self._hook is not None: + return + self._original = vars(module)[_SPAN_START_NAME] + self._hook = FunctionWrapper(self._original, _skip_owned_span) + setattr(module, _SPAN_START_NAME, self._hook) + + def uninstall(self) -> None: + if self._hook is None: + return + module = _span_start_module() + if module is not None and vars(module).get(_SPAN_START_NAME) is self._hook: + setattr(module, _SPAN_START_NAME, self._original) + self._hook = None + self._original = None def shares_provider(self, resource: Any) -> bool: """Whether the SDK traces this call of `resource` (a Messages resource) into our provider. @@ -73,32 +129,47 @@ def shares_provider(self, resource: Any) -> bool: logger.debug("Could not read the Anthropic client's OpenTelemetry option", exc_info=True) return False - def without_native_span(self, wrapped: Callable[..., Any], instance: Any) -> Callable[..., Any]: - """The original method to call for this call, with the SDK's own span off when it would duplicate ours. - - `wrapped` is the original method bound to `instance`. When the SDK's span would - reach our provider, the same method is bound to a resource of a client copy - with native tracing disabled. If that copy can't be made, `wrapped` is returned - unchanged: a second span is better than a call without the instrumentor's span. - """ + def _client_to_own(self, instance: Any) -> Any: + """The client whose SDK span this call turns off, or `None` to leave the call as it is.""" if not self.shares_provider(instance): - return wrapped - try: - function = getattr(wrapped, "__func__", None) - if function is None or getattr(wrapped, "__self__", None) is not instance: - raise TypeError("the original method is not bound to the resource") - client = instance._client.with_options(open_telemetry=False) - options = client.open_telemetry - if not isinstance(options, Mapping) or options["traces"]["enabled"]: - raise ValueError("the client copy still has native tracing") - resource = type(instance)(client) - except Exception: - logger.debug("Could not disable Anthropic SDK native tracing for one call", exc_info=True) - if not self._copy_failure_notified: - self._copy_failure_notified = True - logger.warning(_COPY_FAILED) - return wrapped + return None + if self._hook is None: + # A second span is better than a call without the instrumentor's span. + if not self._no_hook_notified: + self._no_hook_notified = True + logger.warning(_NO_HOOK) + return None if not self._notified: self._notified = True logger.info(_NOTICE) - return function.__get__(resource, type(resource)) + return instance._client + + def without_native_span(self, wrapped: Callable[..., Any], instance: Any) -> Callable[..., Any]: + """`wrapped` (the original method bound to `instance`), with the SDK's span off when it would duplicate ours.""" + client = self._client_to_own(instance) + if client is None: + return wrapped + + def call(*args: Any, **kwargs: Any) -> Any: + token = _owned_client.set(client) + try: + return wrapped(*args, **kwargs) + finally: + _owned_client.reset(token) + + return call + + def without_native_span_async(self, wrapped: Callable[..., Any], instance: Any) -> Callable[..., Any]: + """`without_native_span` for coroutine methods: the SDK starts its span while awaited.""" + client = self._client_to_own(instance) + if client is None: + return wrapped + + async def call(*args: Any, **kwargs: Any) -> Any: + token = _owned_client.set(client) + try: + return await wrapped(*args, **kwargs) + finally: + _owned_client.reset(token) + + return call diff --git a/python/frameworks/anthropic/traceai_anthropic/_wrappers.py b/python/frameworks/anthropic/traceai_anthropic/_wrappers.py index e70437ff8..0e9f31331 100644 --- a/python/frameworks/anthropic/traceai_anthropic/_wrappers.py +++ b/python/frameworks/anthropic/traceai_anthropic/_wrappers.py @@ -67,6 +67,12 @@ def _without_native_span(self, wrapped: Callable[..., Any], instance: Any) -> Ca return wrapped return self._native_tracing.without_native_span(wrapped, instance) + def _without_native_span_async(self, wrapped: Callable[..., Any], instance: Any) -> Callable[..., Any]: + """`_without_native_span` for coroutine methods.""" + if self._native_tracing is None: + return wrapped + return self._native_tracing.without_native_span_async(wrapped, instance) + @contextmanager def _start_as_current_span( self, span_name: str, attributes: Optional[Mapping[str, Any]] = None @@ -352,7 +358,7 @@ async def __call__( ) -> Any: if context_api.get_value(context_api._SUPPRESS_INSTRUMENTATION_KEY): return await wrapped(*args, **kwargs) - wrapped = self._without_native_span(wrapped, instance) + wrapped = self._without_native_span_async(wrapped, instance) arguments = kwargs llm_prompt = dict(arguments).pop("system", None) From 54de9ef6d21f854b896d3f4d2e4577311c65139d Mon Sep 17 00:00:00 2001 From: Nikhil Pareek Date: Sun, 1 Nov 2026 16:15:30 -0800 Subject: [PATCH 5/5] docs(anthropic): say which SDK spans the per-call hook turns off OpenTelemetry gives every client with the same tracer provider the same tracer, so while an instrumented Messages call runs, the hook turns off SDK spans for any call to that provider, not only the call's own client. Calls whose SDK spans go to another provider keep them. This corrects the previous commit message, which said calls on other clients keep their spans. Co-Authored-By: Claude Opus 5.5 --- python/frameworks/anthropic/traceai_anthropic/_native.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/python/frameworks/anthropic/traceai_anthropic/_native.py b/python/frameworks/anthropic/traceai_anthropic/_native.py index 998e778c1..4a6c31a75 100644 --- a/python/frameworks/anthropic/traceai_anthropic/_native.py +++ b/python/frameworks/anthropic/traceai_anthropic/_native.py @@ -11,8 +11,10 @@ The call itself is not changed: the original method runs on the application's client, through any other wrappers on it, so client state such as lazily loaded cloud credentials stays on that client. While it runs, the SDK's span start -(`start_api_call_span`, which may return `None` for "no span") returns `None` for this -client's tracer only; other clients' calls inside it keep their spans. +(`start_api_call_span`, which may return `None` for "no span") returns `None` for the +tracer of this client. OpenTelemetry gives clients with the same tracer provider the same +tracer, so an SDK call made inside the Messages call to that same provider (by another +wrapper, say) has no span either; calls whose SDK spans go to another provider keep them. When the SDK's span goes elsewhere (another provider, or nowhere), both are left as they are, so each provider gets the call once. @@ -51,7 +53,7 @@ def _skip_owned_span(wrapped: Callable[..., Any], instance: Any, args: Any, kwargs: Any) -> Any: - """`start_api_call_span`, returning `None` (no span) for the owned client's calls.""" + """`start_api_call_span`, returning `None` (no span) for calls with the owned client's tracer.""" client = _owned_client.get() if client is not None: try: