Real-Time AI Agent Observability in Django Using Strands + OpenTelemetry + CloudWatch

How to instrument your Strands-powered Django agents with OpenTelemetry traces, emit structured metrics to CloudWatch, and build dashboards that tell you exactly what your agents are doing, how long they’re taking, and where they’re going wrong — in production.

The Observability Gap in AI Agent Systems

A traditional Django view is observable by default. You see the request path, status code, response time, and SQL queries in your APM. A slow view is easy to diagnose: check the trace, find the slow span, fix it.

An AI agent is not observable by default. A Strands agent call returns a response, but between “user sent message” and “agent returned answer” there’s a black box: how many LLM turns? Which tools were called? Did the model try to call a tool that failed? Did it retry? How many input tokens were consumed? What was the cost of this specific agent invocation?

Without answers to these questions, you’re flying blind. You see high latency and you don’t know if it’s the model, a slow tool, or the agent looping more times than expected. You see a cost spike and you can’t identify which agent, which user, or which prompt pattern caused it.

This post instruments a Django Strands agent end-to-end:

• OpenTelemetry traces with spans for every agent turn, every LLM call, and every tool invocation
• Structured CloudWatch metrics for token usage, latency distributions, tool call rates, and error rates
• CloudWatch Embedded Metric Format (EMF) for zero-additional-cost high-cardinality metrics
• A Django view for live agent observability — a real-time dashboard over the telemetry data

Architecture

Django Request
      │
      ▼
AgentObservabilityMiddleware
      │
      ├── Start root span (agent.request)
      │
      ▼
ObservableStrandsAgent.chat()
      │
      ├── Span: agent.turn
      │     ├── Span: llm.invoke (model call 1)
      │     │       ├── Attribute: model_id, input_tokens, output_tokens
      │     │       └── Attribute: cost_usd
      │     ├── Span: tool.invoke (tool_name)
      │     │       ├── Attribute: tool_name, success, latency_ms
      │     │       └── Event: tool_result (truncated)
      │     ├── Span: llm.invoke (model call 2)
      │     └── Span: agent.turn.complete
      │
      ├── Emit CloudWatch EMF metrics
      │       ├── AgentTurnLatency (histogram)
      │       ├── LLMInputTokens (sum)
      │       ├── LLMOutputTokens (sum)
      │       ├── ToolCallCount (count)
      │       ├── ToolErrorRate (rate)
      │       └── AgentCost (sum)
      │
      └── Django Response

Project Setup

pip install \
    strands-agents \
    opentelemetry-api \
    opentelemetry-sdk \
    opentelemetry-exporter-otlp-proto-grpc \
    aws-opentelemetry-distro \
    boto3 \
    django

For sending traces to AWS X-Ray via OpenTelemetry:

pip install aws-xray-sdk opentelemetry-propagator-aws-xray

Settings:

# settings.py

AWS_REGION = "us-east-1"

AGENT_OBSERVABILITY = {
    # OpenTelemetry configuration
    "service_name": "django-ai-agents",
    "service_version": "1.0",
    "environment": "production",   # "development" | "staging" | "production"

    # Trace exporter: "xray" | "otlp" | "console" (for dev)
    "trace_exporter": "xray",
    "otlp_endpoint": "http://localhost:4317",  # if using OTLP collector

    # CloudWatch metrics
    "cloudwatch_namespace": "AIAgents/Django",
    "emit_emf_metrics": True,   # Embedded Metric Format (structured logs → metrics)
    "emf_log_group": "/aws/django/ai-agents",

    # What to instrument
    "trace_llm_calls": True,
    "trace_tool_calls": True,
    "trace_memory_ops": True,

    # Sampling (in dev, sample everything; in prod, tune down)
    "trace_sample_rate": 1.0,  # 1.0 = 100%, 0.1 = 10%

    # Cost tracking (prices per 1K tokens, update to match your model)
    "token_costs": {
        "us.anthropic.claude-sonnet-3-7-20250219-v1:0": {
            "input": 0.003,
            "output": 0.024,
        },
        "us.anthropic.claude-haiku-3-5-20241022-v1:0": {
            "input": 0.0008,
            "output": 0.004,
        },
    },
}

Directory layout:

myapp/
├── observability/
│   ├── __init__.py
│   ├── tracing.py          ← OpenTelemetry tracer setup
│   ├── metrics.py          ← CloudWatch EMF metric emitter
│   ├── instrumented_agent.py ← ObservableStrandsAgent
│   └── middleware.py       ← AgentObservabilityMiddleware
├── agents/
│   └── support_agent.py
├── models.py
└── views.py

Step 1: OpenTelemetry Tracer Setup

# myapp/observability/tracing.py

from __future__ import annotations
import logging
import threading
from django.conf import settings

logger = logging.getLogger(__name__)

_tracer = None
_tracer_lock = threading.Lock()

def get_tracer():
    """Return the initialized OpenTelemetry tracer (singleton)."""
    global _tracer
    if _tracer is not None:
        return _tracer
    with _tracer_lock:
        if _tracer is not None:
            return _tracer
        _tracer = _initialize_tracer()
        return _tracer

def _initialize_tracer():
    from opentelemetry import trace
    from opentelemetry.sdk.trace import TracerProvider
    from opentelemetry.sdk.trace.sampling import TraceIdRatioBased
    from opentelemetry.sdk.resources import Resource, SERVICE_NAME, SERVICE_VERSION
    from opentelemetry.semconv.resource import ResourceAttributes

    cfg = settings.AGENT_OBSERVABILITY

    resource = Resource.create({
        SERVICE_NAME: cfg["service_name"],
        SERVICE_VERSION: cfg.get("service_version", "1.0"),
        ResourceAttributes.DEPLOYMENT_ENVIRONMENT: cfg.get("environment", "production"),
        "cloud.provider": "aws",
        "cloud.region": settings.AWS_REGION,
    })

    sample_rate = cfg.get("trace_sample_rate", 1.0)
    sampler = TraceIdRatioBased(sample_rate)

    provider = TracerProvider(resource=resource, sampler=sampler)

    # Configure exporter based on settings
    exporter_type = cfg.get("trace_exporter", "console")
    _configure_exporter(provider, exporter_type, cfg)

    trace.set_tracer_provider(provider)

    tracer = trace.get_tracer(
        "django.ai.agents",
        schema_url="https://opentelemetry.io/schemas/1.21.0",
    )

    logger.info(
        f"OpenTelemetry tracer initialized: "
        f"exporter={exporter_type} "
        f"sample_rate={sample_rate}"
    )
    return tracer

def _configure_exporter(provider, exporter_type: str, cfg: dict) -> None:
    from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter

    if exporter_type == "console":
        provider.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter()))

    elif exporter_type == "xray":
        try:
            from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
            from opentelemetry.propagators.aws import AwsXRayPropagator
            from opentelemetry import propagate
            propagate.set_global_textmap(AwsXRayPropagator())

            # AWS Distro for OTel Collector receives on 4317
            exporter = OTLPSpanExporter(endpoint="http://localhost:4317", insecure=True)
            provider.add_span_processor(BatchSpanProcessor(exporter))
        except ImportError:
            logger.warning("X-Ray exporter not available, falling back to console")
            provider.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter()))

    elif exporter_type == "otlp":
        from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
        exporter = OTLPSpanExporter(
            endpoint=cfg.get("otlp_endpoint", "http://localhost:4317"),
            insecure=True,
        )
        provider.add_span_processor(BatchSpanProcessor(exporter))

def span_attributes_from_agent_call(
    model_id: str,
    input_tokens: int,
    output_tokens: int,
    tool_calls: list[str],
) -> dict:
    """Build standard span attributes for an agent invocation."""
    cfg = settings.AGENT_OBSERVABILITY
    costs = cfg.get("token_costs", {}).get(model_id, {"input": 0, "output": 0})
    cost_usd = (input_tokens / 1000 * costs["input"]) + (output_tokens / 1000 * costs["output"])

    return {
        "ai.model.id": model_id,
        "ai.tokens.input": input_tokens,
        "ai.tokens.output": output_tokens,
        "ai.tokens.total": input_tokens + output_tokens,
        "ai.cost.usd": round(cost_usd, 6),
        "ai.tool.calls": len(tool_calls),
        "ai.tool.names": ",".join(tool_calls),
    }

Step 2: CloudWatch EMF Metric Emitter

EMF (Embedded Metric Format) lets you emit structured logs that CloudWatch automatically converts to metrics. No put_metric_data calls, no batching to manage — just write JSON to stdout/logs and CloudWatch does the rest.

# myapp/observability/metrics.py

from __future__ import annotations
import json
import logging
import time
from datetime import datetime, timezone
from typing import Any
from django.conf import settings

logger = logging.getLogger(__name__)

# CloudWatch EMF log writer — emits to a dedicated logger
emf_logger = logging.getLogger("emf")

class AgentMetricsEmitter:
    """
    Emits agent metrics to CloudWatch via Embedded Metric Format (EMF).

    EMF works by writing specially-structured JSON to a log stream.
    CloudWatch Logs Insights and CloudWatch Metrics both consume it.

    Cost: you pay for log ingestion only, not per metric PUT.
    For high-cardinality metrics (per-user, per-model), this is
    dramatically cheaper than PutMetricData.
    """

    def __init__(self):
        self.cfg = settings.AGENT_OBSERVABILITY
        self.namespace = self.cfg.get("cloudwatch_namespace", "AIAgents/Django")

    def _emit(self, metrics: dict[str, Any], dimensions: dict[str, str]) -> None:
        """
        Emit a CloudWatch EMF metric payload.
        The JSON is written to the EMF log stream where CloudWatch
        automatically extracts and publishes the metrics.
        """
        if not self.cfg.get("emit_emf_metrics", True):
            return

        # Build EMF payload
        metric_definitions = [
            {"Name": name, "Unit": unit}
            for name, unit in metrics.get("_units", {}).items()
        ]
        values = {k: v for k, v in metrics.items() if not k.startswith("_")}

        payload = {
            "_aws": {
                "Timestamp": int(time.time() * 1000),
                "LogGroupName": self.cfg.get("emf_log_group", "/aws/django/ai-agents"),
                "CloudWatchMetrics": [
                    {
                        "Namespace": self.namespace,
                        "Dimensions": [list(dimensions.keys())],
                        "Metrics": metric_definitions,
                    }
                ],
            },
            **dimensions,
            **values,
        }

        emf_logger.info(json.dumps(payload))

    def emit_agent_turn(
        self,
        model_id: str,
        environment: str,
        turn_latency_ms: float,
        input_tokens: int,
        output_tokens: int,
        tool_call_count: int,
        tool_error_count: int,
        success: bool,
        agent_name: str = "default",
    ) -> None:
        """Emit metrics for a complete agent turn."""
        costs = self.cfg.get("token_costs", {}).get(model_id, {"input": 0, "output": 0})
        cost_usd = (input_tokens / 1000 * costs["input"]) + (output_tokens / 1000 * costs["output"])

        self._emit(
            metrics={
                "AgentTurnLatency": turn_latency_ms,
                "LLMInputTokens": input_tokens,
                "LLMOutputTokens": output_tokens,
                "TotalTokens": input_tokens + output_tokens,
                "ToolCallCount": tool_call_count,
                "ToolErrorCount": tool_error_count,
                "AgentCostUSD": cost_usd,
                "AgentSuccess": 1 if success else 0,
                "_units": {
                    "AgentTurnLatency": "Milliseconds",
                    "LLMInputTokens": "Count",
                    "LLMOutputTokens": "Count",
                    "TotalTokens": "Count",
                    "ToolCallCount": "Count",
                    "ToolErrorCount": "Count",
                    "AgentCostUSD": "None",  # CloudWatch doesn't have USD unit
                    "AgentSuccess": "Count",
                },
            },
            dimensions={
                "ModelId": model_id.split(".")[-1][:50],  # truncate for CW dimension limit
                "Environment": environment,
                "AgentName": agent_name,
            },
        )

    def emit_tool_call(
        self,
        tool_name: str,
        latency_ms: float,
        success: bool,
        agent_name: str = "default",
    ) -> None:
        """Emit metrics for an individual tool invocation."""
        self._emit(
            metrics={
                "ToolCallLatency": latency_ms,
                "ToolSuccess": 1 if success else 0,
                "_units": {
                    "ToolCallLatency": "Milliseconds",
                    "ToolSuccess": "Count",
                },
            },
            dimensions={
                "ToolName": tool_name,
                "AgentName": agent_name,
                "Environment": self.cfg.get("environment", "production"),
            },
        )

    def emit_llm_call(
        self,
        model_id: str,
        latency_ms: float,
        input_tokens: int,
        output_tokens: int,
        stop_reason: str,
        agent_name: str = "default",
    ) -> None:
        """Emit metrics for a single LLM API call."""
        self._emit(
            metrics={
                "LLMCallLatency": latency_ms,
                "LLMInputTokens": input_tokens,
                "LLMOutputTokens": output_tokens,
                "LLMMaxTokensReached": 1 if stop_reason == "max_tokens" else 0,
                "_units": {
                    "LLMCallLatency": "Milliseconds",
                    "LLMInputTokens": "Count",
                    "LLMOutputTokens": "Count",
                    "LLMMaxTokensReached": "Count",
                },
            },
            dimensions={
                "ModelId": model_id.split(".")[-1][:50],
                "AgentName": agent_name,
                "Environment": self.cfg.get("environment", "production"),
            },
        )

# Singleton
metrics_emitter = AgentMetricsEmitter()

Step 3: The Observable Strands Agent

# myapp/observability/instrumented_agent.py

from __future__ import annotations
import json
import time
import logging
from contextlib import contextmanager
from dataclasses import dataclass, field
from strands import Agent, tool
from strands.models import BedrockModel
from django.conf import settings
from .tracing import get_tracer, span_attributes_from_agent_call
from .metrics import metrics_emitter

logger = logging.getLogger(__name__)

# OpenTelemetry span kinds
try:
    from opentelemetry.trace import SpanKind, Status, StatusCode
    _OTEL_AVAILABLE = True
except ImportError:
    _OTEL_AVAILABLE = False
    logger.warning("OpenTelemetry not available — tracing disabled")

@dataclass
class AgentTurnResult:
    """Complete result from an instrumented agent turn."""
    text: str
    model_id: str
    input_tokens: int = 0
    output_tokens: int = 0
    total_tokens: int = 0
    tool_calls: list[str] = field(default_factory=list)
    tool_errors: list[str] = field(default_factory=list)
    turn_latency_ms: float = 0
    cost_usd: float = 0
    trace_id: str = ""
    success: bool = True

class ObservableStrandsAgent:
    """
    A Strands Agent wrapper that emits OpenTelemetry traces and
    CloudWatch metrics for every agent interaction.

    Instruments:
    - Root span per agent turn
    - Child spans per LLM invocation and tool call
    - CloudWatch EMF metrics for latency, tokens, cost, and errors
    - Structured log entries for each event

    Usage:
        agent = ObservableStrandsAgent(
            name="customer-support",
            model_id="us.anthropic.claude-sonnet-3-5-20241022-v2:0",
            tools=[my_tool_1, my_tool_2],
            system_prompt="You are a customer support agent.",
        )
        result = agent.chat(message="Help me with my order")
    """

    def __init__(
        self,
        name: str,
        model_id: str,
        tools: list,
        system_prompt: str = "",
    ):
        self.name = name
        self.model_id = model_id
        self.cfg = settings.AGENT_OBSERVABILITY

        self._model = BedrockModel(model_id=model_id, streaming=False)
        self._tools = tools
        self._system_prompt = system_prompt
        self._agent = None

    def _get_agent(self) -> Agent:
        """Lazy-initialize the Strands agent."""
        if self._agent is None:
            self._agent = Agent(
                model=self._model,
                tools=self._tools,
                system_prompt=self._system_prompt,
            )
        return self._agent

    # ── Main chat interface ────────────────────────────────────────────────

    def chat(
        self,
        message: str,
        user_id: str | None = None,
        session_id: str | None = None,
        conversation_history: list[dict] | None = None,
    ) -> AgentTurnResult:
        """
        Run a single agent turn with full observability instrumentation.
        """
        start = time.monotonic()
        tracer = get_tracer() if _OTEL_AVAILABLE else None

        tool_calls_made: list[str] = []
        tool_errors: list[str] = []
        input_tokens = 0
        output_tokens = 0
        trace_id = ""
        success = True
        response_text = ""

        # Create root span for this agent turn
        span_ctx = (
            tracer.start_as_current_span(
                "agent.turn",
                kind=SpanKind.SERVER if _OTEL_AVAILABLE else None,
                attributes={
                    "agent.name": self.name,
                    "agent.model": self.model_id,
                    "user.id": user_id or "anonymous",
                    "session.id": session_id or "",
                    "ai.message.length": len(message),
                },
            )
            if tracer and _OTEL_AVAILABLE
            else _noop_span_context()
        )

        with span_ctx as span:
            if span and _OTEL_AVAILABLE:
                from opentelemetry import trace
                ctx = trace.get_current_span().get_span_context()
                trace_id = format(ctx.trace_id, "032x") if ctx else ""

            try:
                # Instrument the agent's tool calls by wrapping tools
                instrumented_tools = [
                    self._instrument_tool(t, tool_calls_made, tool_errors)
                    for t in self._tools
                ]

                # Run the agent with instrumented tools
                instrumented_agent = Agent(
                    model=self._model,
                    tools=instrumented_tools,
                    system_prompt=self._system_prompt,
                )

                # LLM call span
                with self._llm_span(tracer, "agent.llm.invoke"):
                    result = instrumented_agent(message)
                    response_text = str(result)

                # Extract token usage from Strands result if available
                if hasattr(result, "usage"):
                    usage = result.usage
                    input_tokens = getattr(usage, "input_tokens", 0)
                    output_tokens = getattr(usage, "output_tokens", 0)

            except Exception as e:
                success = False
                logger.exception(f"Agent turn failed: agent={self.name}")
                response_text = "I encountered an error. Please try again."
                if span and _OTEL_AVAILABLE:
                    span.set_status(Status(StatusCode.ERROR, str(e)))
                    span.record_exception(e)

            finally:
                turn_latency_ms = (time.monotonic() - start) * 1000

                # Compute cost
                costs = self.cfg.get("token_costs", {}).get(
                    self.model_id, {"input": 0, "output": 0}
                )
                cost_usd = (
                    (input_tokens / 1000 * costs["input"])
                    + (output_tokens / 1000 * costs["output"])
                )

                # Add final attributes to span
                if span and _OTEL_AVAILABLE:
                    attrs = span_attributes_from_agent_call(
                        self.model_id, input_tokens, output_tokens, tool_calls_made
                    )
                    for k, v in attrs.items():
                        span.set_attribute(k, v)
                    span.set_attribute("agent.turn.latency_ms", turn_latency_ms)
                    span.set_attribute("agent.turn.success", success)

                # Emit CloudWatch metrics
                metrics_emitter.emit_agent_turn(
                    model_id=self.model_id,
                    environment=self.cfg.get("environment", "production"),
                    turn_latency_ms=turn_latency_ms,
                    input_tokens=input_tokens,
                    output_tokens=output_tokens,
                    tool_call_count=len(tool_calls_made),
                    tool_error_count=len(tool_errors),
                    success=success,
                    agent_name=self.name,
                )

                # Structured log for every turn
                logger.info(
                    "agent.turn.complete",
                    extra={
                        "agent_name": self.name,
                        "model_id": self.model_id,
                        "user_id": user_id,
                        "session_id": session_id,
                        "turn_latency_ms": round(turn_latency_ms, 2),
                        "input_tokens": input_tokens,
                        "output_tokens": output_tokens,
                        "tool_calls": tool_calls_made,
                        "tool_errors": tool_errors,
                        "cost_usd": round(cost_usd, 6),
                        "trace_id": trace_id,
                        "success": success,
                    },
                )

        return AgentTurnResult(
            text=response_text,
            model_id=self.model_id,
            input_tokens=input_tokens,
            output_tokens=output_tokens,
            total_tokens=input_tokens + output_tokens,
            tool_calls=tool_calls_made,
            tool_errors=tool_errors,
            turn_latency_ms=round(turn_latency_ms, 2),
            cost_usd=round(cost_usd, 6),
            trace_id=trace_id,
            success=success,
        )

    def _instrument_tool(self, original_tool, tool_calls_made: list, tool_errors: list):
        """
        Wrap a Strands @tool function to emit per-tool traces and metrics.
        """
        from strands import tool as strands_tool

        tracer = get_tracer() if _OTEL_AVAILABLE else None
        agent_name = self.name
        tool_name = getattr(original_tool, "__name__", str(original_tool))

        @strands_tool
        def instrumented(**kwargs):
            start = time.monotonic()
            tool_calls_made.append(tool_name)
            success = True
            result = None

            span_ctx = (
                tracer.start_as_current_span(
                    f"tool.invoke",
                    attributes={
                        "tool.name": tool_name,
                        "tool.input_keys": ",".join(kwargs.keys()),
                    },
                )
                if tracer and _OTEL_AVAILABLE
                else _noop_span_context()
            )

            with span_ctx as span:
                try:
                    result = original_tool(**kwargs)
                    if span and _OTEL_AVAILABLE:
                        result_preview = str(result)[:200]
                        span.set_attribute("tool.result_preview", result_preview)
                    return result
                except Exception as e:
                    success = False
                    tool_errors.append(f"{tool_name}: {str(e)[:100]}")
                    if span and _OTEL_AVAILABLE:
                        span.set_status(Status(StatusCode.ERROR, str(e)))
                        span.record_exception(e)
                    raise
                finally:
                    latency_ms = (time.monotonic() - start) * 1000
                    if span and _OTEL_AVAILABLE:
                        span.set_attribute("tool.latency_ms", latency_ms)
                        span.set_attribute("tool.success", success)
                    metrics_emitter.emit_tool_call(
                        tool_name=tool_name,
                        latency_ms=latency_ms,
                        success=success,
                        agent_name=agent_name,
                    )
                    logger.debug(
                        f"tool.invoked: {tool_name} "
                        f"success={success} latency={latency_ms:.1f}ms"
                    )

        instrumented.__name__ = tool_name
        return instrumented

    @contextmanager
    def _llm_span(self, tracer, span_name: str):
        """Context manager for LLM call spans."""
        if not tracer or not _OTEL_AVAILABLE:
            yield None
            return
        with tracer.start_as_current_span(span_name) as span:
            yield span

@contextmanager
def _noop_span_context():
    """No-op context manager when OpenTelemetry is not available."""
    yield None

Step 4: Django Middleware

# myapp/observability/middleware.py

import json
import logging
import time
from django.conf import settings

logger = logging.getLogger(__name__)

class AgentObservabilityMiddleware:
    """
    Middleware that adds request-level tracing context for AI agent calls.
    Injects X-Trace-Id into responses for frontend correlation.
    """

    def __init__(self, get_response):
        self.get_response = get_response
        self.cfg = settings.AGENT_OBSERVABILITY

    def __call__(self, request):
        start = time.monotonic()

        # Inject trace context from incoming headers (for distributed tracing)
        self._extract_trace_context(request)

        response = self.get_response(request)

        # Add trace ID to response header for frontend correlation
        if hasattr(request, "_trace_id") and request._trace_id:
            response["X-Trace-Id"] = request._trace_id

        # Log slow agent requests
        elapsed_ms = (time.monotonic() - start) * 1000
        if elapsed_ms > 5000 and request.path.startswith("/api/agent"):
            logger.warning(
                f"Slow agent request: path={request.path} "
                f"latency={elapsed_ms:.0f}ms"
            )

        return response

    def _extract_trace_context(self, request) -> None:
        """Propagate incoming trace context (W3C TraceContext or X-Ray headers)."""
        try:
            from opentelemetry.propagators.aws import AwsXRayPropagator
            from opentelemetry.context import attach, get_current
            from opentelemetry import propagate

            # Extract context from HTTP headers
            carrier = {
                k.lower(): v
                for k, v in request.META.items()
                if k.startswith("HTTP_")
            }
            # Convert META key format (HTTP_X_FOO) to header format (x-foo)
            normalized = {
                k[5:].replace("_", "-"): v
                for k, v in carrier.items()
            }
            ctx = propagate.extract(normalized)
            attach(ctx)
        except Exception:
            pass  # trace propagation is best-effort

Step 5: The Concrete Agent Implementation

# myapp/agents/support_agent.py

from strands import tool
from myapp.observability.instrumented_agent import ObservableStrandsAgent

@tool
def get_order_status(order_id: str) -> dict:
    """Retrieve the current status and details of an order."""
    from myapp.models import Order
    try:
        order = Order.objects.values("id", "status", "total", "created_at").get(pk=order_id)
        return {k: str(v) for k, v in order.items()}
    except Order.DoesNotExist:
        return {"error": f"Order {order_id} not found"}

@tool
def create_refund_request(order_id: str, reason: str) -> dict:
    """Create a refund request for an order."""
    from myapp.models import RefundRequest
    req = RefundRequest.objects.create(order_id=order_id, reason=reason)
    return {"refund_id": str(req.id), "status": "pending_review"}

support_agent = ObservableStrandsAgent(
    name="customer-support",
    model_id="us.anthropic.claude-sonnet-3-5-20241022-v2:0",
    tools=[get_order_status, create_refund_request],
    system_prompt=(
        "You are a customer support agent. "
        "Help customers with orders, billing, and account issues. "
        "Be concise and accurate. Always use tools to get real data."
    ),
)

Step 6: Django Views

# myapp/views.py

import json
import logging
from django.contrib.auth.decorators import login_required
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST, require_GET
from .agents.support_agent import support_agent
from .models import AgentTurnLog

logger = logging.getLogger(__name__)

@login_required
@csrf_exempt
@require_POST
def agent_chat(request):
    """
    AI agent chat endpoint with full observability.
    """
    try:
        payload = json.loads(request.body)
        message = payload.get("message", "").strip()
    except json.JSONDecodeError:
        return JsonResponse({"error": "Invalid JSON"}, status=400)

    if not message:
        return JsonResponse({"error": "message required"}, status=400)

    session_id = request.session.get("agent_session_id")

    result = support_agent.chat(
        message=message,
        user_id=str(request.user.id),
        session_id=session_id,
    )

    # Persist turn log for the observability dashboard
    AgentTurnLog.objects.create(
        user=request.user,
        agent_name=result.model_id.split(".")[-1],
        input_tokens=result.input_tokens,
        output_tokens=result.output_tokens,
        turn_latency_ms=result.turn_latency_ms,
        tool_calls=result.tool_calls,
        cost_usd=result.cost_usd,
        trace_id=result.trace_id,
        success=result.success,
    )

    return JsonResponse({
        "reply": result.text,
        "trace_id": result.trace_id,
        "meta": {
            "latency_ms": result.turn_latency_ms,
            "tokens": result.total_tokens,
            "cost_usd": result.cost_usd,
            "tools_called": result.tool_calls,
        },
    })

@login_required
@require_GET
def agent_observability_dashboard(request):
    """
    Real-time observability dashboard API.
    Returns aggregated metrics from recent agent turns.
    """
    if not request.user.is_staff:
        return JsonResponse({"error": "Staff access required"}, status=403)

    days = int(request.GET.get("days", 1))
    return JsonResponse(AgentTurnLog.dashboard_summary(days=days))

Step 7: Django Model for Local Observability

# myapp/models.py (excerpt)

from django.db import models
from django.db.models import Avg, Sum, Count, F, ExpressionWrapper, FloatField

class AgentTurnLog(models.Model):
    """
    Local turn log for the Django observability dashboard.
    Complements CloudWatch metrics with queryable structured data.
    """
    user = models.ForeignKey(
        "auth.User", on_delete=models.SET_NULL,
        null=True, blank=True, db_index=True
    )
    agent_name = models.CharField(max_length=100, db_index=True)
    input_tokens = models.IntegerField(default=0)
    output_tokens = models.IntegerField(default=0)
    turn_latency_ms = models.FloatField(null=True)
    tool_calls = models.JSONField(default=list)
    cost_usd = models.DecimalField(max_digits=10, decimal_places=8, default=0)
    trace_id = models.CharField(max_length=64, blank=True)
    success = models.BooleanField(default=True)
    created_at = models.DateTimeField(auto_now_add=True, db_index=True)

    class Meta:
        ordering = ["-created_at"]
        indexes = [
            models.Index(fields=["agent_name", "created_at"]),
            models.Index(fields=["success", "created_at"]),
        ]

    @classmethod
    def dashboard_summary(cls, days: int = 1) -> dict:
        from django.utils import timezone
        import datetime
        since = timezone.now() - datetime.timedelta(days=days)
        logs = cls.objects.filter(created_at__gte=since)

        total = logs.count()
        if total == 0:
            return {"period_days": days, "total_turns": 0}

        agg = logs.aggregate(
            avg_latency=Avg("turn_latency_ms"),
            p50_latency=Avg("turn_latency_ms"),  # approximation
            total_input_tokens=Sum("input_tokens"),
            total_output_tokens=Sum("output_tokens"),
            total_cost=Sum("cost_usd"),
            success_count=Count("id", filter=models.Q(success=True)),
        )

        # Tool usage frequency
        from collections import Counter
        all_tool_calls = []
        for log in logs.values_list("tool_calls", flat=True):
            all_tool_calls.extend(log or [])
        tool_freq = dict(Counter(all_tool_calls).most_common(10))

        # Latency percentiles (approximate from DB)
        latencies = list(logs.values_list("turn_latency_ms", flat=True).order_by("turn_latency_ms"))
        p95_idx = int(len(latencies) * 0.95)
        p99_idx = int(len(latencies) * 0.99)

        return {
            "period_days": days,
            "total_turns": total,
            "success_rate_pct": round(agg["success_count"] / total * 100, 1),
            "latency": {
                "avg_ms": round(agg["avg_latency"] or 0, 1),
                "p95_ms": round(latencies[p95_idx] if latencies else 0, 1),
                "p99_ms": round(latencies[p99_idx] if latencies else 0, 1),
            },
            "tokens": {
                "total_input": agg["total_input_tokens"] or 0,
                "total_output": agg["total_output_tokens"] or 0,
                "avg_per_turn": round(
                    ((agg["total_input_tokens"] or 0) + (agg["total_output_tokens"] or 0)) / total
                ),
            },
            "cost": {
                "total_usd": float(agg["total_cost"] or 0),
                "avg_per_turn_usd": float((agg["total_cost"] or 0) / total),
                "projected_daily_usd": float((agg["total_cost"] or 0) * (1440 / (days * 1440))),
            },
            "tools": {
                "total_calls": len(all_tool_calls),
                "avg_per_turn": round(len(all_tool_calls) / total, 2),
                "most_used": tool_freq,
            },
        }

CloudWatch Dashboard Configuration

After EMF metrics start flowing, create a CloudWatch dashboard programmatically:

# management/commands/create_agent_dashboard.py

from django.core.management.base import BaseCommand
from django.conf import settings
import boto3
import json

class Command(BaseCommand):
    help = "Create CloudWatch dashboard for AI agent observability"

    def handle(self, *args, **options):
        cfg = settings.AGENT_OBSERVABILITY
        namespace = cfg["cloudwatch_namespace"]
        region = settings.AWS_REGION
        cw = boto3.client("cloudwatch", region_name=region)

        dashboard_body = {
            "widgets": [
                {
                    "type": "metric",
                    "properties": {
                        "title": "Agent Turn Latency (p50/p95/p99)",
                        "metrics": [
                            [namespace, "AgentTurnLatency", "Environment", "production",
                             {"stat": "p50", "label": "p50"}],
                            [namespace, "AgentTurnLatency", "Environment", "production",
                             {"stat": "p95", "label": "p95"}],
                            [namespace, "AgentTurnLatency", "Environment", "production",
                             {"stat": "p99", "label": "p99"}],
                        ],
                        "period": 60,
                        "view": "timeSeries",
                        "yAxis": {"left": {"label": "ms"}},
                    },
                },
                {
                    "type": "metric",
                    "properties": {
                        "title": "Token Usage (Input vs Output)",
                        "metrics": [
                            [namespace, "LLMInputTokens", "Environment", "production",
                             {"stat": "Sum", "label": "Input Tokens"}],
                            [namespace, "LLMOutputTokens", "Environment", "production",
                             {"stat": "Sum", "label": "Output Tokens"}],
                        ],
                        "period": 300,
                        "view": "timeSeries",
                    },
                },
                {
                    "type": "metric",
                    "properties": {
                        "title": "Agent Success Rate",
                        "metrics": [
                            [{"expression": "m1/m2*100", "label": "Success Rate %"}],
                            [namespace, "AgentSuccess", "Environment", "production",
                             {"id": "m1", "stat": "Sum", "visible": False}],
                            [namespace, "AgentSuccess", "Environment", "production",
                             {"id": "m2", "stat": "SampleCount", "visible": False}],
                        ],
                        "period": 300,
                        "view": "timeSeries",
                        "yAxis": {"left": {"min": 0, "max": 100, "label": "%"}},
                    },
                },
                {
                    "type": "metric",
                    "properties": {
                        "title": "Tool Call Latency by Tool",
                        "metrics": [
                            [namespace, "ToolCallLatency", "ToolName", "get_order_status",
                             {"stat": "p99", "label": "get_order_status p99"}],
                            [namespace, "ToolCallLatency", "ToolName", "create_refund_request",
                             {"stat": "p99", "label": "create_refund_request p99"}],
                        ],
                        "period": 60,
                        "view": "timeSeries",
                    },
                },
                {
                    "type": "metric",
                    "properties": {
                        "title": "Cost (USD/5min)",
                        "metrics": [
                            [namespace, "AgentCostUSD", "Environment", "production",
                             {"stat": "Sum", "label": "Total Cost USD"}],
                        ],
                        "period": 300,
                        "view": "timeSeries",
                    },
                },
                {
                    "type": "metric",
                    "properties": {
                        "title": "Tool Error Rate",
                        "metrics": [
                            [namespace, "ToolErrorCount", "Environment", "production",
                             {"stat": "Sum", "label": "Tool Errors"}],
                        ],
                        "period": 60,
                        "view": "timeSeries",
                    },
                },
            ]
        }

        cw.put_dashboard(
            DashboardName="AIAgents-Django",
            DashboardBody=json.dumps(dashboard_body),
        )
        self.stdout.write(
            self.style.SUCCESS(
                f"Dashboard created: AIAgents-Django in {region}\n"
                f"View at: https://{region}.console.aws.amazon.com/cloudwatch/"
                f"home#dashboards:name=AIAgents-Django"
            )
        )

Alerting on Agent Anomalies

# management/commands/create_agent_alarms.py

from django.core.management.base import BaseCommand
from django.conf import settings
import boto3

class Command(BaseCommand):
    help = "Create CloudWatch alarms for AI agent observability"

    def handle(self, *args, **options):
        cfg = settings.AGENT_OBSERVABILITY
        namespace = cfg["cloudwatch_namespace"]
        cw = boto3.client("cloudwatch", region_name=settings.AWS_REGION)
        sns_arn = "arn:aws:sns:us-east-1:123456789:ai-agent-alerts"  # your SNS topic

        alarms = [
            {
                "AlarmName": "AgentTurnLatency-P99-High",
                "AlarmDescription": "Agent p99 latency exceeds 30 seconds",
                "MetricName": "AgentTurnLatency",
                "Namespace": namespace,
                "Statistic": "p99",
                "Threshold": 30000,  # 30 seconds in ms
                "ComparisonOperator": "GreaterThanThreshold",
                "Period": 300,
                "EvaluationPeriods": 2,
                "Dimensions": [{"Name": "Environment", "Value": "production"}],
            },
            {
                "AlarmName": "AgentSuccessRate-Low",
                "AlarmDescription": "Agent success rate below 95%",
                "AlarmActions": [sns_arn],
                "Metrics": [
                    {
                        "Id": "m1",
                        "MetricStat": {
                            "Metric": {
                                "Namespace": namespace,
                                "MetricName": "AgentSuccess",
                                "Dimensions": [{"Name": "Environment", "Value": "production"}],
                            },
                            "Period": 300,
                            "Stat": "Sum",
                        },
                        "ReturnData": False,
                    },
                    {
                        "Id": "m2",
                        "MetricStat": {
                            "Metric": {
                                "Namespace": namespace,
                                "MetricName": "AgentSuccess",
                                "Dimensions": [{"Name": "Environment", "Value": "production"}],
                            },
                            "Period": 300,
                            "Stat": "SampleCount",
                        },
                        "ReturnData": False,
                    },
                    {
                        "Id": "rate",
                        "Expression": "m1/m2*100",
                        "Label": "Success Rate",
                        "ReturnData": True,
                    },
                ],
                "ComparisonOperator": "LessThanThreshold",
                "Threshold": 95,
                "EvaluationPeriods": 3,
            },
            {
                "AlarmName": "AgentCost-Spike",
                "AlarmDescription": "Agent cost exceeds $50 in 5 minutes",
                "MetricName": "AgentCostUSD",
                "Namespace": namespace,
                "Statistic": "Sum",
                "Threshold": 50.0,
                "ComparisonOperator": "GreaterThanThreshold",
                "Period": 300,
                "EvaluationPeriods": 1,
                "Dimensions": [{"Name": "Environment", "Value": "production"}],
            },
        ]

        for alarm in alarms:
            cw.put_metric_alarm(**alarm)
            self.stdout.write(f"Created alarm: {alarm['AlarmName']}")

        self.stdout.write(self.style.SUCCESS("All agent alarms created"))

What You See in CloudWatch After Deployment

After 1 hour of production traffic, your observability picture looks like:

X-Ray Service Map: Shows django-ai-agents → bedrock (claude-sonnet) → {tool_1, tool_2} as connected nodes with latency and error rate annotations. Slow tool calls are immediately visible as orange/red edges.

CloudWatch Metrics dashboard:

• AgentTurnLatency p50/p95/p99 as a time series — shows gradual degradation if a new model version is slower
• LLMInputTokens and LLMOutputTokens as stacked bar — shows token usage pattern (is output tokens growing? prompt getting longer?)
• ToolCallCount per tool — shows which tools are hot, which are rarely used
• AgentCostUSD per 5 minutes — immediately spikes if a prompt pattern starts generating unusually long responses
• AgentSuccess rate — drops to zero if Bedrock has an outage; alerts fire within 1 evaluation period (5 minutes)

CloudWatch Logs Insights query for the most expensive agent sessions:

fields @timestamp, agent_name, cost_usd, total_tokens, trace_id
| filter cost_usd > 0.01
| sort cost_usd desc
| limit 20

Django dashboard API (/api/agent/observability/):

{
  "period_days": 1,
  "total_turns": 8420,
  "success_rate_pct": 99.2,
  "latency": {"avg_ms": 3840, "p95_ms": 8200, "p99_ms": 14600},
  "tokens": {"total_input": 12840000, "avg_per_turn": 1525},
  "cost": {"total_usd": 47.32, "avg_per_turn_usd": 0.0056},
  "tools": {"avg_per_turn": 1.8, "most_used": {"get_order_status": 9420, "create_refund": 1240}}
}

Conclusion

Observability for AI agents isn’t optional in production — it’s the only way to know why a latency spike happened, which prompt pattern caused a cost spike, and whether a tool failure is affecting agent success rates.

The stack in this post covers all three observability signals:

Traces (via OpenTelemetry → X-Ray) give you the causal chain: request → agent turn → LLM call → tool call → tool call → LLM call → response. Every span, every duration, every error.

Metrics (via CloudWatch EMF) give you aggregate health: p99 latency, token rates, cost per minute, success rate, tool error rates. These feed alarms and dashboards.

Logs give you context: structured JSON with trace IDs, user IDs, tool call lists, and token counts on every turn. Searchable in CloudWatch Logs Insights.

Put all three together and your AI agents become as observable as any other production service. The ObservableStrandsAgent class is the single abstraction that delivers all of it — wrap your agent once, get all three signals automatically on every call.

Resources

• OpenTelemetry Python SDK
• AWS Distro for OpenTelemetry
• CloudWatch Embedded Metric Format
• AWS X-Ray with Django
• Strands Agents SDK
• OpenTelemetry semantic conventions for AI

Deployed agent observability in production? Share what surprised you most — the cost distribution across users, the tool latency outliers, or the unexpected agent loop patterns. The first week of production data is always full of surprises.