TealDrift

pip install tealtiger

import asyncio
from tealtiger.drift import TealDrift

drift = TealDrift(
    enabled=True,
    min_samples=100,
    baseline_window="7d",
    sensitivity="medium",
    thresholds={"monitor": 0.3, "escalate": 0.6, "deny": 0.8},
)

async def main():
    result = await drift.evaluate(
        request={
            "action": "database.query",
            "tool": "sql_executor",
            "content": "SELECT * FROM users WHERE role = 'admin'",
        },
        ctx={
            "agent_id": "data-analyst-bot",
            "correlation_id": "req-001",
        },
    )

    print(f"Drift score: {result.drift_score:.2f}")  # 0.72
    print(f"Flagged: {result.flagged}")               # True
    print(f"Recommendation: {result.recommendation}") # "escalate"

    for dev in result.deviations:
        print(f"  {dev.dimension}: z={dev.stats['z_score']:.1f} — {dev.explanation}")

asyncio.run(main())

class TealDrift:
    def __init__(
        self,
        enabled: bool = True,
        min_samples: int = 100,
        baseline_window: str = "7d",
        sensitivity: str = "medium",
        thresholds: dict | None = None,
        dimensions: dict | None = None,
        mcp_monitoring: dict | None = None,
        storage: str = "memory",
    ) -> None: ...

    async def evaluate(self, request: dict, ctx: dict) -> DriftResult: ...

    async def get_baseline(self, agent_id: str) -> Baseline: ...

    async def reset_baseline(self, agent_id: str) -> None: ...

    async def check_definition_drift(
        self,
        agent_id: str,
        current_tools: list[dict],
    ) -> DefinitionDriftResult: ...

drift = TealDrift(
    enabled=True,
    min_samples=100,
    baseline_window="7d",
    sensitivity="medium",  # "low" | "medium" | "high"
    thresholds={
        "monitor": 0.3,
        "escalate": 0.6,
        "deny": 0.8,
    },
    dimensions={
        "actions": True,
        "tools": True,
        "content": True,
        "temporal": True,
        "errors": True,
    },
    mcp_monitoring={
        "enabled": True,
        "alert_on_new_tools": True,
        "alert_on_param_changes": True,
        "alert_on_removed_tools": True,
    },
    storage="memory",  # "memory" | "redis" | "dynamodb"
)

result = await drift.evaluate(request, ctx)

if not result.baseline_ready:
    print(f"Baseline building: {result.sample_count}/100 samples")
    # All requests allowed during this period
    # result.recommendation == "allow"
    # result.drift_score == 0.0

drift_result = await drift.check_definition_drift(
    agent_id="agent-001",
    current_tools=[
        {"name": "web_search", "parameters": {"query": "string"}},
        {"name": "file_read", "parameters": {"path": "string"}},
        {"name": "shell_execute", "parameters": {"command": "string"}},  # new!
    ],
)

print(drift_result.drifted)      # True
print(drift_result.risk_level)   # "high"

for change in drift_result.changes:
    print(f"  {change.type}: {change.tool} (risk={change.risk})")
    # "tool_added: shell_execute (risk=high)"

# Get current baseline
baseline = await drift.get_baseline("agent-001")
print(baseline.sample_count)          # 847
print(baseline.window_start)          # "2026-05-25T00:00:00Z"
print(baseline.action_distribution)
# {"file.read": 0.45, "web_search": 0.30, "database.query": 0.20, "file.write": 0.05}

# Reset baseline (e.g., after legitimate behavior change)
await drift.reset_baseline("agent-001")

from dataclasses import dataclass

@dataclass
class DriftResult:
    drift_score: float              # 0.0 – 1.0
    flagged: bool                   # Exceeds monitor threshold
    recommendation: str             # "allow" | "monitor" | "escalate" | "deny"
    deviations: list[DriftDeviation]
    baseline_ready: bool            # Sufficient samples collected
    sample_count: int               # Samples in current baseline

@dataclass
class DriftDeviation:
    dimension: str      # "actions" | "tools" | "content" | "temporal" | "errors"
    score: float        # Deviation score for this dimension
    explanation: str    # Human-readable explanation
    stats: dict         # baseline_mean, baseline_stddev, current_value, z_score

from tealtiger import TealEngine, TealDrift

engine = TealEngine(
    policy={
        "drift": {
            "enabled": True,
            "min_samples": 100,
            "sensitivity": "medium",
            "actions": {
                "monitor": "log",
                "escalate": "approval_required",
                "deny": "auto_deny",
            },
        }
    },
    modules=[
        TealDrift(
            enabled=True,
            min_samples=100,
            baseline_window="7d",
            sensitivity="medium",
            mcp_monitoring={"enabled": True, "alert_on_new_tools": True},
        )
    ],
)