feat: Add support to disable telemetry instrumentation

pyproject.toml

@@ -72,9 +72,10 @@ exclude = ["tests/"]
 testpaths = ["tests"]
 python_files = "test_*.py"
 python_functions = "test_*"
-addopts = "-ra --strict-markers"
+addopts = "-ra --strict-markers --dist loadgroup"
 markers = [
   "asyncio: mark a test as a coroutine that should be run by pytest-asyncio",
+  "xdist_group: mark a test to run in a specific sequential group for isolation",
 ]
 
 [tool.pytest-asyncio]
@@ -93,6 +94,7 @@ dev = [
   "pytest-asyncio>=0.26.0",
   "pytest-cov>=6.1.1",
   "pytest-mock>=3.14.0",
+  "pytest-xdist>=3.6.1",
   "respx>=0.20.2",
   "ruff>=0.12.8",
   "uv-dynamic-versioning>=0.8.2",

src/a2a/utils/telemetry.py

@@ -18,6 +18,16 @@
 - Automatic recording of exceptions and setting of span status.
 - Selective method tracing in classes using include/exclude lists.
 
+Configuration:
+- Environment Variable Control: OpenTelemetry instrumentation can be
+  disabled using the `OTEL_INSTRUMENTATION_A2A_SDK_ENABLED` environment
+  variable.
+
+  - Default: `true` (tracing enabled when OpenTelemetry is installed)
+  - To disable: Set `OTEL_INSTRUMENTATION_A2A_SDK_ENABLED=false`
+  - Case insensitive: 'true', 'True', 'TRUE' all enable tracing
+  - Any other value disables tracing and logs a debug message
+
 Usage:
     For a single function:
     ```python
@@ -57,6 +67,7 @@ def internal_method(self):
 import functools
 import inspect
 import logging
+import os
 
 from collections.abc import Callable
 from typing import TYPE_CHECKING, Any
@@ -74,11 +85,33 @@ def internal_method(self):
     from opentelemetry.trace import SpanKind as _SpanKind
     from opentelemetry.trace import StatusCode
 
+    otel_installed = True
+
 except ImportError:
     logger.debug(
         'OpenTelemetry not found. Tracing will be disabled. '
         'Install with: \'pip install "a2a-sdk[telemetry]"\''
     )
+    otel_installed = False
+
+ENABLED_ENV_VAR = 'OTEL_INSTRUMENTATION_A2A_SDK_ENABLED'
+INSTRUMENTING_MODULE_NAME = 'a2a-python-sdk'
+INSTRUMENTING_MODULE_VERSION = '1.0.0'
+
+# Check if tracing is enabled via environment variable
+env_value = os.getenv(ENABLED_ENV_VAR, 'true')
+otel_enabled = env_value.lower() == 'true'
+
+# Log when tracing is explicitly disabled via environment variable
+if otel_installed and not otel_enabled:
+    logger.debug(
+        'A2A OTEL instrumentation disabled via environment variable '
+        '%s=%r. Tracing will be disabled.',
+        ENABLED_ENV_VAR,
+        env_value,
+    )
+
+if not otel_installed or not otel_enabled:
 
     class _NoOp:
         """A no-op object that absorbs all tracing calls when OpenTelemetry is not installed."""
@@ -99,12 +132,9 @@ def __getattr__(self, name: str) -> Any:
     _SpanKind = _NoOp()  # type: ignore
     StatusCode = _NoOp()  # type: ignore
 
-SpanKind = _SpanKind
+SpanKind = _SpanKind  # type: ignore
 __all__ = ['SpanKind']
 
-INSTRUMENTING_MODULE_NAME = 'a2a-python-sdk'
-INSTRUMENTING_MODULE_VERSION = '1.0.0'
-
 
 def trace_function(  # noqa: PLR0915
     func: Callable | None = None,

tests/utils/test_telemetry.py

@@ -1,6 +1,8 @@
 import asyncio
+import importlib
+import sys
 
-from collections.abc import Generator
+from collections.abc import Callable, Generator
 from typing import Any, NoReturn
 from unittest import mock
 
@@ -30,6 +32,32 @@ def patch_trace_get_tracer(
         yield
 
 
+@pytest.fixture
+def reload_telemetry_module(
+    monkeypatch: pytest.MonkeyPatch,
+) -> Generator[Callable[[str | None], Any], None, None]:
+    """Fixture to handle telemetry module reloading with env var control."""
+
+    def _reload(env_value: str | None = None) -> Any:
+        if env_value is None:
+            monkeypatch.delenv(
+                'OTEL_INSTRUMENTATION_A2A_SDK_ENABLED', raising=False
+            )
+        else:
+            monkeypatch.setenv(
+                'OTEL_INSTRUMENTATION_A2A_SDK_ENABLED', env_value
+            )
+
+        sys.modules.pop('a2a.utils.telemetry', None)
+        module = importlib.import_module('a2a.utils.telemetry')
+        return module
+
+    yield _reload
+
+    # Cleanup to ensure other tests aren't affected by a "poisoned" sys.modules
+    sys.modules.pop('a2a.utils.telemetry', None)
+
+
 def test_trace_function_sync_success(mock_span: mock.MagicMock) -> None:
     @trace_function
     def foo(x, y):
@@ -198,3 +226,43 @@ def foo(self) -> str:
     assert obj.foo() == 'foo'
     assert hasattr(obj.foo, '__wrapped__')
     assert hasattr(obj, 'x')
+
+
+@pytest.mark.xdist_group(name='telemetry_isolation')
+@pytest.mark.parametrize(
+    'env_value,expected_tracing',
+    [
+        (None, True),  # Default: env var not set, tracing enabled
+        ('true', True),  # Explicitly enabled
+        ('True', True),  # Case insensitive
+        ('false', False),  # Disabled
+        ('', False),  # Empty string = false
+    ],
+)
+def test_env_var_controls_instrumentation(
+    reload_telemetry_module: Callable[[str | None], Any],
+    env_value: str | None,
+    expected_tracing: bool,
+) -> None:
+    """Test OTEL_INSTRUMENTATION_A2A_SDK_ENABLED controls span creation."""
+    telemetry_module = reload_telemetry_module(env_value)
+
+    is_noop = type(telemetry_module.trace).__name__ == '_NoOp'
+
+    assert is_noop != expected_tracing
+
+
+@pytest.mark.xdist_group(name='telemetry_isolation')
+def test_env_var_disabled_logs_message(
+    reload_telemetry_module: Callable[[str | None], Any],
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Test that disabling via env var logs appropriate debug message."""
+    with caplog.at_level('DEBUG', logger='a2a.utils.telemetry'):
+        reload_telemetry_module('false')
+
+    assert (
+        'A2A OTEL instrumentation disabled via environment variable'
+        in caplog.text
+    )
+    assert 'OTEL_INSTRUMENTATION_A2A_SDK_ENABLED' in caplog.text