GenAI Utils | Add _BaseAgent base class and agent creation lifecycle

util/opentelemetry-util-genai/CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+- Add `_BaseAgent` shared base class and `AgentCreation` type for agent creation lifecycle spans
+  ([#4217](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4217))
+
 ## Version 0.3b0 (2026-02-20)
 
 - Add `gen_ai.tool_definitions` to completion hook ([#4181](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4181))

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,70 @@ def llm(
             raise
         self.stop_llm(invocation)
 
+    # ---- Agent lifecycle ----
+
+    def start_agent(
+        self,
+        agent: AgentCreation,
+    ) -> AgentCreation:
+        """Start an agent operation (create or invoke) and create a pending span entry."""
+        span_name = f"{agent.operation_name} {agent.name}".strip()
+        span = self._tracer.start_span(
+            name=span_name,
+            kind=SpanKind.CLIENT,
+        )
+        agent.monotonic_start_s = timeit.default_timer()
+        agent.span = span
+        agent.context_token = otel_context.attach(set_span_in_context(span))
+        return agent
+
+    def stop_agent(self, agent: AgentCreation) -> AgentCreation:  # pylint: disable=no-self-use
+        """Finalize an agent operation successfully and end its span."""
+        if agent.context_token is None or agent.span is None:
+            return agent
+
+        span = agent.span
+        _apply_creation_finish_attributes(span, agent)
+        otel_context.detach(agent.context_token)
+        span.end()
+        return agent
+
+    def fail_agent(  # pylint: disable=no-self-use
+        self, agent: AgentCreation, error: Error
+    ) -> AgentCreation:
+        """Fail an agent operation and end its span with error status."""
+        if agent.context_token is None or agent.span is None:
+            return agent
+
+        span = agent.span
+        _apply_creation_finish_attributes(span, agent)
+        _apply_error_attributes(span, error)
+        otel_context.detach(agent.context_token)
+        span.end()
+        return agent
+
+    @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_agent(creation)
+        try:
+            yield creation
+        except Exception as exc:
+            self.fail_agent(creation, Error(message=str(exc), type=type(exc)))
+            raise
+        self.stop_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,
@@ -279,6 +281,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.model),
+        (GenAI.GEN_AI_PROVIDER_NAME, agent.provider),
+        (GenAI.GEN_AI_AGENT_NAME, agent.name),
+        (GenAI.GEN_AI_AGENT_ID, agent.agent_id),
+        (GenAI.GEN_AI_AGENT_DESCRIPTION, agent.description),
+        ("gen_ai.agent.version", 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.name:
+        return f"{agent.operation_name} {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_instructions
+    ):
+        attributes[GenAI.GEN_AI_SYSTEM_INSTRUCTIONS] = gen_ai_json_dumps(
+            [asdict(p) for p in creation.system_instructions]
+        )
+
+    attributes.update(creation.attributes)
+
+    if attributes:
+        span.set_attributes(attributes)
+
+
 __all__ = [
     "_apply_llm_finish_attributes",
     "_apply_error_attributes",
@@ -287,4 +358,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

@@ -256,6 +256,57 @@ class LLMInvocation(GenAIInvocation):
     monotonic_start_s: float | None = None
 
 
+@dataclass
+class _BaseAgent(GenAIInvocation):
+    """Shared base class for agent lifecycle types. Do not instantiate
+    directly — use AgentCreation (or AgentInvocation in future).
+    The span and context_token attributes are set by the TelemetryHandler.
+    """
+
+    # Agent identity
+    name: str | None = None
+    agent_id: str | None = None
+    description: str | None = None
+    version: str | None = None
+
+    # Operation
+    operation_name: str = ""
+    provider: str | None = None
+
+    # Request
+    model: str | None = None  # primary model if applicable
+
+    # Content (Opt-In)
+    system_instructions: 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 an agent creation/initialization. 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 = GenAI.GenAiOperationNameValues.CREATE_AGENT.value
+
+
 @dataclass
 class Error:
     message: str