From fae36b36f2f8b9f21f6507678c9a172e4370dc69 Mon Sep 17 00:00:00 2001 From: Amit Mukherjee <45551399+amitmukh@users.noreply.github.com> Date: Tue, 3 Mar 2026 14:12:21 -0600 Subject: [PATCH] Python: Add OpenTelemetry instrumentation to ClaudeAgent (#4278) (#4326) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * Python: Add OpenTelemetry instrumentation to ClaudeAgent (#4278) Add inline telemetry to ClaudeAgent.run() so that enable_instrumentation() emits invoke_agent spans and metrics. Covers both streaming and non-streaming paths using the same observability helpers as AgentTelemetryLayer. Adds 5 unit tests for telemetry behavior. Co-Authored-By: amitmukh * Address PR review feedback for ClaudeAgent telemetry - Add justification comment for private observability API imports - Pass system_instructions to capture_messages for system prompt capture - Use monkeypatch instead of try/finally for test global state isolation Co-Authored-By: amitmukh Co-Authored-By: Claude * Adopt AgentTelemetryLayer instead of inline telemetry Restructure ClaudeAgent to inherit from AgentTelemetryLayer via a _ClaudeAgentRunImpl mixin, eliminating duplicated telemetry code and private API imports. MRO: ClaudeAgent → AgentTelemetryLayer → _ClaudeAgentRunImpl → BaseAgent - Remove inline _run_with_telemetry / _run_with_telemetry_stream methods - Remove private observability helper imports (_capture_messages, etc.) - Add default_options property mapping system_prompt → instructions - Net -105 lines by reusing core telemetry layer Co-Authored-By: amitmukh Co-Authored-By: Claude * Fix mypy: align _ClaudeAgentRunImpl.run() signature with AgentTelemetryLayer.run() Remove explicit `options` parameter from mixin's run() signature and extract it from **kwargs to match AgentTelemetryLayer's signature. Also align overload return types (ResponseStream, Awaitable) to match. Co-Authored-By: Claude * Introduce RawClaudeAgent following framework's RawAgent/Agent pattern Replace private _ClaudeAgentRunImpl mixin with public RawClaudeAgent class that contains all core logic (init, run, lifecycle, tools). ClaudeAgent becomes a thin wrapper that adds AgentTelemetryLayer. - RawClaudeAgent(BaseAgent): full implementation without telemetry - ClaudeAgent(AgentTelemetryLayer, RawClaudeAgent): adds OTel tracing - Export RawClaudeAgent from package __init__.py Users who want to skip telemetry or provide their own can use RawClaudeAgent directly. Co-Authored-By: Claude * Address review nits: trim RawClaudeAgent docstring, fix import paths - Simplify RawClaudeAgent docstring to a single basic example (not the primary entry point for most users) - Use agent_framework.anthropic import path in docstrings instead of direct agent_framework_claude path - Add RawClaudeAgent to agent_framework.anthropic lazy re-exports Co-Authored-By: Claude --------- Co-authored-by: Amit Mukherjee Co-authored-by: amitmukh Co-authored-by: Claude Co-authored-by: Dmytro Struk <13853051+dmytrostruk@users.noreply.github.com> --- .../claude/agent_framework_claude/__init__.py | 3 +- .../claude/agent_framework_claude/_agent.py | 189 +++++++++--------- .../claude/tests/test_claude_agent.py | 188 +++++++++++++++++ .../agent_framework/anthropic/__init__.py | 2 + 4 files changed, 290 insertions(+), 92 deletions(-) diff --git a/python/packages/claude/agent_framework_claude/__init__.py b/python/packages/claude/agent_framework_claude/__init__.py index 3c666f4a31..abf522fa4f 100644 --- a/python/packages/claude/agent_framework_claude/__init__.py +++ b/python/packages/claude/agent_framework_claude/__init__.py @@ -2,7 +2,7 @@ import importlib.metadata -from ._agent import ClaudeAgent, ClaudeAgentOptions, ClaudeAgentSettings +from ._agent import ClaudeAgent, ClaudeAgentOptions, ClaudeAgentSettings, RawClaudeAgent try: __version__ = importlib.metadata.version(__name__) @@ -13,5 +13,6 @@ __all__ = [ "ClaudeAgent", "ClaudeAgentOptions", "ClaudeAgentSettings", + "RawClaudeAgent", "__version__", ] diff --git a/python/packages/claude/agent_framework_claude/_agent.py b/python/packages/claude/agent_framework_claude/_agent.py index 43f001b3db..f5aabc43a9 100644 --- a/python/packages/claude/agent_framework_claude/_agent.py +++ b/python/packages/claude/agent_framework_claude/_agent.py @@ -27,6 +27,7 @@ from agent_framework import ( normalize_tools, ) from agent_framework.exceptions import AgentException +from agent_framework.observability import AgentTelemetryLayer from claude_agent_sdk import ( AssistantMessage, ClaudeSDKClient, @@ -171,8 +172,11 @@ OptionsT = TypeVar( ) -class ClaudeAgent(BaseAgent, Generic[OptionsT]): - """Claude Agent using Claude Code CLI. +class RawClaudeAgent(BaseAgent, Generic[OptionsT]): + """Claude Agent using Claude Code CLI without telemetry layers. + + This is the core Claude agent implementation without OpenTelemetry instrumentation. + For most use cases, prefer :class:`ClaudeAgent` which includes telemetry support. Wraps the Claude Agent SDK to provide agentic capabilities including tool use, session management, and streaming responses. @@ -188,45 +192,13 @@ class ClaudeAgent(BaseAgent, Generic[OptionsT]): .. code-block:: python - from agent_framework_claude import ClaudeAgent + from agent_framework.anthropic import RawClaudeAgent - async with ClaudeAgent( + async with RawClaudeAgent( instructions="You are a helpful assistant.", ) as agent: response = await agent.run("Hello!") print(response.text) - - With streaming: - - .. code-block:: python - - async with ClaudeAgent() as agent: - async for update in agent.run("Write a poem"): - print(update.text, end="", flush=True) - - With session management: - - .. code-block:: python - - async with ClaudeAgent() as agent: - session = agent.create_session() - await agent.run("Remember my name is Alice", session=session) - response = await agent.run("What's my name?", session=session) - # Claude will remember "Alice" from the same session - - With Agent Framework tools: - - .. code-block:: python - - from agent_framework import tool - - @tool - def greet(name: str) -> str: - \"\"\"Greet someone by name.\"\"\" - return f"Hello, {name}!" - - async with ClaudeAgent(tools=[greet]) as agent: - response = await agent.run("Greet Alice") """ AGENT_PROVIDER_NAME: ClassVar[str] = "anthropic.claude" @@ -246,7 +218,7 @@ class ClaudeAgent(BaseAgent, Generic[OptionsT]): env_file_path: str | None = None, env_file_encoding: str | None = None, ) -> None: - """Initialize a ClaudeAgent instance. + """Initialize a RawClaudeAgent instance. Args: instructions: System prompt for the agent. @@ -343,7 +315,7 @@ class ClaudeAgent(BaseAgent, Generic[OptionsT]): normalized = normalize_tools(tool) self._custom_tools.extend(normalized) - async def __aenter__(self) -> ClaudeAgent[OptionsT]: + async def __aenter__(self) -> RawClaudeAgent[OptionsT]: """Start the agent when entering async context.""" await self.start() return self @@ -568,61 +540,19 @@ class ClaudeAgent(BaseAgent, Generic[OptionsT]): return "" return "\n".join([msg.text or "" for msg in messages]) - @overload - def run( - self, - messages: AgentRunInputs | None = None, - *, - stream: Literal[True], - session: AgentSession | None = None, - options: OptionsT | MutableMapping[str, Any] | None = None, - **kwargs: Any, - ) -> AsyncIterable[AgentResponseUpdate]: ... + @property + def default_options(self) -> dict[str, Any]: + """Expose options with ``instructions`` key. - @overload - async def run( - self, - messages: AgentRunInputs | None = None, - *, - stream: Literal[False] = ..., - session: AgentSession | None = None, - options: OptionsT | MutableMapping[str, Any] | None = None, - **kwargs: Any, - ) -> AgentResponse[Any]: ... - - def run( - self, - messages: AgentRunInputs | None = None, - *, - stream: bool = False, - session: AgentSession | None = None, - options: OptionsT | MutableMapping[str, Any] | None = None, - **kwargs: Any, - ) -> AsyncIterable[AgentResponseUpdate] | Awaitable[AgentResponse[Any]]: - """Run the agent with the given messages. - - Args: - messages: The messages to process. - - Keyword Args: - stream: If True, returns an async iterable of updates. If False (default), - returns an awaitable AgentResponse. - session: The conversation session. If session has service_session_id set, - the agent will resume that session. - options: Runtime options (model, permission_mode can be changed per-request). - kwargs: Additional keyword arguments. - - Returns: - When stream=True: An ResponseStream for streaming updates. - When stream=False: An Awaitable[AgentResponse] with the complete response. + Maps ``system_prompt`` to ``instructions`` for compatibility with + :class:`AgentTelemetryLayer`, which reads the system prompt from + the ``instructions`` key. """ - response = ResponseStream( - self._get_stream(messages, session=session, options=options, **kwargs), - finalizer=self._finalize_response, - ) - if stream: - return response - return response.get_final_response() + opts = dict(self._default_options) + system_prompt = opts.pop("system_prompt", None) + if system_prompt is not None: + opts["instructions"] = system_prompt + return opts def _finalize_response(self, updates: Sequence[AgentResponseUpdate]) -> AgentResponse[Any]: """Build AgentResponse and propagate structured_output as value. @@ -636,6 +566,61 @@ class ClaudeAgent(BaseAgent, Generic[OptionsT]): structured_output = getattr(self, "_structured_output", None) return AgentResponse.from_updates(updates, value=structured_output) + @overload + def run( + self, + messages: AgentRunInputs | None = None, + *, + stream: Literal[False] = ..., + session: AgentSession | None = None, + **kwargs: Any, + ) -> Awaitable[AgentResponse[Any]]: ... + + @overload + def run( + self, + messages: AgentRunInputs | None = None, + *, + stream: Literal[True], + session: AgentSession | None = None, + **kwargs: Any, + ) -> ResponseStream[AgentResponseUpdate, AgentResponse[Any]]: ... + + def run( + self, + messages: AgentRunInputs | None = None, + *, + stream: bool = False, + session: AgentSession | None = None, + **kwargs: Any, + ) -> Awaitable[AgentResponse[Any]] | ResponseStream[AgentResponseUpdate, AgentResponse[Any]]: + """Run the agent with the given messages. + + Args: + messages: The messages to process. + + Keyword Args: + stream: If True, returns an async iterable of updates. If False (default), + returns an awaitable AgentResponse. + session: The conversation session. If session has service_session_id set, + the agent will resume that session. + kwargs: Additional keyword arguments including 'options' for runtime options + (model, permission_mode can be changed per-request). + + Returns: + When stream=True: An ResponseStream for streaming updates. + When stream=False: An Awaitable[AgentResponse] with the complete response. + """ + options = kwargs.pop("options", None) + response = ResponseStream( + self._get_stream(messages, session=session, options=options, **kwargs), + finalizer=self._finalize_response, + ) + + if stream: + return response + return response.get_final_response() + async def _get_stream( self, messages: AgentRunInputs | None = None, @@ -721,3 +706,25 @@ class ClaudeAgent(BaseAgent, Generic[OptionsT]): # Store structured output for the finalizer self._structured_output = structured_output + + +class ClaudeAgent(AgentTelemetryLayer, RawClaudeAgent[OptionsT], Generic[OptionsT]): + """Claude Agent with OpenTelemetry instrumentation. + + This is the recommended agent class for most use cases. It includes + OpenTelemetry-based telemetry for observability. For a minimal + implementation without telemetry, use :class:`RawClaudeAgent`. + + Examples: + Basic usage with context manager: + + .. code-block:: python + + from agent_framework.anthropic import ClaudeAgent + + async with ClaudeAgent( + instructions="You are a helpful assistant.", + ) as agent: + response = await agent.run("Hello!") + print(response.text) + """ diff --git a/python/packages/claude/tests/test_claude_agent.py b/python/packages/claude/tests/test_claude_agent.py index 0e126c36b9..e48a3b05d9 100644 --- a/python/packages/claude/tests/test_claude_agent.py +++ b/python/packages/claude/tests/test_claude_agent.py @@ -945,3 +945,191 @@ class TestClaudeAgentStructuredOutput: with pytest.raises(AgentException) as exc_info: await agent.run("Hello") assert "Something went wrong" in str(exc_info.value) + + +# region Test ClaudeAgent Telemetry + + +class TestClaudeAgentTelemetry: + """Tests for ClaudeAgent OpenTelemetry instrumentation.""" + + @staticmethod + async def _create_async_generator(items: list[Any]) -> Any: + """Helper to create async generator from list.""" + for item in items: + yield item + + def _create_mock_client(self, messages: list[Any]) -> MagicMock: + """Create a mock ClaudeSDKClient that yields given messages.""" + mock_client = MagicMock() + mock_client.connect = AsyncMock() + mock_client.disconnect = AsyncMock() + mock_client.query = AsyncMock() + mock_client.set_model = AsyncMock() + mock_client.set_permission_mode = AsyncMock() + mock_client.receive_response = MagicMock(return_value=self._create_async_generator(messages)) + return mock_client + + def _create_standard_messages(self) -> list[Any]: + """Create a standard set of mock messages for testing.""" + from claude_agent_sdk import AssistantMessage, ResultMessage, TextBlock + from claude_agent_sdk.types import StreamEvent + + return [ + StreamEvent( + event={ + "type": "content_block_delta", + "delta": {"type": "text_delta", "text": "Hello!"}, + }, + uuid="event-1", + session_id="session-123", + ), + AssistantMessage( + content=[TextBlock(text="Hello!")], + model="claude-sonnet", + ), + ResultMessage( + subtype="success", + duration_ms=100, + duration_api_ms=50, + is_error=False, + num_turns=1, + session_id="session-123", + ), + ] + + async def test_run_emits_span_when_instrumentation_enabled(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test that run() creates an OpenTelemetry span when instrumentation is enabled.""" + from agent_framework.observability import OBSERVABILITY_SETTINGS + + messages = self._create_standard_messages() + mock_client = self._create_mock_client(messages) + + monkeypatch.setattr(OBSERVABILITY_SETTINGS, "enable_instrumentation", True) + + with ( + patch("agent_framework_claude._agent.ClaudeSDKClient", return_value=mock_client), + patch("agent_framework.observability._get_span") as mock_get_span, + ): + mock_span = MagicMock() + mock_get_span.return_value.__enter__ = MagicMock(return_value=mock_span) + mock_get_span.return_value.__exit__ = MagicMock(return_value=False) + + agent = ClaudeAgent(name="test-agent") + response = await agent.run("Hello") + + assert response.text == "Hello!" + mock_get_span.assert_called_once() + call_kwargs = mock_get_span.call_args[1] + assert call_kwargs["attributes"]["gen_ai.agent.name"] == "test-agent" + assert call_kwargs["attributes"]["gen_ai.operation.name"] == "invoke_agent" + + async def test_run_skips_telemetry_when_instrumentation_disabled(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test that run() skips telemetry when instrumentation is disabled.""" + from agent_framework.observability import OBSERVABILITY_SETTINGS + + messages = self._create_standard_messages() + mock_client = self._create_mock_client(messages) + + monkeypatch.setattr(OBSERVABILITY_SETTINGS, "enable_instrumentation", False) + + with ( + patch("agent_framework_claude._agent.ClaudeSDKClient", return_value=mock_client), + patch("agent_framework.observability._get_span") as mock_get_span, + ): + agent = ClaudeAgent(name="test-agent") + response = await agent.run("Hello") + + assert response.text == "Hello!" + mock_get_span.assert_not_called() + + async def test_run_stream_emits_span_when_instrumentation_enabled(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test that run(stream=True) creates a span when instrumentation is enabled.""" + from agent_framework.observability import OBSERVABILITY_SETTINGS + + messages = self._create_standard_messages() + mock_client = self._create_mock_client(messages) + + monkeypatch.setattr(OBSERVABILITY_SETTINGS, "enable_instrumentation", True) + + with ( + patch("agent_framework_claude._agent.ClaudeSDKClient", return_value=mock_client), + patch("agent_framework.observability.get_tracer") as mock_get_tracer, + ): + mock_span = MagicMock() + mock_tracer = MagicMock() + mock_tracer.start_span.return_value = mock_span + mock_get_tracer.return_value = mock_tracer + + agent = ClaudeAgent(name="stream-agent") + updates: list[AgentResponseUpdate] = [] + async for update in agent.run("Hello", stream=True): + updates.append(update) + + assert len(updates) == 1 + mock_tracer.start_span.assert_called_once() + span_name = mock_tracer.start_span.call_args[0][0] + assert "stream-agent" in span_name + assert "invoke_agent" in span_name + + async def test_run_captures_exception_in_span(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test that exceptions during run() are captured in the telemetry span.""" + from agent_framework.exceptions import AgentException + from agent_framework.observability import OBSERVABILITY_SETTINGS + from claude_agent_sdk import ResultMessage + + error_messages = [ + ResultMessage( + subtype="error", + duration_ms=100, + duration_api_ms=50, + is_error=True, + num_turns=0, + session_id="error-session", + result="Model not found", + ), + ] + mock_client = self._create_mock_client(error_messages) + + monkeypatch.setattr(OBSERVABILITY_SETTINGS, "enable_instrumentation", True) + + with ( + patch("agent_framework_claude._agent.ClaudeSDKClient", return_value=mock_client), + patch("agent_framework.observability._get_span") as mock_get_span, + patch("agent_framework.observability.capture_exception") as mock_capture_exc, + ): + mock_span = MagicMock() + mock_get_span.return_value.__enter__ = MagicMock(return_value=mock_span) + mock_get_span.return_value.__exit__ = MagicMock(return_value=False) + + agent = ClaudeAgent(name="error-agent") + with pytest.raises(AgentException): + await agent.run("Hello") + + mock_capture_exc.assert_called_once() + exc_kwargs = mock_capture_exc.call_args[1] + assert exc_kwargs["span"] is mock_span + assert isinstance(exc_kwargs["exception"], AgentException) + + async def test_telemetry_uses_correct_provider_name(self, monkeypatch: pytest.MonkeyPatch) -> None: + """Test that telemetry uses AGENT_PROVIDER_NAME as provider.""" + from agent_framework.observability import OBSERVABILITY_SETTINGS + + messages = self._create_standard_messages() + mock_client = self._create_mock_client(messages) + + monkeypatch.setattr(OBSERVABILITY_SETTINGS, "enable_instrumentation", True) + + with ( + patch("agent_framework_claude._agent.ClaudeSDKClient", return_value=mock_client), + patch("agent_framework.observability._get_span") as mock_get_span, + ): + mock_span = MagicMock() + mock_get_span.return_value.__enter__ = MagicMock(return_value=mock_span) + mock_get_span.return_value.__exit__ = MagicMock(return_value=False) + + agent = ClaudeAgent(name="test-agent") + await agent.run("Hello") + + call_kwargs = mock_get_span.call_args[1] + assert call_kwargs["attributes"]["gen_ai.provider.name"] == "anthropic.claude" diff --git a/python/packages/core/agent_framework/anthropic/__init__.py b/python/packages/core/agent_framework/anthropic/__init__.py index 242554cf16..8be2a7d208 100644 --- a/python/packages/core/agent_framework/anthropic/__init__.py +++ b/python/packages/core/agent_framework/anthropic/__init__.py @@ -11,6 +11,7 @@ Supported classes: - AnthropicChatOptions - ClaudeAgent - ClaudeAgentOptions +- RawClaudeAgent """ import importlib @@ -21,6 +22,7 @@ _IMPORTS: dict[str, tuple[str, str]] = { "AnthropicChatOptions": ("agent_framework_anthropic", "agent-framework-anthropic"), "ClaudeAgent": ("agent_framework_claude", "agent-framework-claude"), "ClaudeAgentOptions": ("agent_framework_claude", "agent-framework-claude"), + "RawClaudeAgent": ("agent_framework_claude", "agent-framework-claude"), }