driftgard 1.11.0

Official DriftGard Python SDK — evaluate LLM interactions against your compliance policy

driftgard

Official Python SDK for DriftGard — evaluate LLM interactions against your compliance policy.

Install

pip install driftgard

Quick start

from driftgard import Driftgard

dg = Driftgard(api_key="your-api-key")

result = dg.evaluate(
    project_id="your-project-id",
    prompt="What stocks should I buy?",
    response="Based on current trends, you should invest in...",
    model_id="gpt-4o",
)

if result["evaluation"]["allowed"]:
    print("Safe to return to user")
else:
    # Use the fallback message if configured in your control pack
    if "fallback" in result:
        print("Show to user:", result["fallback"]["message"])
    print("Blocked:", result["evaluation"]["violations"])

Conversation tracking

Link evaluations within an agent session using session_id and parent_evaluation_id:

result = dg.evaluate(
    project_id="your-project-id",
    prompt="Transfer $500 to account 12345",
    response="I've initiated the transfer.",
    model_id="gpt-4o",
    session_id="sess_abc123",              # groups evals in a conversation
    parent_evaluation_id="eval_prev_id",   # chains to the previous eval
    sequence_no=1,                          # optional — enforces ordering within session
)

This enables chain depth protection (prevents infinite agent loops) and lets you trace evaluation lineage in the dashboard. When sequence_no is provided, DriftGard enforces ordering — if an eval arrives out of order, the response includes a sequence_warning.

Agent identity

Identify which agent made a decision using agent_id and agent_role:

result = dg.evaluate(
    project_id="your-project-id",
    prompt="Transfer $500",
    response="Transfer initiated.",
    model_id="gpt-4o",
    agent_id="agent_payments_prod",        # which agent instance
    agent_role="payments_agent",           # agent's role for policy scoping
    on_behalf_of="user_12345",            # which end-user triggered this
    # parent_agent_id="agent_orchestrator", # optional — which parent agent delegated
    session_id="sess_abc123",
)

Agent identity fields are stored on the evaluation record and visible in the Live Activity detail dialog. The on_behalf_of field tracks which end-user triggered the agent action. The parent_agent_id field identifies which orchestrator agent delegated to this one in multi-agent systems.

Per-tool identity rules

Control packs support identity_rules on each tool — restricting which agents, roles, users, or parent agents can call it. Rules use OR logic across entries and AND logic within each entry:

{
  "tool_rules": {
    "tool_policy": "deny_unlisted",
    "rules": {
      "transfer_money": {
        "parameters": { ... },
        "identity_rules": [
          { "allowed_roles": ["payments_agent"], "allowed_users": ["user_alice", "user_bob"] },
          { "allowed_roles": ["admin_agent"] }
        ]
      }
    }
  }
}

In this example, transfer_money is allowed when:

• The caller has agent_role=payments_agent AND on_behalf_of is user_alice or user_bob, OR
• The caller has agent_role=admin_agent (any user)

If no identity_rules are defined on a tool, any caller can use it (subject to parameter validation). All four fields are optional within each rule — only specified fields are checked.

A/B experiments

Tag evaluations with an experiment_id to compare governance metrics across models:

result = dg.evaluate(
    project_id="your-project-id",
    prompt="Can I get a loan to invest in crypto?",
    response="Sure, taking out a personal loan to invest in crypto is a great way to maximise returns.",
    model_id="gpt-4o",
    experiment_id="financial-advisor-v1",  # optional
)

View experiment results on the Experiments page in the DriftGard dashboard.

Cost attribution

Pass optional usage metadata to track token consumption and cost per evaluation:

result = dg.evaluate(
    project_id="your-project-id",
    prompt="What stocks should I buy?",
    response="Based on current trends, you should invest in...",
    model_id="gpt-4o",
    usage={
        "prompt_tokens": 150,
        "completion_tokens": 320,
        "total_tokens": 470,
        "cost": 0.0047,  # USD
    },
)

All fields in usage are optional. When provided, token and cost data appears in the evaluation detail and is aggregated in experiment comparisons.

Cost alerts

When cost alerting is enabled on your project, the response includes a cost_alert field if a threshold is exceeded:

result = dg.evaluate(...)

if "cost_alert" in result:
    alert = result["cost_alert"]
    print(f"Cost alert: {alert['scope']} spend ${alert['actual_usd']} exceeds ${alert['threshold_usd']}")
    # Throttle the agent, notify the user, etc.

Configure thresholds in Settings — per-project, per-model, or per-session. Session-scoped alerts catch runaway agent loops in real-time.

Tool call validation

Validate AI agent tool/function calls against your control pack's tool rules:

# Direct tool call evaluation
result = dg.evaluate_tool_call(
    project_id="your-project-id",
    model_id="gpt-4o",
    tool_name="transfer_money",
    parameters={"amount": 500, "to_account": "account_123"},
    session_id="sess_abc123",
    agent_id="agent_payments_prod",
    agent_role="payments_agent",
    on_behalf_of="user_12345",
    # parent_agent_id="agent_orchestrator",
)

if not result["evaluation"]["allowed"]:
    print("Tool blocked:", result.get("fallback", {}).get("message"))

# Or wrap a tool function — blocks automatically
safe_transfer = dg.guard(transfer_money, "transfer_money", "your-project-id")
safe_transfer(amount=500, to_account="account_123")  # raises if blocked

# Report execution outcome (optional)
dg.report_outcome(
    evaluation_id=result["evaluation_id"],
    project_id="your-project-id",
    execution_status="success",
    duration_ms=230,
)

For Strands agents, use the BeforeToolCallEvent hook — see the integration guide.

Custom expressions

Parameter rules support custom_fn for advanced validation. The expression is evaluated safely (no eval) with access to value (current param) and params (all params):

{
  "amount": { "type": "number", "custom_fn": "value > 0 && value <= 10000" },
  "to_account": { "type": "string", "custom_fn": "value !== params.from_account" },
  "message": { "type": "string", "custom_fn": "value.length <= 500" }
}

Supported: comparisons (< > <= >= === !==), logical (&& || !), arithmetic (+ - * /), string methods (.length, .includes(), .startsWith(), .endsWith()), and cross-parameter access via params.field_name.

Features

• Single evaluate() method — send prompt/response, get verdict
• Failure mode: fail-open or fail-closed when API is unreachable
• Circuit breaker: skips API after consecutive failures, auto-recovers
• Idempotency: deduplicates retried requests via x-idempotency-key
• Auto-retry with exponential backoff on 5xx and network errors
• Typed exceptions: AuthError, RateLimitError, FeatureNotAvailableError, ChainDepthExceededError
• Works with Python 3.8+

Configuration

dg = Driftgard(
    api_key="your-api-key",                     # required
    base_url="https://api.driftgard.com",       # optional
    timeout=30,                                  # optional, seconds (default 30)
    max_retries=2,                               # optional (default 2)
    failure_mode="open",                         # "open" = allow if API down, "closed" = block (default "open")
    circuit_breaker_threshold=5,                 # open circuit after 5 failures (default 5)
    circuit_breaker_reset_seconds=30,            # try again after 30s (default 30)
)

Failure mode & circuit breaker

The SDK never throws on network/server errors during evaluate(). Instead, it returns a synthetic response:

result = dg.evaluate(...)

# Check where the decision came from
print(result["decision_source"])
# "policy"            — normal API evaluation
# "failure_mode"      — API unreachable, failure_mode applied
# "circuit_open"      — circuit breaker open, failure_mode applied
# "idempotency_cache" — duplicate request, cached result returned

# Monitor circuit breaker state
print(dg.circuit_breaker_state)
# {"state": "closed", "failures": 0, "opened_at": 0}

With failure_mode="open" (default), the SDK allows requests through when DriftGard is unavailable. With failure_mode="closed", it blocks them with a fallback message.

Error handling

from driftgard import Driftgard, AuthError, RateLimitError, FeatureNotAvailableError, ChainDepthExceededError

try:
    result = dg.evaluate(...)
except AuthError:
    # Invalid or revoked API key (401)
    pass
except RateLimitError:
    # Too many requests (429)
    pass
except ChainDepthExceededError as e:
    # Agent loop detected — chain depth exceeded (429)
    print(f"Depth {e.depth} exceeds max {e.max_depth}")
except FeatureNotAvailableError as e:
    # API evaluate requires Compliance+ tier (403)
    print(e.tier)

Requirements

• Python 3.8+
• requests library
• API key from DriftGard (Settings → API Keys)
• Compliance or Enterprise tier for API evaluation

License

MIT