saf3ai-sdk 0.2.2

Observability and Telemetry SDK for AI Agents with Local Storage

Saf3AI SDK Documentation

Saf3AI SDK delivers unified telemetry and security scanning for Google ADK agents. Works with any agents built on ADK including the sample ADK agents. This document covers the required configuration and integration steps for production use.

---

1. Environment Preparation

Set the following environment variables before launching your agent. These can be exported in the shell, set in your process manager, or injected via your deployment platform.

# Mandatory telemetry configuration
export SAF3AI_COLLECTOR_AGENT=<your collector URL>
export SAF3AI_SERVICE_NAME=<your-service-name>

# SDK authentication
export SAF3AI_AUTH_ENABLED=true
export SAF3AI_API_KEY=<your-api-key>

# Optional: use a custom header instead of X-API-Key
export SAF3AI_API_KEY_HEADER=X-API-Key

# Security scanning target (hosted default or on-prem override)
export SAF3AI_API_ENDPOINT=<your scanner URL>

# For on-premise deployments:
# export SAF3AI_API_ENDPOINT=<your on-prem scanner URL>

# Optional diagnostics
export SAF3AI_DEBUG_MODE=false
export SAF3AI_CONSOLE_OUTPUT=false

Note: For the actual URLs, API keys, and tokens, refer to the [Configuration Reference] document provided by Saf3AI.

---

2. Installation

pip install saf3ai-sdk

For Poetry-based ADK samples:

poetry add saf3ai-sdk

---

3. Initialize Telemetry in Your Agent

Add the following near the entry point of your ADK service (for example, agent.py or __init__.py). Initialization must happen before any ADK classes are instantiated.

import os
from saf3ai_sdk import init

tracer = init(
    service_name=os.getenv("SAF3AI_SERVICE_NAME", "my-adk-agent"),
    agent_id="<Your-Agent-ID>",  # ← Your agent ID from Saf3AI platform
    framework="adk",  # ← Framework being used (supports 23+ frameworks!)
    safeai_collector_agent=os.getenv("SAF3AI_COLLECTOR_AGENT"),
    api_key=os.getenv("SAF3AI_API_KEY"),
    api_key_header_name=os.getenv("SAF3AI_API_KEY_HEADER", "X-API-Key"),
    console_output=os.getenv("SAF3AI_CONSOLE_OUTPUT", "false").lower() == "true",
    debug_mode=os.getenv("SAF3AI_DEBUG_MODE", "false").lower() == "true",
)

This call configures the Saf3AI tracer provider, ensures spans are exported to the Saf3AI Collector Agent, and automatically instruments ADK components.

---

4. Register Security Scanning

Define a policy and attach Saf3AI security callbacks to your LlmAgent. Scans include PII detection, sentiment analysis, prompt injection, jailbreak attempts, and native safety filters.

import os
from saf3ai_sdk import create_security_callback
from google.adk.agents import LlmAgent

def saf3ai_policy(text: str, scan_results: dict, text_type: str) -> bool:
    detections = scan_results.get("detection_results", {})
    for threat_type, result in detections.items():
        if result.get("result") == "MATCH_FOUND":
            return False
    return True

security_callback = create_security_callback(
    api_endpoint=os.getenv("SAF3AI_API_ENDPOINT"),
    api_key=os.getenv("SAF3AI_API_KEY"),
    timeout=15,
    on_scan_complete=saf3ai_policy,
    scan_responses=False,  # set True to inspect model outputs
)

agent = LlmAgent(
    name="my_adk_agent",
    model="gemini-2.5-flash",
    before_model_callback=security_callback,
)

All scans are linked to the active Saf3AI Collector Agent trace, making policy decisions visible in the Saf3AI Analyzer.

---

5. Verify the Integration

1. Ensure the Saf3AI Collector Agent endpoint is reachable from your environment (verify connectivity to the URL specified in SAF3AI_COLLECTOR_AGENT).
2. Launch the ADK service:

   poetry run adk web --port 8000
   

3. Interact with the agent and confirm spans appear in your telemetry backend.
4. Trigger a blocked prompt to confirm the security callback returns False and the agent response is intercepted.

---

6. Programmatic API Reference

init()

from saf3ai_sdk import init

init(
    service_name="my-agent",
    safeai_collector_agent=os.getenv("SAF3AI_COLLECTOR_AGENT"),
    api_key=os.getenv("SAF3AI_API_KEY"),
)

create_security_callback()

from saf3ai_sdk import create_security_callback

callback = create_security_callback(
    api_endpoint=os.getenv("SAF3AI_API_ENDPOINT"),
    api_key=os.getenv("SAF3AI_API_KEY"),
    on_scan_complete=my_policy,
    scan_responses=False,
)

Manual scans

from saf3ai_sdk import scan_prompt

results = scan_prompt(
    prompt="Your prompt text here",
    api_endpoint=os.getenv("SAF3AI_API_ENDPOINT"),
    model_name="gemini-2.5-flash",
)

---

7. Troubleshooting

Symptom	Resolution
No spans in telemetry backend	Ensure init() is called before ADK classes are created and verify SAF3AI_COLLECTOR_AGENT.
Security scan skipped	Confirm SAF3AI_API_ENDPOINT is reachable and before_model_callback is assigned.
Authentication error	Validate SAF3AI_API_KEY and the configured header name.
Excessive debug output	Set SAF3AI_CONSOLE_OUTPUT=false and SAF3AI_DEBUG_MODE=false.