Observability

The agenticapi.observability subpackage provides OpenTelemetry tracing and Prometheus metrics. All entry points gracefully no-op when the optional dependencies aren't installed.

See the Observability guide for setup patterns.

Tracing

configure_tracing

configure_tracing(
    *,
    service_name: str = "agenticapi",
    otlp_endpoint: str | None = None,
    resource_attributes: Mapping[str, str] | None = None,
    record_prompt_bodies: bool = False,
) -> None

Initialise the OpenTelemetry tracer provider for AgenticAPI.

Safe to call multiple times — subsequent calls are no-ops. Safe to call when opentelemetry-sdk is not installed: logs a warning and leaves instrumentation in no-op mode.

Parameters:

Name	Type	Description	Default
service_name	str	The service.name resource attribute. Shown in APM dashboards as the service identifier.	'agenticapi'
otlp_endpoint	str | None	When set, configures an OTLP HTTP exporter pointing at this endpoint. Typical: "http://localhost:4318" (the default OpenTelemetry Collector address).	None
resource_attributes	Mapping[str, str] | None	Extra resource attributes to attach to every span (e.g. {"deployment.environment": "prod"}).	None
record_prompt_bodies	bool	When True, the framework includes the full prompt text in spans (truncated to 500 chars). When False (default), only metadata flows so PII stays out of traces. Toggle on cautiously and only when audit requirements demand it.	False

Source code in src/agenticapi/observability/tracing.py

def configure_tracing(
    *,
    service_name: str = "agenticapi",
    otlp_endpoint: str | None = None,
    resource_attributes: Mapping[str, str] | None = None,
    record_prompt_bodies: bool = False,
) -> None:
    """Initialise the OpenTelemetry tracer provider for AgenticAPI.

    Safe to call multiple times — subsequent calls are no-ops. Safe to
    call when ``opentelemetry-sdk`` is not installed: logs a warning
    and leaves instrumentation in no-op mode.

    Args:
        service_name: The ``service.name`` resource attribute. Shown
            in APM dashboards as the service identifier.
        otlp_endpoint: When set, configures an OTLP HTTP exporter
            pointing at this endpoint. Typical: ``"http://localhost:4318"``
            (the default OpenTelemetry Collector address).
        resource_attributes: Extra resource attributes to attach to
            every span (e.g. ``{"deployment.environment": "prod"}``).
        record_prompt_bodies: When True, the framework includes the
            full prompt text in spans (truncated to 500 chars). When
            False (default), only metadata flows so PII stays out of
            traces. Toggle on cautiously and only when audit
            requirements demand it.
    """
    global _CONFIGURED
    if _CONFIGURED:
        return

    if _OTEL_TRACE is None:
        logger.warning(
            "otel_not_installed",
            message=(
                "configure_tracing() called but opentelemetry-api is not installed. "
                "Install with: pip install opentelemetry-api opentelemetry-sdk opentelemetry-exporter-otlp"
            ),
        )
        return

    try:
        from opentelemetry.sdk.resources import Resource  # type: ignore[import-not-found]
        from opentelemetry.sdk.trace import TracerProvider  # type: ignore[import-not-found]
        from opentelemetry.sdk.trace.export import (  # type: ignore[import-not-found]
            BatchSpanProcessor,
        )
    except ImportError:
        logger.warning(
            "otel_sdk_not_installed",
            message=(
                "configure_tracing() called but opentelemetry-sdk is not installed. "
                "Install with: pip install opentelemetry-sdk"
            ),
        )
        return

    attrs: dict[str, str] = {"service.name": service_name}
    if resource_attributes:
        attrs.update(resource_attributes)
    resource = Resource.create(attrs)
    provider = TracerProvider(resource=resource)

    if otlp_endpoint:
        try:
            from opentelemetry.exporter.otlp.proto.http.trace_exporter import (  # type: ignore[import-not-found]
                OTLPSpanExporter,
            )

            exporter = OTLPSpanExporter(endpoint=f"{otlp_endpoint.rstrip('/')}/v1/traces")
            provider.add_span_processor(BatchSpanProcessor(exporter))
        except ImportError:
            logger.warning(
                "otel_otlp_exporter_not_installed",
                message=(
                    "otlp_endpoint set but opentelemetry-exporter-otlp is not installed. "
                    "Install with: pip install opentelemetry-exporter-otlp"
                ),
            )

    _OTEL_TRACE.set_tracer_provider(provider)
    _CONFIGURED = True
    _record_prompt_bodies_state["enabled"] = record_prompt_bodies
    logger.info(
        "tracing_configured",
        service_name=service_name,
        otlp_endpoint=otlp_endpoint,
        record_prompt_bodies=record_prompt_bodies,
    )

get_tracer

get_tracer() -> Any

Return the active tracer.

When opentelemetry-api is installed, returns the global AgenticAPI tracer (opentelemetry.trace.get_tracer("agenticapi")). Otherwise returns a no-op tracer that satisfies the same API.

Source code in src/agenticapi/observability/tracing.py

def get_tracer() -> Any:
    """Return the active tracer.

    When ``opentelemetry-api`` is installed, returns the global
    AgenticAPI tracer (``opentelemetry.trace.get_tracer("agenticapi")``).
    Otherwise returns a no-op tracer that satisfies the same API.
    """
    if _OTEL_TRACE is None:
        return _NOOP_TRACER
    return _OTEL_TRACE.get_tracer("agenticapi")

is_otel_available

is_otel_available() -> bool

True when opentelemetry-api is importable.

Source code in src/agenticapi/observability/tracing.py

def is_otel_available() -> bool:
    """True when ``opentelemetry-api`` is importable."""
    return _OTEL_TRACE is not None

is_tracing_configured

is_tracing_configured() -> bool

True when :func:configure_tracing has been called successfully.

Source code in src/agenticapi/observability/tracing.py

def is_tracing_configured() -> bool:
    """True when :func:`configure_tracing` has been called successfully."""
    return _CONFIGURED

should_record_prompt_bodies

should_record_prompt_bodies() -> bool

Whether prompt text should be attached to spans (default False).

Source code in src/agenticapi/observability/tracing.py

def should_record_prompt_bodies() -> bool:
    """Whether prompt text should be attached to spans (default False)."""
    return _record_prompt_bodies_state["enabled"]

reset_for_tests

reset_for_tests() -> None

Reset module-global state. Test-only helper.

Source code in src/agenticapi/observability/tracing.py

def reset_for_tests() -> None:
    """Reset module-global state. Test-only helper."""
    global _CONFIGURED
    _CONFIGURED = False
    _record_prompt_bodies_state["enabled"] = False

Metrics

configure_metrics

configure_metrics(
    *,
    service_name: str = "agenticapi",
    enable_prometheus: bool = True,
) -> None

Initialise the OpenTelemetry meter provider.

Safe to call multiple times — subsequent calls are no-ops. Safe to call when opentelemetry-sdk is not installed: logs a warning and leaves the recorder in no-op mode.

Parameters:

Name	Type	Description	Default
service_name	str	service.name resource attribute.	'agenticapi'
enable_prometheus	bool	When True (default) and the optional opentelemetry-exporter-prometheus package is installed, wires up an in-process Prometheus reader so the /metrics HTTP endpoint can scrape it.	True

Source code in src/agenticapi/observability/metrics.py

def configure_metrics(
    *,
    service_name: str = "agenticapi",
    enable_prometheus: bool = True,
) -> None:
    """Initialise the OpenTelemetry meter provider.

    Safe to call multiple times — subsequent calls are no-ops. Safe
    to call when ``opentelemetry-sdk`` is not installed: logs a
    warning and leaves the recorder in no-op mode.

    Args:
        service_name: ``service.name`` resource attribute.
        enable_prometheus: When True (default) and the optional
            ``opentelemetry-exporter-prometheus`` package is installed,
            wires up an in-process Prometheus reader so the
            ``/metrics`` HTTP endpoint can scrape it.
    """
    global _METER, _PROMETHEUS_READER

    if _METER is not None:
        return

    if _OTEL_METRICS is None:
        logger.warning(
            "otel_metrics_not_installed",
            message=(
                "configure_metrics() called but opentelemetry-api is not installed. "
                "Install with: pip install opentelemetry-api opentelemetry-sdk"
            ),
        )
        return

    try:
        from opentelemetry.sdk.metrics import MeterProvider  # type: ignore[import-not-found]
        from opentelemetry.sdk.resources import Resource  # type: ignore[import-not-found]
    except ImportError:
        logger.warning(
            "otel_metrics_sdk_not_installed",
            message=(
                "configure_metrics() called but opentelemetry-sdk is not installed. "
                "Install with: pip install opentelemetry-sdk"
            ),
        )
        return

    readers: list[Any] = []
    if enable_prometheus:
        try:
            from opentelemetry.exporter.prometheus import (  # type: ignore[import-not-found]
                PrometheusMetricReader,
            )

            _PROMETHEUS_READER = PrometheusMetricReader()
            readers.append(_PROMETHEUS_READER)
        except ImportError:
            logger.info(
                "prometheus_exporter_not_installed",
                message=(
                    "Prometheus metrics requested but opentelemetry-exporter-prometheus "
                    "is not installed. Skipping. Install with: "
                    "pip install opentelemetry-exporter-prometheus"
                ),
            )

    resource = Resource.create({"service.name": service_name})
    provider = MeterProvider(resource=resource, metric_readers=readers)
    _OTEL_METRICS.set_meter_provider(provider)
    _METER = _OTEL_METRICS.get_meter("agenticapi")
    _build_instruments()
    logger.info(
        "metrics_configured",
        service_name=service_name,
        prometheus_enabled=_PROMETHEUS_READER is not None,
    )

is_metrics_available

is_metrics_available() -> bool

True when opentelemetry-api is importable.

Source code in src/agenticapi/observability/metrics.py

def is_metrics_available() -> bool:
    """True when ``opentelemetry-api`` is importable."""
    return _OTEL_METRICS is not None

record_request

record_request(
    *, endpoint: str, status: str, duration_seconds: float
) -> None

Record one completed agent request.

Source code in src/agenticapi/observability/metrics.py

def record_request(*, endpoint: str, status: str, duration_seconds: float) -> None:
    """Record one completed agent request."""
    _record_counter("requests_total", 1, {"endpoint": endpoint, "status": status})
    _record_histogram("request_duration_seconds", duration_seconds, {"endpoint": endpoint})

record_policy_denial

record_policy_denial(*, policy: str, endpoint: str) -> None

Source code in src/agenticapi/observability/metrics.py

def record_policy_denial(*, policy: str, endpoint: str) -> None:
    _record_counter("policy_denials_total", 1, {"policy": policy, "endpoint": endpoint})

record_sandbox_violation

record_sandbox_violation(
    *, kind: str, endpoint: str
) -> None

Source code in src/agenticapi/observability/metrics.py

def record_sandbox_violation(*, kind: str, endpoint: str) -> None:
    _record_counter("sandbox_violations_total", 1, {"kind": kind, "endpoint": endpoint})

record_llm_usage

record_llm_usage(
    *,
    model: str,
    input_tokens: int,
    output_tokens: int,
    cost_usd: float | None = None,
    latency_seconds: float | None = None,
) -> None

Source code in src/agenticapi/observability/metrics.py

def record_llm_usage(
    *,
    model: str,
    input_tokens: int,
    output_tokens: int,
    cost_usd: float | None = None,
    latency_seconds: float | None = None,
) -> None:
    _record_counter("llm_tokens_total", input_tokens, {"model": model, "kind": "input"})
    _record_counter("llm_tokens_total", output_tokens, {"model": model, "kind": "output"})
    if cost_usd is not None:
        _record_counter("llm_cost_usd_total", cost_usd, {"model": model})
    if latency_seconds is not None:
        _record_histogram("llm_latency_seconds", latency_seconds, {"model": model})

record_tool_call

record_tool_call(*, tool: str, endpoint: str) -> None

Source code in src/agenticapi/observability/metrics.py

def record_tool_call(*, tool: str, endpoint: str) -> None:
    _record_counter("tool_calls_total", 1, {"tool": tool, "endpoint": endpoint})

record_budget_block

record_budget_block(*, scope: str) -> None

Source code in src/agenticapi/observability/metrics.py

def record_budget_block(*, scope: str) -> None:
    _record_counter("budget_blocks_total", 1, {"scope": scope})

render_prometheus_exposition

render_prometheus_exposition() -> tuple[bytes, str]

Return the current Prometheus exposition (body, content_type).

Returns (b"", "text/plain") when metrics are not configured.

Source code in src/agenticapi/observability/metrics.py

def render_prometheus_exposition() -> tuple[bytes, str]:
    """Return the current Prometheus exposition (body, content_type).

    Returns ``(b"", "text/plain")`` when metrics are not configured.
    """
    if _PROMETHEUS_READER is None:
        return (b"", "text/plain; version=0.0.4")
    try:
        from prometheus_client import generate_latest  # type: ignore[import-not-found]
        from prometheus_client.exposition import CONTENT_TYPE_LATEST  # type: ignore[import-not-found]
    except ImportError:
        return (b"", "text/plain; version=0.0.4")

    body = generate_latest()  # The OTEL Prometheus reader registers with the default registry.
    return (body, CONTENT_TYPE_LATEST)

Distributed Trace Context Propagation

W3C traceparent and tracestate header propagation so AgenticAPI services join traces started by other systems and pass their own trace IDs onward to downstream calls. All functions gracefully no-op when OpenTelemetry propagators aren't installed.

See the Observability guide → Distributed Propagation for usage patterns.

is_propagation_available

is_propagation_available() -> bool

True when the OpenTelemetry propagation API is importable.

Source code in src/agenticapi/observability/propagation.py

def is_propagation_available() -> bool:
    """True when the OpenTelemetry propagation API is importable."""
    return _PROPAGATE is not None

extract_context_from_headers

extract_context_from_headers(
    headers: Mapping[str, str] | None,
) -> Any

Build an OpenTelemetry context from incoming HTTP headers.

Parameters:

Name	Type	Description	Default
headers	Mapping[str, str] | None	A header mapping (case-insensitive lookups). May be None for callers without an upstream request.	required

Returns:

Type	Description
Any	An OpenTelemetry Context object that the next
Any	start_as_current_span call should use as parent. When
Any	opentelemetry-api is not installed, returns None;
Any	callers can pass context=None safely to no-op tracers.

Source code in src/agenticapi/observability/propagation.py

def extract_context_from_headers(headers: Mapping[str, str] | None) -> Any:
    """Build an OpenTelemetry context from incoming HTTP headers.

    Args:
        headers: A header mapping (case-insensitive lookups). May be
            ``None`` for callers without an upstream request.

    Returns:
        An OpenTelemetry ``Context`` object that the next
        ``start_as_current_span`` call should use as parent. When
        ``opentelemetry-api`` is not installed, returns ``None``;
        callers can pass ``context=None`` safely to no-op tracers.
    """
    if _PROPAGATE is None or not headers:
        return None
    try:
        return _PROPAGATE.extract(dict(headers))
    except Exception as exc:
        logger.warning("traceparent_extract_failed", error=str(exc))
        return None

inject_context_into_headers

inject_context_into_headers(
    headers: dict[str, str],
) -> dict[str, str]

Mutate headers in place with the current trace context.

Safe to call when no span is active and when OpenTelemetry is not installed — both cases leave the headers unchanged.

Parameters:

Name	Type	Description	Default
headers	dict[str, str]	The outgoing HTTP headers dict to update.	required

Returns:

Type	Description
dict[str, str]	The same headers dict (returned for fluent chaining).

Source code in src/agenticapi/observability/propagation.py

def inject_context_into_headers(headers: dict[str, str]) -> dict[str, str]:
    """Mutate ``headers`` in place with the current trace context.

    Safe to call when no span is active and when OpenTelemetry is not
    installed — both cases leave the headers unchanged.

    Args:
        headers: The outgoing HTTP headers dict to update.

    Returns:
        The same ``headers`` dict (returned for fluent chaining).
    """
    if _PROPAGATE is None:
        return headers
    try:
        _PROPAGATE.inject(headers)
    except Exception as exc:
        logger.warning("traceparent_inject_failed", error=str(exc))
    return headers

headers_with_traceparent

headers_with_traceparent(
    base: Mapping[str, str] | None = None,
) -> dict[str, str]

Return a fresh dict with the current traceparent injected.

Convenience wrapper for callers that don't already have a headers dict to mutate. Equivalent to::

out = dict(base or {})
inject_context_into_headers(out)
return out

Source code in src/agenticapi/observability/propagation.py

def headers_with_traceparent(base: Mapping[str, str] | None = None) -> dict[str, str]:
    """Return a fresh dict with the current traceparent injected.

    Convenience wrapper for callers that don't already have a headers
    dict to mutate. Equivalent to::

        out = dict(base or {})
        inject_context_into_headers(out)
        return out
    """
    out: dict[str, str] = dict(base or {})
    inject_context_into_headers(out)
    return out

Semantic Conventions

String enums for OpenTelemetry span attributes. GenAIAttributes follows the OpenTelemetry GenAI SIG conventions; AgenticAPIAttributes adds framework-specific fields.

GenAIAttributes

Bases: StrEnum

OpenTelemetry GenAI semantic conventions (gen_ai.*).

Pinned to the stable subset of the OpenTelemetry GenAI SIG conventions so AgenticAPI traces interoperate with vendor APMs.

Source code in src/agenticapi/observability/semconv.py

class GenAIAttributes(StrEnum):
    """OpenTelemetry GenAI semantic conventions (``gen_ai.*``).

    Pinned to the stable subset of the OpenTelemetry GenAI SIG
    conventions so AgenticAPI traces interoperate with vendor APMs.
    """

    SYSTEM = "gen_ai.system"
    OPERATION_NAME = "gen_ai.operation.name"

    REQUEST_MODEL = "gen_ai.request.model"
    REQUEST_TEMPERATURE = "gen_ai.request.temperature"
    REQUEST_MAX_TOKENS = "gen_ai.request.max_tokens"
    REQUEST_TOP_P = "gen_ai.request.top_p"

    RESPONSE_MODEL = "gen_ai.response.model"
    RESPONSE_FINISH_REASONS = "gen_ai.response.finish_reasons"

    USAGE_INPUT_TOKENS = "gen_ai.usage.input_tokens"
    USAGE_OUTPUT_TOKENS = "gen_ai.usage.output_tokens"
    USAGE_CACHE_READ_INPUT_TOKENS = "gen_ai.usage.cache_read_input_tokens"
    USAGE_CACHE_WRITE_INPUT_TOKENS = "gen_ai.usage.cache_write_input_tokens"

    TOOL_NAME = "gen_ai.tool.name"
    TOOL_CALL_ID = "gen_ai.tool.call.id"

AgenticAPIAttributes

Bases: StrEnum

AgenticAPI-specific span and metric attributes.

These cover the harness-specific data that no generic APM has but every operator running AgenticAPI in production wants to see in their traces.

Source code in src/agenticapi/observability/semconv.py

class AgenticAPIAttributes(StrEnum):
    """AgenticAPI-specific span and metric attributes.

    These cover the harness-specific data that no generic APM has but
    every operator running AgenticAPI in production wants to see in
    their traces.
    """

    # Endpoint / request
    ENDPOINT_NAME = "agenticapi.endpoint.name"
    AUTONOMY_LEVEL = "agenticapi.endpoint.autonomy_level"
    REQUEST_TRACE_ID = "agenticapi.trace_id"
    SESSION_ID = "agenticapi.session_id"
    USER_ID = "agenticapi.user_id"

    # Intent
    INTENT_RAW = "agenticapi.intent.raw"
    INTENT_ACTION = "agenticapi.intent.action"
    INTENT_DOMAIN = "agenticapi.intent.domain"
    INTENT_CONFIDENCE = "agenticapi.intent.confidence"
    INTENT_PAYLOAD_SCHEMA = "agenticapi.intent.payload_schema"

    # Code generation
    CODE_LINES = "agenticapi.code.lines"

    # Policy
    POLICY_NAME = "agenticapi.policy.name"
    POLICY_ALLOWED = "agenticapi.policy.allowed"
    POLICY_VIOLATIONS = "agenticapi.policy.violations"

    # Sandbox
    SANDBOX_BACKEND = "agenticapi.sandbox.backend"
    SANDBOX_VIOLATION = "agenticapi.sandbox.violation"
    SANDBOX_DURATION_MS = "agenticapi.sandbox.duration_ms"

    # Approval
    APPROVAL_REQUIRED = "agenticapi.approval.required"
    APPROVAL_REQUEST_ID = "agenticapi.approval.request_id"

    # Cost
    COST_USD = "agenticapi.cost.usd"
    COST_BUDGET_LIMIT = "agenticapi.cost.budget_limit_usd"
    COST_BUDGET_SCOPE = "agenticapi.cost.budget_scope"

SpanNames

Bases: StrEnum

Canonical span names emitted by AgenticAPI instrumentation.

Source code in src/agenticapi/observability/semconv.py

class SpanNames(StrEnum):
    """Canonical span names emitted by AgenticAPI instrumentation."""

    AGENT_REQUEST = "agent.request"
    INTENT_PARSE = "agent.intent_parse"
    CODE_GENERATE = "agent.code_generate"
    POLICY_EVALUATE = "agent.policy_evaluate"
    STATIC_ANALYSIS = "agent.static_analysis"
    APPROVAL_WAIT = "agent.approval_wait"
    SANDBOX_EXECUTE = "agent.sandbox_execute"
    AUDIT_RECORD = "agent.audit_record"
    GEN_AI_CHAT = "gen_ai.chat"
    TOOL_CALL = "gen_ai.tool.call"