Merge branch 'feat/move-stream-to-genai-util-as-abc' of https://github.com/eternalcuriouslearner/opentelemetry-python-contrib into feat/move-stream-to-genai-util-as-abc

Author: eternalcuriouslearner

.github/workflows/test.yml

@@ -214,25 +214,6 @@ jobs:
       - name: Run tests
         run: tox -e py314-test-instrumentation-openai-v2-latest -- -ra
 
-  py313-test-instrumentation-openai-v2-pydantic1_ubuntu-latest:
-    name: instrumentation-openai-v2-pydantic1 3.13 Ubuntu
-    runs-on: ubuntu-latest
-    timeout-minutes: 30
-    steps:
-      - name: Checkout repo @ SHA - ${{ github.sha }}
-        uses: actions/checkout@v4
-
-      - name: Set up Python 3.13
-        uses: actions/setup-python@v5
-        with:
-          python-version: "3.13"
-
-      - name: Install tox
-        run: pip install tox-uv
-
-      - name: Run tests
-        run: tox -e py313-test-instrumentation-openai-v2-pydantic1 -- -ra
-
   pypy3-test-instrumentation-openai-v2-oldest_ubuntu-latest:
     name: instrumentation-openai-v2-oldest pypy-3.10 Ubuntu
     runs-on: ubuntu-latest

AGENTS.md

@@ -66,6 +66,9 @@ uv run tox -e typecheck
 - `tox.ini` defines the test matrix - check it for available test environments.
 - Do not add `type: ignore` comments. If a type error arises, solve it properly or write a follow-up plan to address it in another PR.
 - Annotate function signatures (parameters and return types) and class attributes. Prefer `from __future__ import annotations` over runtime-quoted strings.
+- When a file uses `from __future__ import annotations`, do not quote type annotations just to
+  avoid forward references. Keep quotes only for expressions still evaluated at runtime, such as
+  `typing.cast(...)`, unless the referenced type is imported at runtime.
 - Whenever applicable, all code changes should have tests that actually validate the changes.
 
 ## Instrumentation rules

CHANGELOG.md

@@ -33,6 +33,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#3898](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3898))
 - `opentelemetry-instrumentation`: Add experimental metrics attributes Labeler utility
   ([#4288](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4288))
+- `opentelemetry-instrumentation`: Register `OTEL_SEMCONV_STABILITY_OPT_IN` in `environment_variables.py` so `opentelemetry-instrument` exposes a `--semconv_stability_opt_in` CLI argument
+  ([#4438](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4438))
 
 ### Fixed
 
@@ -56,6 +58,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#4368](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4368))
 - `opentelemetry-instrumentation-celery`: Coerce timelimit values to strings in `set_attributes_from_context()` to prevent mixed-type span attribute warning
   ([#4361](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4361))
+- `opentelemetry-instrumentation-mysqlclient`: Update unit tests to properly validate trace context trace flag values.
+  ([#4560](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4560))
+
+### Fixed
+
+- `opentelemetry-instrumentation-confluent-kafka`: Populate `server.address` and `server.port` span attributes from the producer/consumer `bootstrap.servers` config; previously `KafkaPropertiesExtractor.extract_bootstrap_servers` was defined but never called
+  ([#4423](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4423))
 
 ### Breaking changes
 

CONTRIBUTING.md

@@ -406,6 +406,21 @@ Instrumentations that relate to [Generative AI](https://opentelemetry.io/docs/sp
 
 OpenTelemetry is an open source community, and as such, greatly encourages contributions from anyone interested in the project. With that being said, there is a certain level of expectation from contributors even after a pull request is merged, specifically pertaining to instrumentations. The OpenTelemetry Python community expects contributors to maintain a level of support and interest in the instrumentations they contribute. This is to ensure that the instrumentation does not become stale and still functions the way the original contributor intended. Some instrumentations also pertain to libraries that the current members of the community are not so familiar with, so it is necessary to rely on the expertise of the original contributing parties.
 
+### Use of AI coding assistants
+
+AI coding assistants can be valuable tools for contributors. They may help with understanding the codebase, drafting changes, writing tests and improving documentation. Contributors are welcome to use AI tools as a part of their contribution workflow. 
+
+However, contributors remain responsible for the quality and correctness of their contributions. AI generated material must be carefully reviewed, validated and adapted by the contributor before being submitted to the project.
+
+Maintainers reserve the right to close pull requests or other contributions that appear to have been primarily generated by AI with little or no meaningful review from the contributor. This includes, but is not limited to:
+
+- Opening a large number of low quality pull requests in a short period of time.
+- Submitting changes that are incomplete, incorrect, unrelated to the issue or not aligned with project conventions.
+- Posting review replies, issue comments or other responses that appear to be AI generated and/or do not meaningfully address maintainer feedback.
+- Submitting pull requests that require maintainers to spend significant time correcting or explaining basic issues that should have been caught by the contributor before submission.
+
+This policy is intended to reduce unnecessary burden on maintainers and ensure that AI tools are used responsibly in support of high quality contributions.
+
 ### Guidelines for native OpenTelemetry instrumentation
 
 The preferred approach to supporting instrumentation of a tool or framework is native OpenTelemetry instrumentation. Compared to adding a new instrumentation library to this community repository, native instrumentation is better for:

instrumentation-genai/AGENTS.md

@@ -68,6 +68,11 @@ except Exception as exc:
     raise
 ```
 
+Content capture decisions must come from the shared handler, not from instrumentation-local
+environment checks or duplicated helper logic. Evaluate the handler's content-capture API once
+when creating wrappers (for example, `capture_content = handler.should_capture_content()`) and
+pass that value through invocation/request helpers.
+
 ## 4. Semantic conventions
 
 Attributes, spans, events, and metrics follow the

instrumentation-genai/opentelemetry-instrumentation-anthropic/CHANGELOG.md

@@ -14,6 +14,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Add instrumentation for Anthropic `Messages.stream()` helper method
+  ([#4499](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4499))
 - Add async Anthropic message stream wrappers and manager wrappers, with wrapper
   tests ([#4346](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4346))
   - `AsyncMessagesStreamWrapper` for async message stream telemetry

instrumentation-genai/opentelemetry-instrumentation-anthropic/src/opentelemetry/instrumentation/anthropic/__init__.py

@@ -42,14 +42,16 @@
     wrap_function_wrapper,  # pyright: ignore[reportUnknownVariableType]
 )
 
-from opentelemetry.instrumentation.anthropic.package import _instruments
-from opentelemetry.instrumentation.anthropic.patch import (
-    messages_create,
-)
 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
 from opentelemetry.instrumentation.utils import unwrap
 from opentelemetry.util.genai.handler import TelemetryHandler
 
+from .package import _instruments
+from .patch import (
+    messages_create,
+    messages_stream,
+)
+
 
 class AnthropicInstrumentor(BaseInstrumentor):
     """An instrumentor for the Anthropic Python SDK.
@@ -88,12 +90,17 @@ def _instrument(self, **kwargs: Any) -> None:
             logger_provider=logger_provider,
         )
 
-        # Patch Messages.create
+        # Patch Messages.create and Messages.stream
         wrap_function_wrapper(
             "anthropic.resources.messages",
             "Messages.create",
             messages_create(handler),
         )
+        wrap_function_wrapper(
+            "anthropic.resources.messages",
+            "Messages.stream",
+            messages_stream(handler),
+        )
 
     def _uninstrument(self, **kwargs: Any) -> None:
         """Disable Anthropic instrumentation.
@@ -106,3 +113,7 @@ def _uninstrument(self, **kwargs: Any) -> None:
             anthropic.resources.messages.Messages,  # pyright: ignore[reportAttributeAccessIssue,reportUnknownMemberType,reportUnknownArgumentType]
             "create",
         )
+        unwrap(
+            anthropic.resources.messages.Messages,  # pyright: ignore[reportAttributeAccessIssue,reportUnknownMemberType,reportUnknownArgumentType]
+            "stream",
+        )

instrumentation-genai/opentelemetry-instrumentation-anthropic/src/opentelemetry/instrumentation/anthropic/messages_extractors.py

@@ -16,9 +16,9 @@
 from opentelemetry.semconv._incubating.attributes import (
     server_attributes as ServerAttributes,
 )
+from opentelemetry.util.genai.invocation import InferenceInvocation
 from opentelemetry.util.genai.types import (
     InputMessage,
-    LLMInvocation,
     MessagePart,
     OutputMessage,
 )
@@ -144,7 +144,7 @@ def get_output_messages_from_message(
 
 
 def set_invocation_response_attributes(
-    invocation: LLMInvocation,
+    invocation: InferenceInvocation,
     message: Message | None,
     capture_content: bool,
 ) -> None:
@@ -209,18 +209,15 @@ def extract_params(  # pylint: disable=too-many-locals
     )
 
 
-def _set_server_address_and_port(
+def get_server_address_and_port(
     client_instance: "Messages",
-    attributes: dict[str, AttributeValue | None],
-) -> None:
+) -> tuple[str | None, int | None]:
     base_url = client_instance._client.base_url
-    host = base_url.host
-    if host:
-        attributes[ServerAttributes.SERVER_ADDRESS] = host
-
     port = base_url.port
-    if port and port != 443 and port > 0:
-        attributes[ServerAttributes.SERVER_PORT] = port
+    return (
+        base_url.host or None,
+        port if port and port != 443 and port > 0 else None,
+    )
 
 
 def get_llm_request_attributes(
@@ -236,5 +233,9 @@ def get_llm_request_attributes(
         GenAIAttributes.GEN_AI_REQUEST_TOP_K: params.top_k,
         GenAIAttributes.GEN_AI_REQUEST_STOP_SEQUENCES: params.stop_sequences,
     }
-    _set_server_address_and_port(client_instance, attributes)
+    address, port = get_server_address_and_port(client_instance)
+    if address is not None:
+        attributes[ServerAttributes.SERVER_ADDRESS] = address
+    if port is not None:
+        attributes[ServerAttributes.SERVER_PORT] = port
     return {k: v for k, v in attributes.items() if v is not None}

instrumentation-genai/opentelemetry-instrumentation-anthropic/src/opentelemetry/instrumentation/anthropic/patch.py

@@ -15,26 +15,25 @@
     gen_ai_attributes as GenAIAttributes,
 )
 from opentelemetry.util.genai.handler import TelemetryHandler
-from opentelemetry.util.genai.types import (
-    Error,
-    LLMInvocation,  # TODO: migrate to InferenceInvocation
-)
-from opentelemetry.util.genai.utils import (
-    should_capture_content_on_spans_in_experimental_mode,
-)
+from opentelemetry.util.genai.invocation import InferenceInvocation
 
 from .messages_extractors import (
     extract_params,
     get_input_messages,
     get_llm_request_attributes,
+    get_server_address_and_port,
     get_system_instruction,
 )
 from .wrappers import (
+    MessagesStreamManagerWrapper,
     MessagesStreamWrapper,
     MessageWrapper,
 )
 
 if TYPE_CHECKING:
+    from anthropic.lib.streaming._messages import (  # pylint: disable=no-name-in-module
+        MessageStreamManager,
+    )
     from anthropic.resources.messages import Messages
     from anthropic.types import RawMessageStreamEvent
 
@@ -48,73 +47,109 @@ def messages_create(
 ) -> Callable[
     ...,
     Union[
-        "AnthropicMessage",
-        "AnthropicStream[RawMessageStreamEvent]",
+        AnthropicMessage,
+        AnthropicStream[RawMessageStreamEvent],
         MessagesStreamWrapper[None],
     ],
 ]:
     """Wrap the `create` method of the `Messages` class to trace it."""
-    capture_content = should_capture_content_on_spans_in_experimental_mode()
+    capture_content = handler.should_capture_content()
 
     def traced_method(
         wrapped: Callable[
             ...,
             Union[
-                "AnthropicMessage",
-                "AnthropicStream[RawMessageStreamEvent]",
+                AnthropicMessage,
+                AnthropicStream[RawMessageStreamEvent],
             ],
         ],
-        instance: "Messages",
+        instance: Messages,
         args: tuple[Any, ...],
         kwargs: dict[str, Any],
     ) -> Union[
-        "AnthropicMessage",
-        "AnthropicStream[RawMessageStreamEvent]",
+        AnthropicMessage,
+        AnthropicStream[RawMessageStreamEvent],
         MessagesStreamWrapper[None],
     ]:
-        params = extract_params(*args, **kwargs)
-        attributes = get_llm_request_attributes(params, instance)
-        request_model_attribute = attributes.get(
-            GenAIAttributes.GEN_AI_REQUEST_MODEL
-        )
-        request_model = (
-            request_model_attribute
-            if isinstance(request_model_attribute, str)
-            else params.model
+        invocation = _create_invocation(
+            handler, instance, args, kwargs, capture_content
         )
-
-        invocation = LLMInvocation(
-            request_model=request_model,
-            provider=ANTHROPIC,
-            input_messages=get_input_messages(params.messages)
-            if capture_content
-            else [],
-            system_instruction=get_system_instruction(params.system)
-            if capture_content
-            else [],
-            attributes=attributes,
-        )
-
-        # Use manual lifecycle management for both streaming and non-streaming
-        handler.start_llm(invocation)
         try:
             result = wrapped(*args, **kwargs)
             if isinstance(result, AnthropicStream):
                 return MessagesStreamWrapper(
-                    result, handler, invocation, capture_content
+                    result, invocation, capture_content
                 )
 
             wrapper = MessageWrapper(result, capture_content)
             wrapper.extract_into(invocation)
-            handler.stop_llm(invocation)
+            invocation.stop()
             return wrapper.message
         except Exception as exc:
-            handler.fail_llm(
-                invocation, Error(message=str(exc), type=type(exc))
-            )
+            invocation.fail(exc)
             raise
 
     return cast(
         'Callable[..., Union["AnthropicMessage", "AnthropicStream[RawMessageStreamEvent]", MessagesStreamWrapper[None]]]',
         traced_method,
     )
+
+
+def _create_invocation(
+    handler: TelemetryHandler,
+    instance: Messages,
+    args: tuple[Any, ...],
+    kwargs: dict[str, Any],
+    capture_content: bool,
+) -> InferenceInvocation:
+    params = extract_params(*args, **kwargs)
+    attributes = get_llm_request_attributes(params, instance)
+    request_model_attribute = attributes.get(
+        GenAIAttributes.GEN_AI_REQUEST_MODEL
+    )
+    request_model = (
+        request_model_attribute
+        if isinstance(request_model_attribute, str)
+        else params.model
+    )
+
+    server_address, server_port = get_server_address_and_port(instance)
+    invocation = handler.start_inference(
+        provider=ANTHROPIC,
+        request_model=request_model,
+        server_address=server_address,
+        server_port=server_port,
+    )
+    invocation.input_messages = (
+        get_input_messages(params.messages) if capture_content else []
+    )
+    invocation.system_instruction = (
+        get_system_instruction(params.system) if capture_content else []
+    )
+    invocation.attributes = attributes
+    return invocation
+
+
+def messages_stream(
+    handler: TelemetryHandler,
+) -> Callable[..., MessagesStreamManagerWrapper[Any]]:
+    """Wrap the sync `stream` method of the `Messages` class."""
+    capture_content = handler.should_capture_content()
+
+    def traced_method(
+        wrapped: Callable[..., MessageStreamManager],
+        instance: Messages,
+        args: tuple[Any, ...],
+        kwargs: dict[str, Any],
+    ) -> MessagesStreamManagerWrapper[Any]:
+        return MessagesStreamManagerWrapper(
+            wrapped(*args, **kwargs),
+            lambda: _create_invocation(
+                handler, instance, args, kwargs, capture_content
+            ),
+            capture_content,
+        )
+
+    return cast(
+        "Callable[..., MessagesStreamManagerWrapper[Any]]", traced_method
+    )