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..3b30a34ec 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`, `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` (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_otlp.py b/python/frameworks/anthropic/tests/test_native_otlp.py new file mode 100644 index 000000000..5a1945976 --- /dev/null +++ b/python/frameworks/anthropic/tests/test_native_otlp.py @@ -0,0 +1,157 @@ +"""Each Messages call reaches the Future AGI collector path once, with what Observe reads. + +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 +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 + + +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" + + +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: + provider.shutdown() + + assert_collector_auth(collector) + 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_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: + 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..4f05ccebb --- /dev/null +++ b/python/frameworks/anthropic/tests/test_native_ownership.py @@ -0,0 +1,618 @@ +"""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 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 +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 fi_instrumentation import using_attributes +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" +FI_SCOPE = "traceai_anthropic" +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_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) == [FI_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) == [FI_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) == [FI_SCOPE] + 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) == [FI_SCOPE] + + +@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) == [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_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 == "Messages" else Decision.RECORD_AND_SAMPLE + return SamplingResult(decision) + + def get_description(self): + return "DropInstrumentor" + + fi, fi_exporter = make_provider(sampler=DropInstrumentor()) + try: + instrument(tracer_provider=fi) + call("create", False, open_telemetry={"tracer_provider": fi}) + assert scopes(fi_exporter) == [] + finally: + fi.shutdown() + + +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") + 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) + # 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 + 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_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}) + 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 (await stream.__anext__()).type == "message_start" + + 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 next(stream).type == "message_start" + assert scopes(fi_exporter) == [FI_SCOPE] + + +@pytest.mark.parametrize("asynchronous", [False, True]) +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] == [FI_SCOPE] + assert spans[0].status.status_code == trace.StatusCode.ERROR + + +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.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() + + +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 FI_SCOPE in scopes(fi_exporter) + + +@pytest.mark.parametrize("asynchronous", [False, True]) +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) + 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}) + 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"]) + + +@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 + + +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_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) + 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()) + + +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 e60118782..bf81c28aa 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, @@ -33,6 +34,7 @@ class AnthropicInstrumentor(BaseInstrumentor): # type: ignore[misc] "_original_messages_count_tokens", "_original_async_messages_create", "_original_protect", + "_native_tracing", "_instruments", "_tracer", ) @@ -57,6 +59,12 @@ def _instrument(self, **kwargs: Any) -> None: logger.warning("ai-evaluation is not installed, please install it to trace protect") Protect = None + # 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")) + 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")): @@ -93,21 +101,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: @@ -150,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 new file mode 100644 index 000000000..4a6c31a75 --- /dev/null +++ b/python/frameworks/anthropic/traceai_anthropic/_native.py @@ -0,0 +1,177 @@ +"""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 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 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 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. +""" + +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__) + +_NOTICE = ( + "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." +) + +_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 calls with the owned client's tracer.""" + 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.""" + + 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._hook: Optional[FunctionWrapper] = None + self._original: Any = None + self._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. + + 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) + 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() + 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 + + 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 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 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 c21251a9f..0e9f31331 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,29 @@ 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 _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) + + 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( @@ -245,6 +265,7 @@ def __call__( ) -> Any: 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) @@ -337,6 +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_async(wrapped, instance) arguments = kwargs llm_prompt = dict(arguments).pop("system", None) @@ -409,6 +431,7 @@ def __call__( ): 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)