diff --git a/docs/decisions/0040-python-foundry-trace-attribution.md b/docs/decisions/0040-python-foundry-trace-attribution.md new file mode 100644 index 00000000000..b45acef3f2f --- /dev/null +++ b/docs/decisions/0040-python-foundry-trace-attribution.md @@ -0,0 +1,60 @@ +--- +status: proposed +contact: jpalvarezl +date: 2026-09-09 +deciders: [eavanvalkenburg, moonbox3] +--- + +# Separate Foundry trace attribution from exporter configuration + +## Context and Problem Statement + +[Issue #7492](https://github.com/microsoft/agent-framework/issues/7492) reports +Foundry client spans reaching Application Insights but missing from the Foundry +agent trace view. Exporting telemetry and identifying the project/agent an +operation belongs to are separate concerns, yet the initial fix resolved project +identity only through `FoundryAgent.configure_azure_monitor()`. That leaves +applications configuring their own exporters without an attribution path. How +should a `FoundryAgent` obtain its project identity independently of how +telemetry is exported? + +## Decision Drivers + +- Support application-managed exporters without reconfiguring global providers. +- Keep identity scoped to the agent/project rather than a process-wide setting. +- Avoid implicit network work in the invocation path. +- Do not generalize one successful span arrangement into a universal requirement + that every application root carry project attributes. + +## Considered Options + +- **Helper-only discovery.** Resolve identity solely inside + `configure_azure_monitor()`. Simple, but couples identity to one exporter + helper and leaves application-managed configurations unsupported. +- **Automatic discovery on each run.** Convenient, but introduces implicit + network work, latency, and failure handling into the invocation path. +- **Explicit per-agent identity plus cached helper discovery.** Accept the + project ARM ID on the agent; the helper discovers and caches it only when none + was supplied. + +## Decision Outcome + +Chosen option: **explicit per-agent identity plus cached helper discovery**, +because it supports both setup styles without run-time discovery. + +- Add keyword-only `project_arm_id` to `RawFoundryAgent` and `FoundryAgent`, + validated as a full project ARM ID at construction; it is not inferred from the + data-plane endpoint or read implicitly from an environment variable. +- The Azure Monitor helper discovers and caches identity only when no ID was + supplied. Expected discovery/metadata errors log a warning and preserve export; + unexpected programming errors and cancellation still propagate. +- Identity is emitted on the agent's `invoke_agent` span, which may be nested + beneath an application span, keeping attribution separate from exporter + configuration. + +The public SDK does not yet expose project identity directly +([Azure/azure-sdk-for-python#48825](https://github.com/Azure/azure-sdk-for-python/issues/48825)). +Connection-ID parsing remains a bounded discovery workaround, not a requirement +for applications that already know their project ARM ID. Supported examples use +full project ARM IDs; alternate formats and service-side legacy-key behavior are +not new guarantees of this API. diff --git a/python/packages/core/agent_framework/_agents.py b/python/packages/core/agent_framework/_agents.py index a115f600f3d..3f70e3e8fb8 100644 --- a/python/packages/core/agent_framework/_agents.py +++ b/python/packages/core/agent_framework/_agents.py @@ -64,7 +64,10 @@ normalize_messages, ) from .exceptions import AgentInvalidRequestException, AgentInvalidResponseException, UserInputRequiredException -from .observability import AgentTelemetryLayer +from .observability import ( + AgentTelemetryLayer, + _capture_agent_response_id, # pyright: ignore[reportPrivateUsage] +) if sys.version_info >= (3, 13): from typing import TypeVar # pragma: no cover @@ -1219,6 +1222,7 @@ async def _parse_non_streaming_response( if not response: raise AgentInvalidResponseException("Chat client did not return a response.") + _capture_agent_response_id(self, response) for message in response.messages: if message.author_name is None: message.author_name = context["agent_name"] @@ -1290,7 +1294,9 @@ def _finalizer(updates: Sequence[AgentResponseUpdate]) -> AgentResponse[Any]: response_format=context["chat_options"].get("response_format"), ) - stream = stream_response.map( + # Capture the completed chat operation, including finalizer-only metadata, + # before mapping to the public response and invoking after-run providers. + stream = stream_response.with_result_hook(partial(_capture_agent_response_id, self)).map( transform=partial( map_chat_to_agent_update, agent_name=self.name, diff --git a/python/packages/core/agent_framework/observability.py b/python/packages/core/agent_framework/observability.py index 5dc9fa8828c..a4418974540 100644 --- a/python/packages/core/agent_framework/observability.py +++ b/python/packages/core/agent_framework/observability.py @@ -23,6 +23,7 @@ import sys import weakref from collections.abc import Awaitable, Callable, Generator, Mapping, Sequence +from dataclasses import dataclass, field from enum import Enum from time import perf_counter, time_ns from typing import ( @@ -119,15 +120,21 @@ otel_event_logger = get_otel_logger("agent_framework", version_info) -INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS: Final[contextvars.ContextVar[set[str] | None]] = contextvars.ContextVar( - "inner_response_telemetry_captured_fields", default=None -) INNER_RESPONSE_ID_CAPTURED_FIELD: Final[str] = "response_id" INNER_USAGE_CAPTURED_FIELD: Final[str] = "usage" -# Tracks accumulated token usage from all inner chat completion spans within an agent invoke. -INNER_ACCUMULATED_USAGE: Final[contextvars.ContextVar[UsageDetails | None]] = contextvars.ContextVar( - "inner_accumulated_usage", default=None + +@dataclass +class _InnerResponseTelemetryState: + owner: object + captured_fields: set[str] = field(default_factory=set[str]) + response_id: str | None = None + response_id_recorded: bool = False + accumulated_usage: UsageDetails = field(default_factory=lambda: cast("UsageDetails", {})) + + +_INNER_RESPONSE_TELEMETRY_STATE: Final[contextvars.ContextVar[_InnerResponseTelemetryState | None]] = ( + contextvars.ContextVar("inner_response_telemetry_state", default=None) ) # Allows protocol adapters to supply an application-managed conversation identity for one execution @@ -2290,6 +2297,14 @@ def __init__( self.token_usage_histogram = _get_token_usage_histogram() self.duration_histogram = _get_duration_histogram() + def _get_additional_otel_agent_attributes(self) -> Mapping[str, Any]: + """Return provider-specific attributes emitted on agent spans.""" + return {} + + def _should_capture_agent_response_id(self) -> bool: + """Return whether the agent span must retain an inner response ID.""" + return False + def _trace_agent_invocation( self, *, @@ -2333,18 +2348,18 @@ def _trace_agent_invocation( all_options=dict(merged_options), **merged_client_kwargs, ) + attributes.update(self._get_additional_otel_agent_attributes()) if stream: - # Do NOT set the inner-telemetry context vars here: this synchronous run() body executes + # Do NOT set the inner-telemetry context var here: this synchronous run() body executes # in the CALLER's context, but the ResponseStream may be consumed in a different context # (e.g. ``stream = agent.run(stream=True)`` then ``await asyncio.create_task(consume(stream))``). # The cleanup-hook reset (in _finalize_stream) runs in the consuming context, so a token # created here would raise ``ValueError: was created in a different Context``. - # Instead the tokens are set lazily on the first pull (see _inner_telemetry_pull_context + # Instead the token is set lazily on the first pull (see _inner_telemetry_pull_context # below), so set and reset both happen in the consumer's context. - inner_response_telemetry_captured_fields: set[str] = set() - inner_response_telemetry_captured_fields_token: contextvars.Token[set[str] | None] | None = None - inner_accumulated_usage_token: contextvars.Token[UsageDetails | None] | None = None + inner_telemetry_state = _InnerResponseTelemetryState(owner=self) + inner_telemetry_token: contextvars.Token[_InnerResponseTelemetryState | None] | None = None # Agent Framework's agents run in-process (the actual network call happens on a nested # chat span), so invoke_agent spans use the default INTERNAL kind. span = _start_streaming_span(attributes, OtelAttr.AGENT_NAME) @@ -2411,11 +2426,15 @@ async def _finalize_stream() -> None: response_attributes = _get_response_attributes( attributes, response, - capture_response_id=INNER_RESPONSE_ID_CAPTURED_FIELD - not in inner_response_telemetry_captured_fields, - capture_usage=INNER_USAGE_CAPTURED_FIELD not in inner_response_telemetry_captured_fields, + capture_response_id=( + self._should_capture_agent_response_id() + or INNER_RESPONSE_ID_CAPTURED_FIELD not in inner_telemetry_state.captured_fields + ), + capture_usage=INNER_USAGE_CAPTURED_FIELD not in inner_telemetry_state.captured_fields, ) - _apply_accumulated_usage(response_attributes, inner_response_telemetry_captured_fields) + if self._should_capture_agent_response_id(): + _apply_captured_response_id(response_attributes, inner_telemetry_state) + _apply_accumulated_usage(response_attributes, inner_telemetry_state) _capture_response(span=span, attributes=response_attributes, duration=duration) if ( OBSERVABILITY_SETTINGS.SENSITIVE_DATA_ENABLED @@ -2433,33 +2452,28 @@ async def _finalize_stream() -> None: finally: # Reset only if the lazy set actually ran (it may not have if the stream was # never pulled). These run in the consuming context — the same context the - # pull-context factory below set the tokens in — so the reset is cross-context safe. - if inner_response_telemetry_captured_fields_token is not None: - INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.reset(inner_response_telemetry_captured_fields_token) - if inner_accumulated_usage_token is not None: - INNER_ACCUMULATED_USAGE.reset(inner_accumulated_usage_token) + # pull-context factory below set the token in — so the reset is cross-context safe. + if inner_telemetry_token is not None: + _INNER_RESPONSE_TELEMETRY_STATE.reset(inner_telemetry_token) _close_span() def _inner_telemetry_pull_context() -> contextlib.AbstractContextManager[Any]: # Invoked at the start of every pull (and during stream resolution), in the - # consuming context. On the first pull it sets the inner-telemetry context vars so + # consuming context. On the first pull it sets the inner-telemetry context var so # that set and the reset in _finalize_stream both run in the consumer's context, # avoiding the cross-context Token reset failure. Setting happens before the # underlying iterator is pulled, so inner chat completion spans created during the # pull can still accumulate usage / mark captured fields. - nonlocal inner_response_telemetry_captured_fields_token, inner_accumulated_usage_token - if inner_response_telemetry_captured_fields_token is None: - inner_response_telemetry_captured_fields_token = INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.set( - inner_response_telemetry_captured_fields - ) - inner_accumulated_usage_token = INNER_ACCUMULATED_USAGE.set({}) + nonlocal inner_telemetry_token + if inner_telemetry_token is None: + inner_telemetry_token = _INNER_RESPONSE_TELEMETRY_STATE.set(inner_telemetry_state) return _activate_span(span) # The pull context manager attaches the span around each underlying iterator pull so # that child spans created during the pull (e.g. inner chat completion spans from the # underlying ChatTelemetryLayer) are parented under this agent invoke span. Attach and # detach happen in the same async context as the pull, avoiding cross-context cleanup - # issues. It also lazily sets the inner-telemetry context vars on the first pull (see + # issues. It also lazily sets the inner-telemetry context var on the first pull (see # _inner_telemetry_pull_context). The weakref finalizer ensures the span is closed even # if the stream is garbage collected without being consumed. wrapped_stream: ResponseStream[AgentResponseUpdate, AgentResponse[Any]] = ( @@ -2472,18 +2486,15 @@ def _inner_telemetry_pull_context() -> contextlib.AbstractContextManager[Any]: return wrapped_stream async def _run() -> AgentResponse[Any]: - # Set the inner-telemetry context vars inside the coroutine so the set and the + # Set the inner-telemetry context var inside the coroutine so the set and the # reset in `finally` always happen in the same execution context. `run()` is a sync # method that returns this coroutine, which may be awaited in a different context than # the one that called `run()` (e.g. `asyncio.create_task(agent.run(...))`, as used by # BackgroundAgentsProvider). A contextvars.Token can only be reset in the context that # created it, so setting eagerly in `run()`/`_trace_agent_invocation` and resetting # here would raise "Token was created in a different Context". - inner_response_telemetry_captured_fields: set[str] = set() - inner_response_telemetry_captured_fields_token = INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.set( - inner_response_telemetry_captured_fields - ) - inner_accumulated_usage_token = INNER_ACCUMULATED_USAGE.set({}) + inner_telemetry_state = _InnerResponseTelemetryState(owner=self) + inner_telemetry_token = _INNER_RESPONSE_TELEMETRY_STATE.set(inner_telemetry_state) try: with _get_span(attributes=attributes, span_name_attribute=OtelAttr.AGENT_NAME) as span: try: @@ -2500,15 +2511,17 @@ async def _run() -> AgentResponse[Any]: response_attributes = _get_response_attributes( attributes, response, - capture_response_id=INNER_RESPONSE_ID_CAPTURED_FIELD - not in inner_response_telemetry_captured_fields, - capture_usage=( - INNER_USAGE_CAPTURED_FIELD not in inner_response_telemetry_captured_fields + capture_response_id=( + self._should_capture_agent_response_id() + or INNER_RESPONSE_ID_CAPTURED_FIELD not in inner_telemetry_state.captured_fields ), + capture_usage=(INNER_USAGE_CAPTURED_FIELD not in inner_telemetry_state.captured_fields), ) + if self._should_capture_agent_response_id(): + _apply_captured_response_id(response_attributes, inner_telemetry_state) _apply_accumulated_usage( response_attributes, - inner_response_telemetry_captured_fields, + inner_telemetry_state, ) _capture_response( span=span, @@ -2530,8 +2543,7 @@ async def _run() -> AgentResponse[Any]: capture_exception(span=span, exception=exception, timestamp=time_ns()) raise finally: - INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.reset(inner_response_telemetry_captured_fields_token) - INNER_ACCUMULATED_USAGE.reset(inner_accumulated_usage_token) + _INNER_RESPONSE_TELEMETRY_STATE.reset(inner_telemetry_token) return _run() @@ -3434,28 +3446,45 @@ def _mark_inner_response_telemetry_captured( response: ChatResponse | AgentResponse, ) -> None: """Record when an inner chat telemetry span already captured response metadata.""" - captured_fields = INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.get() - if captured_fields is None: + state = _INNER_RESPONSE_TELEMETRY_STATE.get() + if state is None: return if response.response_id: - captured_fields.add(INNER_RESPONSE_ID_CAPTURED_FIELD) + state.captured_fields.add(INNER_RESPONSE_ID_CAPTURED_FIELD) if response.usage_details: - captured_fields.add(INNER_USAGE_CAPTURED_FIELD) - accumulated = INNER_ACCUMULATED_USAGE.get() - if accumulated is not None: - from ._types import add_usage_details + from ._types import add_usage_details + + state.captured_fields.add(INNER_USAGE_CAPTURED_FIELD) + state.accumulated_usage = add_usage_details(state.accumulated_usage, response.usage_details) + - INNER_ACCUMULATED_USAGE.set(add_usage_details(accumulated, response.usage_details)) +def _capture_agent_response_id(owner: object, response: ChatResponse[Any]) -> None: # pyright: ignore[reportUnusedFunction] + """Record the owned chat result before conversion, suppression, and after-run callbacks.""" + state = _INNER_RESPONSE_TELEMETRY_STATE.get() + # Raw agents do not establish their own telemetry state, so a nested raw agent + # must not replace the enclosing instrumented agent's response identity. + if state is not None and state.owner is owner: + state.response_id = response.response_id + state.response_id_recorded = True + + +def _apply_captured_response_id(attributes: dict[str, Any], state: _InnerResponseTelemetryState) -> None: + """Apply the owned chat response ID to an agent span when the provider requires it.""" + if not state.response_id_recorded: + return + if state.response_id: + attributes[OtelAttr.RESPONSE_ID] = state.response_id + else: + attributes.pop(OtelAttr.RESPONSE_ID, None) -def _apply_accumulated_usage(attributes: dict[str, Any], captured_fields: set[str]) -> None: +def _apply_accumulated_usage(attributes: dict[str, Any], state: _InnerResponseTelemetryState) -> None: """Apply accumulated usage from inner chat spans to the invoke_agent span attributes.""" - if INNER_USAGE_CAPTURED_FIELD not in captured_fields: + if INNER_USAGE_CAPTURED_FIELD not in state.captured_fields: return - accumulated = INNER_ACCUMULATED_USAGE.get() - if not accumulated: + if not state.accumulated_usage: return - _apply_usage_attributes(attributes, accumulated) + _apply_usage_attributes(attributes, state.accumulated_usage) def _apply_usage_attributes(attributes: dict[str, Any], usage: Mapping[str, Any]) -> None: diff --git a/python/packages/core/tests/core/test_observability.py b/python/packages/core/tests/core/test_observability.py index f496e44d07f..5b49f3623ca 100644 --- a/python/packages/core/tests/core/test_observability.py +++ b/python/packages/core/tests/core/test_observability.py @@ -3,7 +3,7 @@ import asyncio import logging from collections.abc import AsyncIterable, Awaitable, Mapping, Sequence -from typing import Any, cast +from typing import Any, Literal, cast from unittest.mock import Mock, patch import pytest @@ -15,14 +15,17 @@ AGENT_FRAMEWORK_USER_AGENT, Agent, AgentResponse, + AgentSession, BaseChatClient, ChatResponse, ChatResponseUpdate, Content, ContextProvider, + InMemoryHistoryProvider, Message, RawAgent, ResponseStream, + SessionContext, SupportsAgentRun, UsageDetails, prepend_agent_framework_to_user_agent, @@ -4865,10 +4868,17 @@ async def _get() -> ChatResponse: @pytest.mark.parametrize("stream", [False, True]) -async def test_agent_and_chat_spans_do_not_duplicate_response_telemetry( - span_exporter: InMemorySpanExporter, stream: bool +@pytest.mark.parametrize("retain_agent_response_id", [False, True]) +@pytest.mark.parametrize("chunk_response_id", [None, "chunk_resp"]) +@pytest.mark.parametrize("final_response_id", [None, "nested_resp_123"]) +async def test_agent_provider_hooks_control_response_telemetry( + span_exporter: InMemorySpanExporter, + stream: bool, + retain_agent_response_id: bool, + chunk_response_id: str | None, + final_response_id: str | None, ): - """The inner chat span owns response-id; usage is aggregated on the agent span.""" + """Provider hooks can add root attributes and retain an inner response ID when required.""" class NestedTelemetryChatClient(ChatTelemetryLayer, BaseChatClient[Any]): def service_url(self): @@ -4885,13 +4895,17 @@ def _inner_get_response( # pyrefly: ignore[bad-override] if stream: async def _stream() -> AsyncIterable[ChatResponseUpdate]: - yield ChatResponseUpdate(contents=[Content.from_text("Nested")], role="assistant") - yield ChatResponseUpdate(contents=[Content.from_text(" response")], role="assistant") + yield ChatResponseUpdate( + contents=[Content.from_text("Nested")], role="assistant", response_id=chunk_response_id + ) + yield ChatResponseUpdate( + contents=[Content.from_text(" response")], role="assistant", response_id=chunk_response_id + ) def _finalize(updates: Sequence[ChatResponseUpdate]) -> ChatResponse: return ChatResponse( messages=[Message(role="assistant", contents=["Nested response"])], - response_id="nested_resp_123", + response_id=final_response_id, usage_details=UsageDetails(input_token_count=11, output_token_count=22), finish_reason="stop", ) @@ -4901,14 +4915,21 @@ def _finalize(updates: Sequence[ChatResponseUpdate]) -> ChatResponse: async def _get() -> ChatResponse: return ChatResponse( messages=[Message(role="assistant", contents=["Nested response"])], - response_id="nested_resp_123", + response_id=final_response_id, usage_details=UsageDetails(input_token_count=11, output_token_count=22), finish_reason="stop", ) return _get() - agent = Agent( + class ProviderTelemetryAgent(Agent): + def _get_additional_otel_agent_attributes(self) -> Mapping[str, Any]: + return {"test.provider.attribute": "provider-value"} + + def _should_capture_agent_response_id(self) -> bool: + return retain_agent_response_id + + agent = ProviderTelemetryAgent( client=NestedTelemetryChatClient(), # ty: ignore[invalid-argument-type] id="nested_agent_id", name="nested_agent", @@ -4935,16 +4956,299 @@ async def _get() -> ChatResponse: agent_span = span_by_operation[OtelAttr.AGENT_INVOKE_OPERATION] chat_span = span_by_operation[OtelAttr.CHAT_COMPLETION_OPERATION] - assert chat_span.attributes[OtelAttr.RESPONSE_ID] == "nested_resp_123" # type: ignore[index] # pyrefly: ignore[unsupported-operation] # ty: ignore[not-subscriptable] + assert chat_span.attributes is not None + assert chat_span.attributes.get(OtelAttr.RESPONSE_ID) == final_response_id assert chat_span.attributes[OtelAttr.INPUT_TOKENS] == 11 # type: ignore[index] # pyrefly: ignore[unsupported-operation] # ty: ignore[not-subscriptable] assert chat_span.attributes[OtelAttr.OUTPUT_TOKENS] == 22 # type: ignore[index] # pyrefly: ignore[unsupported-operation] # ty: ignore[not-subscriptable] - assert OtelAttr.RESPONSE_ID not in agent_span.attributes # type: ignore[operator] # pyrefly: ignore[not-iterable] # ty: ignore[unsupported-operator] + assert agent_span.attributes["test.provider.attribute"] == "provider-value" # type: ignore[index] # pyrefly: ignore[unsupported-operation] # ty: ignore[not-subscriptable] + assert agent_span.attributes is not None + expected_id = ( + final_response_id + if retain_agent_response_id + else chunk_response_id + if stream and not final_response_id + else None + ) + assert agent_span.attributes.get(OtelAttr.RESPONSE_ID) == expected_id # The agent span carries the aggregated usage from all inner chat completions assert agent_span.attributes[OtelAttr.INPUT_TOKENS] == 11 # type: ignore[index] # pyrefly: ignore[unsupported-operation] # ty: ignore[not-subscriptable] assert agent_span.attributes[OtelAttr.OUTPUT_TOKENS] == 22 # type: ignore[index] # pyrefly: ignore[unsupported-operation] # ty: ignore[not-subscriptable] +@pytest.mark.parametrize("stream", [False, True]) +@pytest.mark.parametrize("callback_kind", ["same_client", "other_client", "agent", "raw_agent"]) +@pytest.mark.parametrize("final_response_id", ["owned-final", None]) +async def test_agent_response_identity_ignores_provider_callbacks( + span_exporter: InMemorySpanExporter, + stream: bool, + callback_kind: str, + final_response_id: str | None, +) -> None: + """Only the completed owned operation supplies identity; all direct child chats supply usage.""" + from agent_framework._middleware import ChatMiddlewareLayer + from agent_framework._tools import FunctionInvocationLayer + + calls: list[str] = [] + tool_calls: list[str] = [] + + @tool + async def lookup() -> str: + """Look up the requested value.""" + tool_calls.append("lookup") + return "found" + + class IdentityClient(FunctionInvocationLayer, ChatMiddlewareLayer, ChatTelemetryLayer, BaseChatClient[Any]): + def service_url(self) -> str: + return "https://test.example.com" + + def _inner_get_response( # pyrefly: ignore[bad-override] + self, + *, + messages: Sequence[Message], + stream: bool, + options: Mapping[str, Any], + **kwargs: Any, # type: ignore[override] + ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]: + label = next(message.text for message in reversed(messages) if message.role == "user") + is_tool_call = label == "owned" and not any(message.role == "tool" for message in messages) + response_id = ( + ("owned-first" if final_response_id else None) + if is_tool_call + else final_response_id + if label == "owned" + else label + ) + contents = ( + [Content.from_function_call(call_id="lookup-call", name="lookup", arguments="{}")] + if is_tool_call + else [Content.from_text(f"{label} result")] + ) + + async def _get() -> ChatResponse: + await asyncio.sleep(0) + calls.append(response_id or "no-id") + return ChatResponse( + messages=[Message("assistant", contents)], + response_id=response_id, + usage_details=UsageDetails(input_token_count=11, output_token_count=22), + ) + + if stream: + + async def _stream() -> AsyncIterable[ChatResponseUpdate]: + response = await _get() + assert response.usage_details is not None + yield ChatResponseUpdate( + role="assistant", + contents=[*contents, Content.from_usage(response.usage_details)], + response_id=response_id, + ) + + return ResponseStream(_stream(), finalizer=ChatResponse.from_updates) + return _get() + + class IdentityAgent(Agent): + def _should_capture_agent_response_id(self) -> bool: + return True + + client = IdentityClient() + callback_client = client if callback_kind == "same_client" else IdentityClient() + callback_agent = ( + IdentityAgent(client=callback_client, name="callback") + if callback_kind == "agent" + else RawAgent(client=callback_client, name="callback") + ) + + async def invoke_callback(label: str) -> None: + if callback_kind in {"agent", "raw_agent"}: + if stream: + await callback_agent.run(label, stream=True).get_final_response() + else: + await callback_agent.run(label) + elif stream: + await callback_client.get_response([Message("user", [label])], stream=True).get_final_response() + else: + await callback_client.get_response([Message("user", [label])]) + + class CallbackProvider(ContextProvider): + def __init__(self) -> None: + super().__init__(source_id="callback") + + async def before_run( + self, *, agent: SupportsAgentRun, session: AgentSession, context: SessionContext, state: dict[str, Any] + ) -> None: + await invoke_callback("before") + + async def after_run( + self, *, agent: SupportsAgentRun, session: AgentSession, context: SessionContext, state: dict[str, Any] + ) -> None: + assert context.response is not None + assert context.response.response_id is None + await invoke_callback("after") + + history = InMemoryHistoryProvider() + agent = IdentityAgent( + client=client, + name="owner", + context_providers=[history, CallbackProvider()], + require_per_service_call_history_persistence=True, + tools=[lookup], + ) + session = AgentSession() + span_exporter.clear() + if stream: + result = agent.run("owned", stream=True, session=session, options={"store": False}) + updates = [update async for update in result] + assert all(update.response_id is None for update in updates) + response = await result.get_final_response() + else: + response = await agent.run("owned", session=session, options={"store": False}) + + assert response.response_id is None + assert session.service_session_id is None + assert calls == ["before", "owned-first" if final_response_id else "no-id", final_response_id or "no-id", "after"] + assert tool_calls == ["lookup"] + assert response.text == "owned result" + saved = await history.get_messages(session.session_id, state=session.state[history.source_id]) + assert [content.type for message in saved for content in message.contents] == [ + "text", + "function_call", + "function_result", + "text", + ] + + agent_spans = [ + span + for span in span_exporter.get_finished_spans() + if span.attributes and span.attributes.get(OtelAttr.AGENT_NAME) == "owner" + ] + assert len(agent_spans) == 1 + attributes = agent_spans[0].attributes + assert attributes is not None + assert attributes.get(OtelAttr.RESPONSE_ID) == final_response_id + chat_count = 2 if callback_kind == "agent" else 4 + assert attributes[OtelAttr.INPUT_TOKENS] == 11 * chat_count + assert attributes[OtelAttr.OUTPUT_TOKENS] == 22 * chat_count + + +@pytest.mark.parametrize("stream", [False, True]) +@pytest.mark.parametrize("execution", ["nested", "concurrent"]) +async def test_agent_response_telemetry_state_isolated_per_invocation( + span_exporter: InMemorySpanExporter, + stream: bool, + execution: Literal["nested", "concurrent"], +) -> None: + """Re-entering one agent and consuming its streams in other tasks never shares invocation state.""" + from agent_framework.observability import _INNER_RESPONSE_TELEMETRY_STATE, _InnerResponseTelemetryState + + ready = {label: asyncio.Event() for label in ("first", "second")} + + class IdentityClient(ChatTelemetryLayer, BaseChatClient[Any]): + def service_url(self) -> str: + return "https://test.example.com" + + def _inner_get_response( # pyrefly: ignore[bad-override] + self, + *, + messages: Sequence[Message], + stream: bool, + options: Mapping[str, Any], + **kwargs: Any, # type: ignore[override] + ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]: + label = messages[-1].text + usage = 1 if label == "first" else 10 + + async def _get() -> ChatResponse: + if execution == "concurrent": + ready[label].set() + await ready["second" if label == "first" else "first"].wait() + return ChatResponse( + messages=[Message("assistant", [label])], + response_id=label, + usage_details=UsageDetails(input_token_count=usage, output_token_count=2 * usage), + ) + + if stream: + response: ChatResponse | None = None + + async def _stream() -> AsyncIterable[ChatResponseUpdate]: + nonlocal response + response = await _get() + yield ChatResponseUpdate(role="assistant", contents=[Content.from_text(label)]) + + def _finalize(updates: Sequence[ChatResponseUpdate]) -> ChatResponse: + assert response is not None + return response + + return ResponseStream(_stream(), finalizer=_finalize) + return _get() + + class IdentityAgent(Agent): + def _should_capture_agent_response_id(self) -> bool: + return True + + class ReenterProvider(ContextProvider): + def __init__(self) -> None: + super().__init__(source_id="reenter") + + async def after_run( + self, *, agent: SupportsAgentRun, session: AgentSession, context: SessionContext, state: dict[str, Any] + ) -> None: + assert context.response is not None + if execution == "nested" and context.response.text == "first": + await consume("second") + + agent = IdentityAgent( + client=IdentityClient(), # ty: ignore[invalid-argument-type] + name="shared", + context_providers=[InMemoryHistoryProvider(), ReenterProvider()], + require_per_service_call_history_persistence=True, + ) + + async def consume(label: str) -> AgentResponse: + previous = _INNER_RESPONSE_TELEMETRY_STATE.get() + if stream: + response = await agent.run(label, stream=True).get_final_response() + else: + response = await agent.run(label) + assert _INNER_RESPONSE_TELEMETRY_STATE.get() is previous + assert response.response_id is None + return response + + sentinel = _InnerResponseTelemetryState( + owner=object(), + captured_fields={"sentinel"}, + response_id="sentinel", + accumulated_usage=UsageDetails(input_token_count=100), + ) + token = _INNER_RESPONSE_TELEMETRY_STATE.set(sentinel) + span_exporter.clear() + try: + if execution == "concurrent": + await asyncio.gather(consume("first"), consume("second")) + else: + await consume("first") + assert _INNER_RESPONSE_TELEMETRY_STATE.get() is sentinel + assert sentinel.captured_fields == {"sentinel"} + assert sentinel.response_id == "sentinel" + assert sentinel.accumulated_usage == {"input_token_count": 100} + finally: + _INNER_RESPONSE_TELEMETRY_STATE.reset(token) + + agent_attributes = [ + span.attributes + for span in span_exporter.get_finished_spans() + if span.attributes and span.attributes.get(OtelAttr.OPERATION) == OtelAttr.AGENT_INVOKE_OPERATION + ] + assert len(agent_attributes) == 2 + by_id = {attributes[OtelAttr.RESPONSE_ID]: attributes for attributes in agent_attributes} + assert by_id["first"][OtelAttr.INPUT_TOKENS] == 1 + assert by_id["first"][OtelAttr.OUTPUT_TOKENS] == 2 + assert by_id["second"][OtelAttr.INPUT_TOKENS] == 10 + assert by_id["second"][OtelAttr.OUTPUT_TOKENS] == 20 + + # region Test non-ASCII character handling in JSON serialization @@ -6734,8 +7038,8 @@ async def test_agent_streaming_execute_failure_closes_span_and_resets_contextvar """If ``execute()`` raises synchronously during streaming agent invocation, the agent span is ended, the exception is recorded, and the telemetry contextvars are reset.""" from agent_framework.observability import ( - INNER_ACCUMULATED_USAGE, - INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS, + _INNER_RESPONSE_TELEMETRY_STATE, + _InnerResponseTelemetryState, ) class _FailingExecuteAgent: @@ -6772,10 +7076,8 @@ class FailingExecuteAgent(AgentTelemetryLayer, _FailingExecuteAgent): # type: i pass # Sentinel values to detect that contextvars were reset to their pre-call state. - sentinel_fields: set[str] = set() - sentinel_usage: dict[str, Any] = {} - fields_token = INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.set(sentinel_fields) - usage_token = INNER_ACCUMULATED_USAGE.set(sentinel_usage) # type: ignore[arg-type] # pyrefly: ignore[bad-argument-type] # ty: ignore[invalid-argument-type] + sentinel = _InnerResponseTelemetryState(owner=object()) + token = _INNER_RESPONSE_TELEMETRY_STATE.set(sentinel) try: agent = FailingExecuteAgent() span_exporter.clear() @@ -6783,11 +7085,9 @@ class FailingExecuteAgent(AgentTelemetryLayer, _FailingExecuteAgent): # type: i agent.run(messages="Hello", stream=True) # Contextvars must be back to the sentinel values registered before the call. - assert INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.get() is sentinel_fields - assert INNER_ACCUMULATED_USAGE.get() is sentinel_usage + assert _INNER_RESPONSE_TELEMETRY_STATE.get() is sentinel finally: - INNER_ACCUMULATED_USAGE.reset(usage_token) - INNER_RESPONSE_TELEMETRY_CAPTURED_FIELDS.reset(fields_token) + _INNER_RESPONSE_TELEMETRY_STATE.reset(token) spans = span_exporter.get_finished_spans() agent_spans = [s for s in spans if s.attributes.get(OtelAttr.OPERATION.value) == OtelAttr.AGENT_INVOKE_OPERATION] # type: ignore[union-attr] # ty: ignore[unresolved-attribute] diff --git a/python/packages/foundry/README.md b/python/packages/foundry/README.md index 1bc0051fec0..9ea9a976ed9 100644 --- a/python/packages/foundry/README.md +++ b/python/packages/foundry/README.md @@ -2,6 +2,73 @@ This package contains the Microsoft Foundry integrations for Microsoft Agent Framework, including Foundry chat clients, preconfigured Foundry agents, Foundry embedding clients, and Foundry memory providers. +## Tracing an existing Foundry agent + +Telemetry export and project attribution are separate concerns. For an existing +prompt or hosted agent, `FoundryAgent.configure_azure_monitor()` configures the +exporter and, unless supplied explicitly, discovers the project's ARM resource +ID from its Application Insights connection. Discovery is cached per agent. + +```python +import os + +from agent_framework.foundry import FoundryAgent +from azure.identity.aio import AzureCliCredential + +async with AzureCliCredential() as credential, FoundryAgent( + project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"], + agent_name=os.environ["FOUNDRY_AGENT_NAME"], + credential=credential, +) as agent: + await agent.configure_azure_monitor() + response = await agent.run("Hello!") +``` + +If your application already configures OpenTelemetry providers/exporters, pass +`project_arm_id` when constructing each `FoundryAgent` instead of calling the +helper again. This sets `microsoft.foundry.project.id` on that agent's invocation +span without changing global exporters or making a discovery request: + +```python +import os + +from agent_framework.foundry import FoundryAgent +from azure.identity.aio import AzureCliCredential + +# The application has already configured its OpenTelemetry exporters. +async with AzureCliCredential() as credential, FoundryAgent( + project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"], + project_arm_id=os.environ["FOUNDRY_PROJECT_ARM_ID"], + agent_name=os.environ["FOUNDRY_AGENT_NAME"], + credential=credential, +) as agent: + response = await agent.run("Hello!") +``` + +The sample environment variable is read explicitly by the application, not +automatically by `FoundryAgent`. Use the full **project** ARM ID: +`/subscriptions/{subscription}/resourceGroups/{group}/providers/Microsoft.CognitiveServices/accounts/{account}/projects/{project}`. +The project endpoint and Application Insights connection string are not substitutes. +Keep each agent's project identity consistent with its endpoint; do not use one +process-wide project attribute when agents address different projects. + +If optional discovery fails, the helper logs a warning and still configures Azure +Monitor. Telemetry can reach Application Insights without appearing in Foundry's +project/agent view. Supply the project ARM ID explicitly or repair the connection +metadata before relying on portal attribution. The SDK exposure gap is tracked in +[Azure/azure-sdk-for-python#48825](https://github.com/Azure/azure-sdk-for-python/issues/48825). + +`FoundryAgent` also retains its own chat operation's response ID on the invocation +span, including streaming, without changing public history/continuation behavior. +An unrelated model call from a context provider cannot replace that identity. +The invocation span may itself be a child of an application span; there is no +requirement for your application parent to carry the same attributes. + +See [the existing-agent tracing sample](../../samples/02-agents/observability/foundry_agent_tracing.py) +for helper/manual setup and streaming. This differs from tracing a local +`Agent(client=FoundryChatClient(...))`: calling the chat client's helper does not +initialize identity on a separate `FoundryAgent` instance. + ## Evaluations `FoundryEvals` implements the provider-neutral `Evaluator` protocol with diff --git a/python/packages/foundry/agent_framework_foundry/_agent.py b/python/packages/foundry/agent_framework_foundry/_agent.py index cd44b460887..8c5f1f7bba2 100644 --- a/python/packages/foundry/agent_framework_foundry/_agent.py +++ b/python/packages/foundry/agent_framework_foundry/_agent.py @@ -10,6 +10,7 @@ from __future__ import annotations import logging +import re import sys import warnings from collections.abc import Awaitable, Callable, Mapping, MutableMapping, Sequence @@ -36,8 +37,10 @@ from agent_framework.observability import AgentTelemetryLayer, ChatTelemetryLayer from agent_framework_openai._chat_client import OpenAIChatOptions, RawOpenAIChatClient from azure.ai.projects.aio import AIProjectClient +from azure.ai.projects.models import ConnectionType from azure.core.credentials import TokenCredential from azure.core.credentials_async import AsyncTokenCredential +from azure.core.exceptions import HttpResponseError, ServiceRequestError, ServiceResponseError from agent_framework_foundry._oauth_helpers import try_parse_oauth_consent_event @@ -95,6 +98,22 @@ class FoundryAgentSettings(TypedDict, total=False): FOUNDRY_HOSTED_AGENT_SESSION_ID_KEY = "foundry_hosted_agent_session_id" +_FOUNDRY_PROJECT_ARM_ID_ATTRIBUTE = "microsoft.foundry.project.id" +_FOUNDRY_PROJECT_ARM_ID_PATTERN = re.compile( + r"/subscriptions/[^/]+/resourceGroups/[^/]+/providers/" + r"Microsoft\.CognitiveServices/accounts/[^/]+/projects/[^/]+", + re.IGNORECASE, +) + + +def _validate_project_arm_id(project_arm_id: str) -> str: + if not _FOUNDRY_PROJECT_ARM_ID_PATTERN.fullmatch(project_arm_id): + raise ValueError( + "project_arm_id must be the full Foundry project ARM resource ID: " + "/subscriptions/{subscription}/resourceGroups/{group}/providers/" + "Microsoft.CognitiveServices/accounts/{account}/projects/{project}." + ) + return project_arm_id class FoundryAgentOptions(OpenAIChatOptions, total=False): @@ -665,6 +684,7 @@ def __init__( self, *, project_endpoint: str | None = None, + project_arm_id: str | None = None, agent_name: str | None = None, agent_version: str | None = None, credential: AzureCredentialTypes | None = None, @@ -694,6 +714,10 @@ def __init__( Keyword Args: project_endpoint: The Foundry project endpoint URL. Can also be set via environment variable FOUNDRY_PROJECT_ENDPOINT. + project_arm_id: Full Foundry project ARM resource ID used by ``FoundryAgent`` + for telemetry attribution, independently of exporter configuration. + Supply this when configuring exporters yourself. When omitted, + ``configure_azure_monitor()`` attempts to discover it from project connections. agent_name: The name of the Foundry agent to connect to. Can also be set via environment variable FOUNDRY_AGENT_NAME. agent_version: The version of the agent (required for PromptAgents, optional for HostedAgents). @@ -723,6 +747,7 @@ def __init__( timeout: HTTP timeout in seconds for requests. When not provided, the OpenAI SDK default is used (connect: 5s, total: 600s). """ + self._foundry_project_arm_id = _validate_project_arm_id(project_arm_id) if project_arm_id is not None else None # Create the client actual_client_type = client_type or _FoundryAgentChatClient if not issubclass(actual_client_type, RawFoundryAgentChatClient): @@ -841,6 +866,22 @@ def _update_session_from_chat_response_update( if session is not None and isinstance(agent_session_id, str) and agent_session_id: session.state[FOUNDRY_HOSTED_AGENT_SESSION_ID_KEY] = agent_session_id + async def _get_foundry_project_arm_id(self) -> str: + """Get the Foundry project ARM ID from its Application Insights connection.""" + client = cast(RawFoundryAgentChatClient, self.client) + # AIProjectClient does not expose the project ARM ID directly. Derive it from the + # project-scoped connection until https://github.com/Azure/azure-sdk-for-python/issues/48825 is addressed. + connections = client.project_client.connections.list(connection_type=ConnectionType.APPLICATION_INSIGHTS) + async for connection in connections: + connection_suffix = f"/connections/{connection.name}" + if not connection.id.lower().endswith(connection_suffix.lower()): + raise ValueError( + f"The Foundry Application Insights connection ID has an unexpected format: {connection.id!r}." + ) + return _validate_project_arm_id(connection.id[: -len(connection_suffix)]) + + raise ValueError("The Foundry project does not have an Application Insights connection.") + async def configure_azure_monitor( self, enable_sensitive_data: bool = False, @@ -850,6 +891,14 @@ async def configure_azure_monitor( This method configures Azure Monitor for telemetry collection using the connection string from the Foundry project client (accessed via the internal client). + If ``project_arm_id`` was not supplied at construction, it also discovers and + caches project identity for this agent. If that optional lookup fails, a + warning is logged and export continues without project attribution; client + traces may then be absent from the Foundry portal. + + Applications that configure their own exporters can instead supply + ``project_arm_id`` at construction without calling this helper. Project + identity belongs to each agent, not to the process-wide exporter. Args: enable_sensitive_data: Enable sensitive data logging (prompts, responses). @@ -897,6 +946,18 @@ async def configure_azure_monitor( "Install it with: pip install azure-monitor-opentelemetry" ) from exc + if self._foundry_project_arm_id is None: + try: + self._foundry_project_arm_id = await self._get_foundry_project_arm_id() + except (HttpResponseError, ServiceRequestError, ServiceResponseError, ValueError) as exc: + logger.warning( + "Could not resolve the Foundry project ARM ID: %s. " + "Azure Monitor export will continue without project attribution; " + "client traces may not appear in Foundry. Supply project_arm_id on the agent " + "to configure attribution without this lookup.", + exc, + ) + if "resource" not in kwargs: kwargs["resource"] = create_resource() @@ -951,10 +1012,23 @@ class FoundryAgent( # type: ignore[misc] ) """ + @override + def _get_additional_otel_agent_attributes(self) -> Mapping[str, Any]: + """Return Foundry attributes required to discover the agent trace.""" + if self._foundry_project_arm_id: + return {_FOUNDRY_PROJECT_ARM_ID_ATTRIBUTE: self._foundry_project_arm_id} + return {} + + @override + def _should_capture_agent_response_id(self) -> bool: + """Keep the response ID on the client agent span for Foundry trace discovery.""" + return True + def __init__( self, *, project_endpoint: str | None = None, + project_arm_id: str | None = None, agent_name: str | None = None, agent_version: str | None = None, credential: AzureCredentialTypes | None = None, @@ -997,6 +1071,10 @@ def __init__( Keyword Args: project_endpoint: The Foundry project endpoint URL. + project_arm_id: Full Foundry project ARM resource ID for telemetry attribution. + Supply this when configuring exporters yourself; no Azure Monitor setup + or network lookup is performed by the constructor. When omitted, + ``configure_azure_monitor()`` attempts discovery and caches the result. agent_name: The name of the Foundry agent to connect to. agent_version: The version of the agent (for PromptAgents). credential: Azure credential for authentication. @@ -1028,6 +1106,7 @@ def __init__( """ super().__init__( project_endpoint=project_endpoint, + project_arm_id=project_arm_id, agent_name=agent_name, agent_version=agent_version, credential=credential, diff --git a/python/packages/foundry/tests/foundry/test_foundry_agent.py b/python/packages/foundry/tests/foundry/test_foundry_agent.py index a4c3b8ca25a..f28832920f4 100644 --- a/python/packages/foundry/tests/foundry/test_foundry_agent.py +++ b/python/packages/foundry/tests/foundry/test_foundry_agent.py @@ -6,6 +6,7 @@ import json import os import sys +from asyncio import CancelledError from collections.abc import Awaitable, Callable from types import SimpleNamespace from typing import Any, cast @@ -36,7 +37,7 @@ from agent_framework_openai._chat_client import RawOpenAIChatClient from agent_framework_openai._feature_usage import FeatureIndex as OpenAIFeatureIndex from azure.ai.projects import models as projects_models -from azure.core.exceptions import ResourceNotFoundError +from azure.core.exceptions import HttpResponseError, ResourceNotFoundError, ServiceRequestError, ServiceResponseError from azure.identity import AzureCliCredential from azure.identity.aio import AzureCliCredential as AsyncAzureCliCredential from openai import AsyncOpenAI @@ -947,6 +948,120 @@ def test_raw_foundry_agent_init_creates_client() -> None: assert agent.client is not None assert cast(Any, agent.client).agent_name == "test-agent" + assert agent.name == "test-agent" + + +async def test_get_foundry_project_arm_id_from_application_insights_connection() -> None: + """Test that project attribution uses the public project-scoped connection ID.""" + + project_arm_id = ( + "/subscriptions/test-sub/resourceGroups/test-rg/providers/" + "Microsoft.CognitiveServices/accounts/test-account/projects/test-project" + ) + project_client = MagicMock() + + async def connections(): + yield SimpleNamespace(id=f"{project_arm_id}/connections/appinsights", name="appinsights") + + project_client.connections.list.return_value = connections() + agent = RawFoundryAgent(project_client=project_client, agent_name="test-agent") + + assert await agent._get_foundry_project_arm_id() == project_arm_id + project_client.connections.list.assert_called_once_with( + connection_type=projects_models.ConnectionType.APPLICATION_INSIGHTS + ) + + +async def test_get_foundry_project_arm_id_rejects_unexpected_connection_id() -> None: + """Test that malformed connection metadata does not silently disable portal attribution.""" + + project_client = MagicMock() + + async def connections(): + yield SimpleNamespace(id="appinsights", name="appinsights") + + project_client.connections.list.return_value = connections() + agent = RawFoundryAgent(project_client=project_client, agent_name="test-agent") + + with pytest.raises(ValueError, match="unexpected format"): + await agent._get_foundry_project_arm_id() + + +@pytest.mark.parametrize("agent_type", [RawFoundryAgent, FoundryAgent]) +def test_foundry_agent_explicit_project_arm_id(agent_type: type[RawFoundryAgent]) -> None: + project_arm_id = ( + "/subscriptions/test-sub/resourceGroups/test-rg/providers/" + "Microsoft.CognitiveServices/accounts/test-account/projects/test-project" + ) + project_client = MagicMock() + agent = agent_type(project_client=project_client, agent_name="test-agent", project_arm_id=project_arm_id) + + assert agent._foundry_project_arm_id == project_arm_id + project_client.connections.list.assert_not_called() + project_client.telemetry.get_application_insights_connection_string.assert_not_called() + if isinstance(agent, FoundryAgent): + assert agent._get_additional_otel_agent_attributes() == {"microsoft.foundry.project.id": project_arm_id} + + +@pytest.mark.parametrize( + "project_arm_id", + [ + "", + "test-project", + "https://test-account.services.ai.azure.com/api/projects/test-project", + "/subscriptions/test-sub/resourceGroups/test-rg/providers/Microsoft.CognitiveServices/accounts/test-account", + ( + "/subscriptions/test-sub/resourceGroups/test-rg/providers/" + "Microsoft.CognitiveServices/accounts/test-account/projects/test-project/connections/appinsights" + ), + ], +) +def test_foundry_agent_rejects_invalid_explicit_project_arm_id(project_arm_id: str) -> None: + project_client = MagicMock() + with pytest.raises(ValueError, match="full Foundry project ARM resource ID"): + FoundryAgent(project_client=project_client, agent_name="test-agent", project_arm_id=project_arm_id) + project_client.get_openai_client.assert_not_called() + + +def test_foundry_agent_project_attribution_is_per_instance() -> None: + prefix = ( + "/subscriptions/test-sub/resourceGroups/test-rg/providers/" + "Microsoft.CognitiveServices/accounts/test-account/projects/" + ) + agents = [ + FoundryAgent(project_client=MagicMock(), agent_name="test-agent", project_arm_id=f"{prefix}{name}") + for name in ("first", "second") + ] + assert [agent._get_additional_otel_agent_attributes() for agent in agents] == [ + {"microsoft.foundry.project.id": f"{prefix}first"}, + {"microsoft.foundry.project.id": f"{prefix}second"}, + ] + unconfigured = FoundryAgent(project_client=MagicMock(), agent_name="test-agent") + assert unconfigured._get_additional_otel_agent_attributes() == {} + + +async def test_get_foundry_project_arm_id_rejects_account_scoped_connection() -> None: + project_client = MagicMock() + + async def connections(): + yield SimpleNamespace( + id="/subscriptions/test-sub/resourceGroups/test-rg/providers/" + "Microsoft.CognitiveServices/accounts/test-account/connections/appinsights", + name="appinsights", + ) + + project_client.connections.list.return_value = connections() + agent = RawFoundryAgent(project_client=project_client, agent_name="test-agent") + with pytest.raises(ValueError, match="full Foundry project ARM resource ID"): + await agent._get_foundry_project_arm_id() + + +async def test_get_foundry_project_arm_id_requires_connection() -> None: + project_client = MagicMock() + project_client.connections.list.return_value.__aiter__.return_value = [] + agent = RawFoundryAgent(project_client=project_client, agent_name="test-agent") + with pytest.raises(ValueError, match="does not have an Application Insights connection"): + await agent._get_foundry_project_arm_id() def test_raw_foundry_agent_init_passes_default_headers_to_client() -> None: @@ -1243,6 +1358,8 @@ def test_foundry_agent_init() -> None: assert agent.client is not None assert cast(Any, agent.client).agent_name == "test-agent" + assert agent.name == "test-agent" + assert agent._should_capture_agent_response_id() def test_foundry_agent_init_with_middleware() -> None: @@ -1264,7 +1381,8 @@ async def process(self, context: ChatContext, call_next) -> None: assert agent.client is not None -async def test_foundry_agent_configure_azure_monitor() -> None: +@pytest.mark.parametrize("explicit_project_id", [False, True]) +async def test_foundry_agent_configure_azure_monitor(explicit_project_id: bool) -> None: """Test configure_azure_monitor delegates through the underlying client.""" mock_project = MagicMock() @@ -1272,12 +1390,19 @@ async def test_foundry_agent_configure_azure_monitor() -> None: mock_project.telemetry.get_application_insights_connection_string = AsyncMock( return_value="InstrumentationKey=test-key;IngestionEndpoint=https://test.endpoint" ) - agent = FoundryAgent(project_client=mock_project, agent_name="test-agent") - mock_configure = MagicMock() mock_views = MagicMock(return_value=[]) mock_resource = MagicMock() mock_enable = MagicMock() + project_arm_id = ( + "/subscriptions/test-sub/resourceGroups/test-rg/providers/" + "Microsoft.CognitiveServices/accounts/test-account/projects/test-project" + ) + agent = FoundryAgent( + project_client=mock_project, + agent_name="test-agent", + project_arm_id=project_arm_id if explicit_project_id else None, + ) with ( patch.dict( @@ -1287,15 +1412,82 @@ async def test_foundry_agent_configure_azure_monitor() -> None: patch("agent_framework.observability.create_metric_views", mock_views), patch("agent_framework.observability.create_resource", return_value=mock_resource), patch("agent_framework.observability.enable_instrumentation", mock_enable), + patch( + "agent_framework_foundry._agent.RawFoundryAgent._get_foundry_project_arm_id", + new_callable=AsyncMock, + return_value=project_arm_id, + ) as mock_get_project_arm_id, ): await agent.configure_azure_monitor(enable_sensitive_data=True) + await agent.configure_azure_monitor(enable_sensitive_data=True) - mock_project.telemetry.get_application_insights_connection_string.assert_called_once() + assert mock_project.telemetry.get_application_insights_connection_string.await_count == 2 + if explicit_project_id: + mock_get_project_arm_id.assert_not_awaited() + else: + mock_get_project_arm_id.assert_awaited_once_with() call_kwargs = mock_configure.call_args.kwargs assert call_kwargs["connection_string"] == "InstrumentationKey=test-key;IngestionEndpoint=https://test.endpoint" assert call_kwargs["views"] == [] assert call_kwargs["resource"] is mock_resource - mock_enable.assert_called_once_with(enable_sensitive_data=True) + assert mock_enable.call_count == 2 + mock_enable.assert_called_with(enable_sensitive_data=True) + assert agent._get_additional_otel_agent_attributes() == { + "microsoft.foundry.project.id": project_arm_id, + } + + +@pytest.mark.parametrize( + "lookup_error", + [ + ValueError("Unexpected connection ID"), + HttpResponseError("Forbidden"), + ResourceNotFoundError("Connection not found"), + ServiceRequestError("Connection unavailable"), + ServiceResponseError("Incomplete response"), + ], +) +async def test_foundry_agent_configure_azure_monitor_project_lookup_failure( + lookup_error: Exception, caplog: pytest.LogCaptureFixture +) -> None: + mock_project = MagicMock() + mock_project.telemetry.get_application_insights_connection_string = AsyncMock(return_value="test-connection-string") + agent = FoundryAgent(project_client=mock_project, agent_name="test-agent") + mock_configure = MagicMock() + mock_enable = MagicMock() + + with ( + patch.dict("sys.modules", {"azure.monitor.opentelemetry": MagicMock(configure_azure_monitor=mock_configure)}), + patch("agent_framework.observability.create_metric_views", return_value=[]), + patch("agent_framework.observability.create_resource"), + patch("agent_framework.observability.enable_instrumentation", mock_enable), + patch.object(agent, "_get_foundry_project_arm_id", side_effect=lookup_error), + caplog.at_level("WARNING", logger="agent_framework.foundry"), + ): + await agent.configure_azure_monitor() + + mock_configure.assert_called_once() + assert mock_configure.call_args.kwargs["connection_string"] == "test-connection-string" + mock_enable.assert_called_once_with(enable_sensitive_data=False) + assert agent._get_additional_otel_agent_attributes() == {} + assert "export will continue without project attribution" in caplog.text + assert "project_arm_id" in caplog.text + + +@pytest.mark.parametrize("lookup_error", [RuntimeError("Programming failure"), CancelledError()]) +async def test_foundry_agent_project_lookup_does_not_swallow_unexpected_errors(lookup_error: BaseException) -> None: + mock_project = MagicMock() + mock_project.telemetry.get_application_insights_connection_string = AsyncMock(return_value="test-connection-string") + agent = FoundryAgent(project_client=mock_project, agent_name="test-agent") + mock_configure = MagicMock() + + with ( + patch.dict("sys.modules", {"azure.monitor.opentelemetry": MagicMock(configure_azure_monitor=mock_configure)}), + patch.object(agent, "_get_foundry_project_arm_id", side_effect=lookup_error), + pytest.raises(type(lookup_error)), + ): + await agent.configure_azure_monitor() + mock_configure.assert_not_called() async def test_foundry_agent_configure_azure_monitor_resource_not_found() -> None: diff --git a/python/samples/02-agents/observability/README.md b/python/samples/02-agents/observability/README.md index 822006b17dd..09b5f254360 100644 --- a/python/samples/02-agents/observability/README.md +++ b/python/samples/02-agents/observability/README.md @@ -103,7 +103,8 @@ configure_azure_monitor( enable_sensitive_telemetry() ``` -For Microsoft Foundry projects, use `client.configure_azure_monitor()` which retrieves the connection string from the project and configures everything: +For model calls through `FoundryChatClient`, use `client.configure_azure_monitor()` +to retrieve the connection string and configure Azure Monitor: ```python from agent_framework.foundry import FoundryChatClient @@ -119,6 +120,28 @@ client = FoundryChatClient( await client.configure_azure_monitor(enable_sensitive_data=True) ``` +For calls to an **existing prompt or hosted agent**, use +[`foundry_agent_tracing.py`](foundry_agent_tracing.py). It shows two supported paths: + +- `await agent.configure_azure_monitor()` configures Azure Monitor and discovers + project attribution for that `FoundryAgent` instance. +- For application-managed exporters, supply the full `project_arm_id` when + constructing the `FoundryAgent`; the constructor does not configure exporters. + +Run from `python/` using the workspace packages: + +```powershell +uv run python samples\02-agents\observability\foundry_agent_tracing.py +uv run python samples\02-agents\observability\foundry_agent_tracing.py --manual-setup --stream +``` + +Manual setup requires `FOUNDRY_PROJECT_ARM_ID` and +`APPLICATIONINSIGHTS_CONNECTION_STRING` in addition to the endpoint and agent +name. These are explicitly read by the sample. Optional discovery failures warn +and preserve Application Insights export, but may prevent Foundry discovery. +Confirm the printed trace ID appears under the agent in Foundry, not merely in +Application Insights. Keep the agent available during inspection. + Or with [Langfuse](https://langfuse.com/integrations/frameworks/microsoft-agent-framework): ```python diff --git a/python/samples/02-agents/observability/foundry_agent_tracing.py b/python/samples/02-agents/observability/foundry_agent_tracing.py new file mode 100644 index 00000000000..b9c05e999e1 --- /dev/null +++ b/python/samples/02-agents/observability/foundry_agent_tracing.py @@ -0,0 +1,112 @@ +# /// script +# requires-python = ">=3.10" +# dependencies = [ +# "agent-framework-foundry", +# "azure-monitor-opentelemetry", +# ] +# /// +# Run from python/ with the workspace environment: +# uv run python samples/02-agents/observability/foundry_agent_tracing.py + +# Copyright (c) Microsoft. All rights reserved. + +from __future__ import annotations + +import argparse +import asyncio +import os + +from agent_framework.foundry import FoundryAgent +from agent_framework.observability import create_resource, get_tracer +from azure.identity.aio import AzureCliCredential +from azure.monitor.opentelemetry import configure_azure_monitor +from dotenv import load_dotenv +from opentelemetry import trace +from opentelemetry.sdk.trace import TracerProvider + +""" +Trace calls to an existing Foundry agent without coupling identity to exporter setup. + +Default: the agent helper configures Azure Monitor and discovers the project's ARM ID. +With --manual-setup: configure the exporter yourself and pass project_arm_id to the agent. +Add --stream for streaming output. Message-content recording remains disabled. + +Environment variables: + FOUNDRY_PROJECT_ENDPOINT -- Foundry project endpoint. + FOUNDRY_AGENT_NAME -- Existing prompt or hosted agent name. + FOUNDRY_AGENT_VERSION -- Optional agent version. + FOUNDRY_PROJECT_ARM_ID -- Full project ARM ID; required for --manual-setup. + APPLICATIONINSIGHTS_CONNECTION_STRING -- Required for --manual-setup; must target + the Application Insights resource connected to the agent's Foundry project. + +The sample reads FOUNDRY_PROJECT_ARM_ID explicitly; FoundryAgent does not automatically +read that environment variable. The ARM ID includes subscription, resource group, +account and project, and is different from the data-plane endpoint. + +After running, open the existing agent in Foundry, select Traces, and search for the +printed trace ID. Look for the client invoke_agent and chat spans. The sample does not +create or delete the agent, so it remains available for inspection. +""" + +load_dotenv() + + +async def main() -> None: + parser = argparse.ArgumentParser(description="Compare helper and application-managed Foundry tracing setup.") + parser.add_argument("--manual-setup", action="store_true", help="Configure Azure Monitor outside the agent helper.") + parser.add_argument("--stream", action="store_true", help="Stream the agent response.") + args = parser.parse_args() + project_arm_id = os.getenv("FOUNDRY_PROJECT_ARM_ID") + if args.manual_setup and not project_arm_id: + parser.error("--manual-setup requires FOUNDRY_PROJECT_ARM_ID.") + + # 1. Application-managed exporters are configured once, independently of agent identity. + if args.manual_setup: + configure_azure_monitor( + connection_string=os.environ["APPLICATIONINSIGHTS_CONNECTION_STRING"], + resource=create_resource(), + ) + + async with ( + AzureCliCredential() as credential, + FoundryAgent( + project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"], + project_arm_id=project_arm_id, + agent_name=os.environ["FOUNDRY_AGENT_NAME"], + agent_version=os.getenv("FOUNDRY_AGENT_VERSION"), + credential=credential, + ) as agent, + ): + # 2. Alternatively, discover attribution and configure Azure Monitor through the agent. + # A supplied project_arm_id bypasses discovery in either setup. + if not args.manual_setup: + await agent.configure_azure_monitor() + + # 3. Group the client operation beneath an application span without project attributes. + with get_tracer().start_as_current_span("foundry-agent-tracing") as span: + print(f"Trace ID: {span.get_span_context().trace_id:032x}") + if args.stream: + result_stream = agent.run("Say hello in one sentence.", stream=True) + async for update in result_stream: + if update.text: + print(update.text, end="", flush=True) + print() + response = await result_stream.get_final_response() + else: + response = await agent.run("Say hello in one sentence.") + print(response.text) + print(f"Agent response ID: {response.response_id}") + + # 4. Flush before exit; exporting does not by itself prove Foundry portal discovery. + provider = trace.get_tracer_provider() + if isinstance(provider, TracerProvider) and not provider.force_flush(): + raise TimeoutError("Trace export did not finish before the flush timeout.") + + +if __name__ == "__main__": + asyncio.run(main()) + +# Example output: +# Trace ID: +# Hello! How can I help you today? +# Agent response ID: diff --git a/python/samples/02-agents/observability/foundry_tracing.py b/python/samples/02-agents/observability/foundry_tracing.py index 7cefa111d9a..38d0edafd71 100644 --- a/python/samples/02-agents/observability/foundry_tracing.py +++ b/python/samples/02-agents/observability/foundry_tracing.py @@ -29,6 +29,10 @@ This sample shows how to setup telemetry in Microsoft Foundry for a custom agent using ``FoundryChatClient.configure_azure_monitor()``. +For an existing Foundry prompt or hosted agent, see ``foundry_agent_tracing.py`` +instead. A separate FoundryAgent needs its own project attribution; configuring +this chat client's exporter does not initialize that agent's project identity. + First ensure you have a Foundry workspace with Application Insights enabled. And use the Operate tab to Register an Agent. Set the OpenTelemetry agent ID to the value used below in the Agent creation: ``weather-agent`` diff --git a/python/samples/02-agents/providers/foundry/README.md b/python/samples/02-agents/providers/foundry/README.md index 606b81af0ef..3de60ad1bb8 100644 --- a/python/samples/02-agents/providers/foundry/README.md +++ b/python/samples/02-agents/providers/foundry/README.md @@ -10,6 +10,7 @@ This folder contains Microsoft Foundry and Foundry Local samples for Agent Frame | [`foundry_agent_custom_client.py`](foundry_agent_custom_client.py) | Foundry Agent custom client configuration | | [`foundry_agent_hosted.py`](foundry_agent_hosted.py) | Foundry Agent for hosted agents | | [`foundry_agent_with_function_tools.py`](foundry_agent_with_function_tools.py) | Foundry Agent with local function tools | +| [`foundry_agent_tracing.py`](../../observability/foundry_agent_tracing.py) | Existing-agent tracing with helper or application-managed exporters and per-agent project attribution | ## FoundryChatClient Samples