Forem: Matteo Vitali

AI agent observability with OpenTelemetry and Grafana LGTM

Matteo Vitali — Sun, 13 Jul 2025 06:52:39 +0000

AI is powerful. But why does observing it matter?

AI agents are becoming more and more common in production. However, they often behave like black boxes: we send a prompt, we get a response… but what happens in between?

In this post, I’ll show why instrumenting AI agents matters more than ever, using open source tools like OpenTelemetry and the Grafana LGTM stack — that is, Loki, Grafana, Tempo, and Mimir (and yes, also “Looks Good To Me”! 😉).

The challenge

LLMs and AI agents are 'black boxes' systems:

Complex, non-deterministic behavior
Internal 'reasoning' is invisible at runtime
Hard to explain, trust, or debug outputs

🛡️ “You can’t govern or secure what you can’t observe.” (semicit.)

An open standards-based approach

The OWASP Agent Observability Standard (AOS)

OWASP AOS wants to bring standardized observability to AI systems:
Mission: Transform AI agents from black boxes into trustworthy systems through standardized observability.

Instrumentation – via OpenTelemetry
Traceability – end-to-end execution flow
Inspectability – insights on inner workings

OpenTelemetry

OpenTelemetry is a CNCF project that provides a standardized framework for telemetry:

Logs (timestamped events)
Metrics (numerical KPIs)
Traces (end-to-end execution paths)

It supports manual and auto instrumentation, and works with many backends — including the Grafana LGTM stack.

💻 Manual instrumentation (code based)

from openai import OpenAI
from opentelemetry.instrumentation.openai_v2 import OpenAIInstrumentor

OpenAIInstrumentor().instrument()

client = OpenAI()
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
      {
        "role": "user", 
        "content": "Write a short poem on OpenTelemetry."
      }
    ],
)

⚡ Auto instrumentation (zero code)

Alternatively, you can automatically instrument a Python app with:

FROM python:3.13-alpine

ENV OTEL_EXPORTER_OTLP_ENDPOINT=otel-collector:4317

RUN pip install --no-cache-dir \
  fastapi \
  opentelemetry-distro \
  opentelemetry-instrumentation

COPY main.py .

CMD ["opentelemetry-instrument", "uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

📦 The OpenTelemetry Collector

A key component for managing telemetry data

receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318
  hostmetrics:  # otel-collector-contrib
    root_path: "/hostfs"
    collection_interval: 10s

processors:
  batch:
  resourcedetection:
    detectors: ["env", "system"]

exporters:
  otlp/traces:
    endpoint: tempo:4317
  otlphttp/metrics:
    endpoint: http://mimir:9009/otlp
  otlphttp/logs:
    endpoint: http://loki:3100/otlp

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [batch, resourcedetection]
      exporters: [otlp/traces]
    metrics:
      receivers: [otlp, hostmetrics]
      processors: [batch, resourcedetection]
      exporters: [otlphttp/metrics]
    logs:
      receivers: [otlp]
      processors: [batch, resourcedetection]
      exporters: [otlphttp/logs]

Observability with OpenTelemetry and Grafana LGTM

Once data flows through the collector, the Grafana LGTM stack lets you explore:

Logs in Loki
Metrics in Mimir
Traces in Tempo

PydanticAI agent instrumentation

PydanticAI is a Pythonic AI agent framework with native OpenTelemetry support

from pydantic_ai import Agent

agent = Agent(
    'google-gla:gemini-2.0-flash',
    system_prompt='Be concise, reply with one sentence.',
    instrument=True,
)

result = agent.run_sync('Where does "hello world" come from?')
print(result.output)

📈 Inspect telemetry data

🤔 Final thoughts & takeaways

Even the simplest “hello world” AI agent can be observed via metrics and traces, revealing how complex and resource-hungry LLM calls are,
benefit from structured observability for debugging and trust building.

No business, No logic The AI agent is kept intentionally simple — a basic “hello world” — because the primary goal is to validate the power of observability in the context of agentic AI, not to showcase advanced intelligence.
Analytical and empirical evaluation With OpenTelemetry and Grafana, it's possible to analyze the agent both quantitatively (through metrics and traces) and qualitatively (by observing its flow), demonstrating the real value of observability in monitoring and understanding AI systems.
Conscious use of AI – Sometimes “a cannon is too much for a fly” Even a minimal AI agent benefits greatly from structured observability — it’s essential for understanding, debugging, and building trust. At the same time, it reveals a key insight: generating trivial output can come with a high computational cost, especially when relying on LLMs.

Secure by design in Python: A FastAPI app with 5 DevSecOps tools and a real time SSTI vulnerability remediation

Matteo Vitali — Sat, 10 May 2025 07:46:16 +0000

🌟 Introduction

Security should not be an afterthought in software development. Instead, it should be a core principle baked into your design, code, and CI/CD workflows.
To demonstrate this approach, I've created an intentionally insecure FastAPI app as a playground.

https://github.com/trottomv/python-insecure-app

In this post, we'll walk through:

5 open-source security tools (SCA, SAST, DAST, container scanning, API fuzzing)
How they help uncover vulnerabilities
How to remediate a real Server-Side Template Injection (SSTI) vulnerability in FastAPI

Let's dive in 👇

1️pip-audit: Python dependency vulnerability scanning

pip-audit

This tool checks for known vulnerabilities in your Python dependencies using safety-db or PyPI advisories. It's simple, fast, and effective.

Sample output:

jinja2 3.0.0 CVE-2022-XXXXX Template injection possible

Remediation:

Upgrade the affected package if a patched version is available
Pin versions in requirements.txt

2️Bandit: static code analysis (SAST) for Python

bandit .

Bandit performs static analysis on Python code to find common security issues like:

Use of dangerous functions (eval, exec, etc.)
Hardcoded passwords or secrets
Insecure file handling
External calls with requests without a timeout — which may lead to denial of service if the remote server hangs

But: Bandit currently does not detect Server-Side Template Injection (SSTI), even when using jinja2 templates unsafely.

Reference: Issue #728

This shows a limitation of SAST — it's useful, but not sufficient alone.

3️Schemathesis: API fuzzing from OpenAPI schemas

schemathesis run http://localhost:8000/openapi.json

Schemathesis generates test cases based on your OpenAPI schema and runs fuzzing-like checks on all endpoints.

Benefits:

Detects unexpected 500s and unhandled edge cases
Validates assumptions in your parameter handling

In our demo, this helped uncover inconsistent behavior in the root / endpoint, especially with template injection payloads.

4️Trivy: container image vulnerability scanner

trivy image trottomv/python-insecure-app

Trivy scans for:

OS package vulnerabilities (APT, APK, etc.)
Python dependency CVEs
Misconfigurations

Hardening tips:

Use python:3.12-slim or smaller base images
Avoid running as root (USER app)
Consider introduce distroless approach

Bonus: Trivy also outputs SBOMs for compliance.

5️OWASP ZAP: dynamic app security testing (DAST)

ZAP acts as a proxy or scanner against your running app (e.g. FastAPI on localhost:8000).

In our case, ZAP successfully discovered a Server-Side Template Injection (SSTI) vulnerability:

GET /?name={{7*6}} → "Hello 42!"

This issue was not detected by Bandit, highlighting the importance of dynamic analysis.

🛠️ Remediation: fixing the SSTI in FastAPI

Vulnerable code:

# main.py
from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from jinja2 import Template

app = FastAPI()

@app.get("/", response_class=HTMLResponse)
async def try_hack_me(name: str = "John Ripper"):
    content = f"<h1>Hello, {name}!</h1>"
    return Template(content).render()

Why it's dangerous:

Template(content).render() evaluates the string using Jinja2, which allows remote code execution if input is unsanitized

Secure fix:

# main.py
from fastapi import FastAPI
from fastapi.responses import HTMLResponse
from jinja2 import Template

app = FastAPI()

@app.get("/", response_class=HTMLResponse)
async def try_hack_me(name: str = "John Ripper"):
    content = "<h1>Hello, {{name}}!</h1>"
    return Template(content).render(name=name)

Test it:

# tests.py
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_root():
    response = client.get("/?name={{7*6}}")
    assert response.status_code == 200
    assert "42" not in response.content.decode()

✅ Conclusion

Security is not a tool — it's a mindset and a workflow.

Combining:

pip-audit for software composition analysis
bandit for static code issues
schemathesis for fuzzing testing on APIs
trivy for vulnerability scanning and SBOMs generation on artifacts
zap for dynamic runtime scanning

...provides layered defense. Fixing real issues like SSTI shows how practical this approach is.

If you're curious:

🌐 Try the demo app
✏️ Suggest improvements or contribute new findings
🚀 Integrate these tools in your next CI/CD pipeline

Stay secure ✨

Boost your Django projects's security with proper `Cache-Control` on views

Matteo Vitali — Sun, 19 Jan 2025 08:46:39 +0000

Caching is great for performance, but sensitive data must never be cached! In Django, properly configuring Cache Control on your views ensures secure handling of sensitive content, such as login pages or user-specific data.

🔒 Why is Cache Control important?
Without proper cache settings, sensitive data may be stored in the user's browser or intermediary proxies, exposing it to potential risks.

💡 How to configure Cache Control in Django?
You can use the @never_cache decorator on function-based views to prevent them from being cached (as mentioned in the official documentation):

from django.views.decorators.cache import never_cache  

@never_cache  
def my_secure_view(request):  
    # Your secure logic here  
    return HttpResponse("This page is protected from caching!")

But what if you want to reuse this logic across multiple class-based views? You can create a reusable mixin for this purpose!

# myproject/views.py

from django.contrib.auth.mixins import LoginRequiredMixin
from django.utils.decorators import method_decorator
from django.views.decorators.cache import never_cache

@method_decorator(never_cache, name="dispatch")
class PrivateAreaMixin(LoginRequiredMixin):
    """Override LoginRequiredMixin by adding Cache-Control rules."""

With this mixin, securing your class-based views becomes much simpler:

# myapp/views.py

from django.views.generic import TemplateView
from myproject.views import PrivateAreaMixin

class IndexView(PrivateAreaMixin, TemplateView):
    """The index view."""

    template_name = "index.html"

🛠 Testing the Implementation
Testing is crucial to ensure everything works as expected. Here's how you can test the PrivateAreaMixin:

# myproject/tests/test_views.py

from django.test import TestCase, RequestFactory
from django.contrib.auth.models import AnonymousUser
from django.contrib.auth import get_user_model
from django.http import HttpResponse
from django.views import View
from myproject.views import PrivateAreaMixin

class PrivateAreaMixinTest(TestCase):
    """Test private area Cache-Control override."""

    factory = RequestFactory()

    @classmethod
    def setUpTestData(cls):
        """Initialize test data."""
        cls.user = get_user_model().objects.create_user(
            username="testuser", 
            email="user@test.xyz", 
            password="5tr0ngP4ssW0rd",
        )

    def test_login_required_with_cache_control(self):
        """Test login required and Cache-Control dispatched."""

        class AView(PrivateAreaMixin, View):
            def get(self, request, *args, **kwargs):
                return HttpResponse()

        view = AView.as_view()

        # Test redirect for anonymous users
        request = self.factory.get("/")
        request.user = AnonymousUser()
        response = view(request)
        self.assertEqual(response.status_code, 302)
        self.assertEqual("/accounts/login/?next=/", response.url)

        # Test authenticated user and cache control headers
        request = self.factory.get("/")
        request.user = self.user
        response = view(request)
        self.assertEqual(response.status_code, 200)
        self.assertIn("Cache-Control", response.headers)
        self.assertEqual(
            response.headers["Cache-Control"],
            "max-age=0, no-cache, no-store, must-revalidate, private",
        )

🔥 Best Practice Tip:
Using @never_cache with a mixin like PrivateAreaMixin makes your code reusable and clean. Combined with rigorous testing, this ensures your sensitive views remain secure and aligned with best practices.

How do you handle caching and sensitive data in your Django projects?

Update:
I’m excited to share that I’ve developed a new package, django-never-cache, inspired by the concepts expressed in this post.

This lightweight package simplifies applying Cache-Control rules to sensitive views by providing reusable utilities like mixins, middlewares and decorators. It is designed to make securing Django applications easier.

Check it out on GitHub: https://github.com/trottomv/django-never-cache

Feel free to try it out, share feedback, or even contribute to its development. 🚀