GenAI Utils | Agent Base Type and Creation Span

Author: etserend

util/opentelemetry-util-genai/src/opentelemetry/util/genai/handler.py

@@ -80,11 +80,16 @@
 )
 from opentelemetry.util.genai.metrics import InvocationMetricsRecorder
 from opentelemetry.util.genai.span_utils import (
+    _apply_creation_finish_attributes,
     _apply_error_attributes,
     _apply_llm_finish_attributes,
     _maybe_emit_llm_event,
 )
-from opentelemetry.util.genai.types import Error, LLMInvocation
+from opentelemetry.util.genai.types import (
+    AgentCreation,
+    Error,
+    LLMInvocation,
+)
 from opentelemetry.util.genai.version import __version__
 
 
@@ -208,6 +213,74 @@ def llm(
             raise
         self.stop_llm(invocation)
 
+    # ---- Agent creation lifecycle ----
+
+    def start_create_agent(
+        self,
+        creation: AgentCreation,
+    ) -> AgentCreation:
+        """Start an agent creation and create a pending span entry."""
+        span_name = f"{creation.operation_name} {creation.agent_name}".strip()
+        span = self._tracer.start_span(
+            name=span_name,
+            kind=SpanKind.CLIENT,
+        )
+        creation.monotonic_start_s = timeit.default_timer()
+        creation.span = span
+        creation.context_token = otel_context.attach(
+            set_span_in_context(span)
+        )
+        return creation
+
+    def stop_create_agent(self, creation: AgentCreation) -> AgentCreation:  # pylint: disable=no-self-use
+        """Finalize an agent creation successfully and end its span."""
+        if creation.context_token is None or creation.span is None:
+            return creation
+
+        span = creation.span
+        _apply_creation_finish_attributes(span, creation)
+        otel_context.detach(creation.context_token)
+        span.end()
+        return creation
+
+    def fail_create_agent(  # pylint: disable=no-self-use
+        self, creation: AgentCreation, error: Error
+    ) -> AgentCreation:
+        """Fail an agent creation and end its span with error status."""
+        if creation.context_token is None or creation.span is None:
+            return creation
+
+        span = creation.span
+        _apply_creation_finish_attributes(span, creation)
+        _apply_error_attributes(span, error)
+        otel_context.detach(creation.context_token)
+        span.end()
+        return creation
+
+    @contextmanager
+    def create_agent(
+        self, creation: AgentCreation | None = None
+    ) -> Iterator[AgentCreation]:
+        """Context manager for agent creation.
+
+        Only set data attributes on the creation object, do not modify the span or context.
+
+        Starts the span on entry. On normal exit, finalizes the creation and ends the span.
+        If an exception occurs inside the context, marks the span as error, ends it, and
+        re-raises the original exception.
+        """
+        if creation is None:
+            creation = AgentCreation()
+        self.start_create_agent(creation)
+        try:
+            yield creation
+        except Exception as exc:
+            self.fail_create_agent(
+                creation, Error(message=str(exc), type=type(exc))
+            )
+            raise
+        self.stop_create_agent(creation)
+
 
 def get_telemetry_handler(
     tracer_provider: TracerProvider | None = None,

util/opentelemetry-util-genai/src/opentelemetry/util/genai/span_utils.py

@@ -32,11 +32,13 @@
 from opentelemetry.trace.propagation import set_span_in_context
 from opentelemetry.trace.status import Status, StatusCode
 from opentelemetry.util.genai.types import (
+    AgentCreation,
     Error,
     InputMessage,
     LLMInvocation,
     MessagePart,
     OutputMessage,
+    _BaseAgent,
 )
 from opentelemetry.util.genai.utils import (
     ContentCapturingMode,
@@ -47,6 +49,21 @@
 )
 
 
+def _agent_attr(name: str, fallback: str) -> str:
+    """Get a semconv attribute, falling back to a string literal if not yet in the package."""
+    return getattr(GenAI, name, fallback)
+
+
+_GEN_AI_AGENT_NAME = _agent_attr("GEN_AI_AGENT_NAME", "gen_ai.agent.name")
+_GEN_AI_AGENT_ID = _agent_attr("GEN_AI_AGENT_ID", "gen_ai.agent.id")
+_GEN_AI_AGENT_DESCRIPTION = _agent_attr(
+    "GEN_AI_AGENT_DESCRIPTION", "gen_ai.agent.description"
+)
+_GEN_AI_AGENT_VERSION = _agent_attr(
+    "GEN_AI_AGENT_VERSION", "gen_ai.agent.version"
+)
+
+
 def _get_llm_common_attributes(
     invocation: LLMInvocation,
 ) -> dict[str, Any]:
@@ -279,6 +296,75 @@ def _get_llm_response_attributes(
     return {key: value for key, value in optional_attrs if value is not None}
 
 
+def _get_base_agent_common_attributes(
+    agent: _BaseAgent,
+) -> dict[str, Any]:
+    """Get common attributes shared by all agent operations (invoke_agent, create_agent)."""
+    optional_attrs = (
+        (GenAI.GEN_AI_REQUEST_MODEL, agent.request_model),
+        (GenAI.GEN_AI_PROVIDER_NAME, agent.provider),
+        (_GEN_AI_AGENT_NAME, agent.agent_name),
+        (_GEN_AI_AGENT_ID, agent.agent_id),
+        (_GEN_AI_AGENT_DESCRIPTION, agent.agent_description),
+        (_GEN_AI_AGENT_VERSION, agent.agent_version),
+        (server_attributes.SERVER_ADDRESS, agent.server_address),
+        (server_attributes.SERVER_PORT, agent.server_port),
+    )
+
+    return {
+        GenAI.GEN_AI_OPERATION_NAME: agent.operation_name,
+        **{key: value for key, value in optional_attrs if value is not None},
+    }
+
+
+def _get_base_agent_span_name(agent: _BaseAgent) -> str:
+    """Get the span name for any agent operation."""
+    if agent.agent_name:
+        return f"{agent.operation_name} {agent.agent_name}"
+    return agent.operation_name
+
+
+def _get_creation_common_attributes(
+    creation: AgentCreation,
+) -> dict[str, Any]:
+    """Get common agent creation attributes."""
+    return _get_base_agent_common_attributes(creation)
+
+
+def _get_creation_span_name(creation: AgentCreation) -> str:
+    """Get the span name for an agent creation."""
+    return _get_base_agent_span_name(creation)
+
+
+def _apply_creation_finish_attributes(
+    span: Span, creation: AgentCreation
+) -> None:
+    """Apply attributes common to agent creation finish() paths."""
+    span.update_name(_get_creation_span_name(creation))
+
+    attributes: dict[str, Any] = {}
+    attributes.update(_get_creation_common_attributes(creation))
+
+    # System instructions (Opt-In)
+    if (
+        is_experimental_mode()
+        and get_content_capturing_mode()
+        in (
+            ContentCapturingMode.SPAN_ONLY,
+            ContentCapturingMode.SPAN_AND_EVENT,
+        )
+        and creation.system_instruction
+    ):
+        attributes[GenAI.GEN_AI_SYSTEM_INSTRUCTIONS] = gen_ai_json_dumps(
+            [asdict(p) for p in creation.system_instruction]
+        )
+
+    attributes.update(creation.attributes)
+
+    if attributes:
+        span.set_attributes(attributes)
+
+
 __all__ = [
     "_apply_llm_finish_attributes",
     "_apply_error_attributes",
@@ -287,4 +373,9 @@ def _get_llm_response_attributes(
     "_get_llm_response_attributes",
     "_get_llm_span_name",
     "_maybe_emit_llm_event",
+    "_get_base_agent_common_attributes",
+    "_get_base_agent_span_name",
+    "_apply_creation_finish_attributes",
+    "_get_creation_common_attributes",
+    "_get_creation_span_name",
 ]

util/opentelemetry-util-genai/src/opentelemetry/util/genai/types.py

@@ -236,6 +236,68 @@ class LLMInvocation(GenAIInvocation):
     monotonic_start_s: float | None = None
 
 
+@dataclass
+class _BaseAgent(GenAIInvocation):
+    """
+    Shared base class for agent lifecycle types (AgentInvocation, AgentCreation).
+
+    Contains fields common to all agent operations: identity, provider,
+    model, system instructions, server info, and telemetry plumbing.
+
+    Follows semconv for GenAI agent spans:
+    https://github.com/open-telemetry/semantic-conventions/blob/main/docs/gen-ai/gen-ai-agent-spans.md
+
+    Do not instantiate directly — use AgentInvocation or AgentCreation.
+    """
+
+    # Agent identity
+    agent_name: str | None = None
+    agent_id: str | None = None
+    agent_description: str | None = None
+    agent_version: str | None = None
+
+    # Operation
+    operation_name: str = ""
+    provider: str | None = None
+
+    # Request
+    request_model: str | None = None
+
+    # Content (Opt-In)
+    system_instruction: list[MessagePart] = field(
+        default_factory=_new_system_instruction
+    )
+
+    # Server
+    server_address: str | None = None
+    server_port: int | None = None
+
+    attributes: dict[str, Any] = field(default_factory=_new_str_any_dict)
+    """
+    Additional attributes to set on spans and/or events.
+    """
+    # Monotonic start time in seconds (from timeit.default_timer) used
+    # for duration calculations to avoid mixing clock sources. This is
+    # populated by the TelemetryHandler when starting an invocation.
+    monotonic_start_s: float | None = None
+
+
+@dataclass
+class AgentCreation(_BaseAgent):
+    """
+    Represents agent creation/initialization (create_agent operation).
+
+    Follows semconv for GenAI agent spans:
+    https://github.com/open-telemetry/semantic-conventions/blob/main/docs/gen-ai/gen-ai-agent-spans.md#create-agent-span
+
+    When creating an AgentCreation object, only update the data attributes.
+    The span and context_token attributes are set by the TelemetryHandler.
+    """
+
+    # Override default operation name
+    operation_name: str = "create_agent"
+
+
 @dataclass
 class Error:
     message: str