driftgard 1.27.1

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"]:
    # If sanitized, use the safe version (PII/secrets redacted)
    if result["evaluation"].get("sanitized"):
        print("Use sanitized:", result["evaluation"]["sanitized_response"])
    else:
        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"])

Inline Enforcement

Use evaluate() when your application already calls the model and sends prompt/response to DriftGard for evaluation. Use gateway_chat_completions() when DriftGard should sit inline, call the provider, enforce policy, and return the final model response.

from driftgard import Driftgard

dg = Driftgard(
    api_key="your-api-key",
    base_url="https://api.driftgard.com",
)

completion = dg.gateway_chat_completions(
    project_id="your-project-id",
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "You are a support assistant."},
        {"role": "user", "content": "Summarize the refund process."},
    ],
    driftgard={
        "gateway_mode": "enforce",
        "agent_id": "support-agent",
        "agent_role": "customer_support",
        "session_id": "sess_123",
    },
)

print(completion["choices"][0]["message"]["content"])
print(completion.get("driftgard", {}).get("gateway_decision"))
print(completion.get("driftgard", {}).get("execution_timeline"))

Gateway V2 stage-aware metadata

Gateway responses may include stage-aware policy action metadata. This is additive; existing SDK calls do not change.

dg_meta = completion.get("driftgard", {})

for action in dg_meta.get("evaluation", {}).get("policy_actions", []):
    print(action.get("type"))    # ALLOW, BLOCK, AUDIT, REDACT_RECOMMENDED, ...
    print(action.get("stage"))   # BEFORE_PROVIDER or AFTER_PROVIDER
    print(action.get("status"))  # selected, reported, or executed

for violation in dg_meta.get("evaluation", {}).get("violations", []):
    print(violation.get("gateway_stage"))           # input or output
    print(violation.get("effective_match_target"))  # prompt or response

Control Pack match_target maps naturally in Gateway mode:

• prompt runs before the provider.
• response runs after the provider.
• both runs at both stages, scoped to the current side.

Managed provider keys configured in DriftGard are used by default. For transient-key projects, pass provider_api_key:

completion = dg.gateway_chat_completions(
    project_id="your-project-id",
    model="gpt-4o-mini",
    messages=messages,
    provider_api_key=os.environ["OPENAI_API_KEY"],
)

Streaming returns the raw requests.Response, so you can consume Server-Sent Events with iter_lines():

stream = dg.stream_gateway_chat_completions(
    project_id="your-project-id",
    model="gpt-4o-mini",
    messages=messages,
    driftgard={"gateway_mode": "observe", "session_id": "sess_123"},
)

for line in stream.iter_lines(decode_unicode=True):
    if line:
        print(line)

Governed Knowledge

Use Knowledge helpers to create sources, index documents, search governed chunks, and inject retrieved context through the gateway.

access_control is optional. Sources are open unless you set "mode": "restricted". For restricted sources, pass a matching principal on direct search or knowledge_principal on Gateway calls; DriftGard records the supplied user, groups, roles, and agent role in the retrieval run.

source = dg.create_knowledge_source(
    project_id="your-project-id",
    name="Refund SOPs",
    source_type="upload",
)

dg.update_knowledge_source_governance(
    project_id="your-project-id",
    source_id=source["source"]["source_id"],
    governance={
        "status": "official",
        "trust_score": 92,
        "owner": "Support Governance",
        "last_reviewed_at": "2026-07-01T00:00:00.000Z",
        "access_control": {
            "mode": "restricted",
            "allowed_groups": ["support-leads"],
            "allowed_agent_roles": ["customer_support"],
        },
    },
)

dg.upload_knowledge_document(
    project_id="your-project-id",
    source_id=source["source"]["source_id"],
    file=b"Refunds can be approved when the service was unavailable.",
    file_name="refund-sop.txt",
    content_type="text/plain",
)

search = dg.search_knowledge(
    project_id="your-project-id",
    query="When can a refund be approved?",
    source_ids=[source["source"]["source_id"]],
    top_k=3,
    principal={
        "user_id": "user_123",
        "groups": ["support-leads"],
        "agent_role": "customer_support",
    },
    audit=True,
)

print(search["retrieval_run_id"])
print(search["results"])
print(search.get("retrieval_decision", {}).get("quality"))
print(search.get("retrieval_decision", {}).get("access_control", {}).get("principal"))

replay = dg.replay_knowledge_retrieval_run(search["retrieval_run_id"], project_id="your-project-id")
print(replay.get("comparison"))

For S3 sources, configure and sync the connector:

s3 = dg.create_knowledge_source(
    project_id="your-project-id",
    name="Refund S3",
    source_type="s3",
)

dg.configure_s3_knowledge_source(
    project_id="your-project-id",
    source_id=s3["source"]["source_id"],
    connector_config={
        "bucket": "customer-knowledge-bucket",
        "prefix": "refunds/",
        "region": "ap-southeast-2",
        "auth_mode": "assume_role",
        "role_arn": "arn:aws:iam::<account-id>:role/driftgard-knowledge-read",
        "external_id": "external-id-from-driftgard",
        "sync_settings": {"max_objects": 1000, "include_patterns": "*.txt\n*.md\n*.pdf"},
    },
)

dg.test_knowledge_connector(s3["source"]["source_id"], project_id="your-project-id")
sync = dg.sync_knowledge_source(s3["source"]["source_id"], project_id="your-project-id")
job = dg.get_knowledge_sync_job(s3["source"]["source_id"], sync["job"]["job_id"], project_id="your-project-id")
print(job["job"]["status"])

For Confluence sources, configure the site, spaces, and optional label filters:

confluence = dg.create_knowledge_source(
    project_id="your-project-id",
    name="Support Confluence",
    source_type="confluence",
)

dg.configure_knowledge_source_connector(
    project_id="your-project-id",
    source_id=confluence["source"]["source_id"],
    connector_config={
        "base_url": "https://your-site.atlassian.net/wiki",
        "email": "[e-mail]",
        "api_token": "confluence-api-token",
        "space_keys": ["SUPPORT", "POLICY"],
        "include_labels": ["approved"],
        "exclude_labels": ["draft", "archive", "archived"],
        "max_pages": 100,
        "auto_sync": {"enabled": False, "interval_minutes": 60},
    },
)

dg.test_knowledge_connector(confluence["source"]["source_id"], project_id="your-project-id")
confluence_sync = dg.sync_knowledge_source(confluence["source"]["source_id"], project_id="your-project-id")
print(confluence_sync["job"]["job_id"])

Gateway knowledge injection uses the same gateway helper:

completion = dg.gateway_chat_completions(
    project_id="your-project-id",
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "When can a refund be approved?"}],
    driftgard={
        "agent_id": "support-agent",
        "agent_role": "customer_support",
        "knowledge_principal": {
            "user_id": "user_123",
            "groups": ["support-leads"],
            "agent_role": "customer_support",
        },
        "knowledge": {
            "enabled": True,
            "source_ids": [source["source"]["source_id"]],
            "top_k": 3,
            "min_score": 0.2,
        },
    },
)

print(completion.get("driftgard", {}).get("knowledge", {}).get("retrieval_run_id"))
print(completion.get("driftgard", {}).get("knowledge", {}).get("citations"))
print(completion.get("driftgard", {}).get("knowledge", {}).get("principal"))

Smoke test the Knowledge SDK helpers

Run the included smoke script against a real project. The script creates a temporary upload source, adds restricted governance metadata, uploads a small document, searches it with a matching principal, fetches the retrieval run, and replays it.

export DRIFTGARD_API_KEY=dg_your_project_api_key
export DRIFTGARD_PROJECT_ID=proj_your_project_id
export DRIFTGARD_BASE_URL=https://api.driftgard.com

python3 scripts/test_knowledge_sdk.py

To also test gateway knowledge injection, enable the optional gateway step:

DRIFTGARD_TEST_GATEWAY=1 DRIFTGARD_TEST_MODEL=gpt-4o-mini python3 scripts/test_knowledge_sdk.py

Model governance responses

Project settings can restrict evaluations to approved model IDs. SDK calls do not change: continue sending model_id as usual.

Remote mode enforces this on the DriftGard API. Local and local-with-audit modes fetch project runtime config during init() and enforce the same model governance policy locally before returning the verdict.

If the project is in approved-model-only mode and the submitted model_id is not approved, the response may include:

{
  "clause_id": "MODEL_NOT_APPROVED",
  "category": "model_governance",
  "severity": "medium",
  "action": "flag",
  "source": "model_policy"
}

With allow and flag, result["evaluation"]["allowed"] can still be True. With block, it will be False and the fallback/decision explanation behaves like other policy blocks. Treat result["evaluation"]["violations"] as the full governance finding list, not only content-rule matches.

Decision metadata

Project settings can append structured JSON metadata to an evaluation based on the final risk score. SDK calls do not change. When a rule matches, read it from result["evaluation"]["decision_metadata"]:

result = dg.evaluate(
    project_id="your-project-id",
    prompt=prompt,
    response=response,
    model_id="gpt-4o-mini",
)

if result["evaluation"].get("decision_metadata", {}).get("requires_review"):
    # App-specific workflow: step-up auth, reviewer route, UI notice, etc.
    pass

Example response field:

{
  "decision_metadata": {
    "risk_tier": "high",
    "requires_review": true,
    "ui_notice": "Show a safer-workflow notice before continuing"
  }
}

Decision metadata is advisory. It does not change allowed, risk_score, or violations. Remote mode applies it on the DriftGard API. Local and local-with-audit modes fetch the same project runtime config during init() and apply matching metadata locally.

Local evaluation mode (beta)

For privacy-sensitive deployments — mental health, clinical, healthcare — where no patient data can leave your environment. The SDK evaluates locally via a compiled WebAssembly engine. No prompt, response, or conversation content is sent to DriftGard.

Requires Node.js 18+ installed (the WASM engine uses a Node.js subprocess bridge).

Local mode — zero network calls after init

from driftgard import Driftgard

dg = Driftgard(
    api_key="your-api-key",
    mode="local",
    project_id="your-project-id",
)

# Fetches the active control pack (one-time network call)
dg.init()

# All evaluate() calls now run locally via WASM
result = dg.evaluate(
    project_id="your-project-id",
    prompt="I feel really anxious today",
    response="I hear you. Would you like to talk about what's triggering it?",
    model_id="gpt-4o",
)

print(result["evaluation"]["allowed"])   # True
print(result["decision_source"])         # "local"
print(result["data_mode"])              # "local"

# Clean up when done
dg.destroy()

Local-with-audit mode — local evaluation, metadata reporting

Same as local mode, but posts verdict metadata (no prompt/response) to DriftGard for compliance dashboards:

dg = Driftgard(
    api_key="your-api-key",
    mode="local-with-audit",
    project_id="your-project-id",
)

dg.init()

result = dg.evaluate(
    project_id="your-project-id",
    prompt="Patient conversation content...",
    response="Clinical response...",
    model_id="gpt-4o",
    agent_role="therapist_agent",
)

# Evaluation ran locally — no patient data sent
# Only verdict metadata reported: evaluation_id, timestamp, allowed, risk_score,
# violation clause IDs, severities, model_id, session_id, agent_role

Control pack sync

On init(), the SDK fetches the active control pack for your project and caches it in memory. A background refresh runs every 60 seconds (configurable). If a refresh fails, the SDK uses the last-known-good pack and marks it as stale.

dg = Driftgard(
    api_key="your-api-key",
    mode="local",
    project_id="your-project-id",
    refresh_interval_seconds=120,  # refresh every 2 minutes (default 60)
    on_control_pack_refresh=lambda e: print(f"Refresh: {e}"),
)

When to use each mode

Mode	Data sent to DriftGard	Use case
remote (default)	Prompt + response + verdict	Standard deployment, full dashboard visibility
local	Control pack fetch only (on init)	Maximum privacy — mental health, clinical, sovereign
local-with-audit	Control pack fetch + verdict metadata, including decision_metadata when configured	Privacy with compliance reporting — healthcare, regulated

Local semantic matching (ONNX)

By default, local mode only runs Layer 1 (pattern matching). Enable local_semantic=True to add Layer 2 (semantic similarity) locally via a quantized ONNX model — no data leaves your environment.

# Install required dependencies
pip install onnxruntime tokenizers

dg = Driftgard(
    api_key="your-api-key",
    mode="local",
    project_id="your-project-id",
    local_semantic=True,  # enables ONNX-based semantic matching
)

# First init() downloads the model (~22MB) and caches it
dg.init()

# Evaluations now run both Layer 1 (patterns) and Layer 2 (semantic)
result = dg.evaluate(...)

The model is downloaded once to .cache/driftgard/ in your working directory and reused on subsequent runs. You can also provide a custom path:

dg = Driftgard(
    ...
    local_semantic=True,
    local_semantic_model_path="/path/to/your/model.onnx",
)

Feature	local_semantic=False (default)	local_semantic=True
Pattern matching (Layer 1)	✓	✓
Semantic matching (Layer 2)	✗	✓
Model size	0	~22MB (one-time download)
Extra dependencies	None	onnxruntime, tokenizers
Coverage vs remote	~60%	~80%

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.

Jurisdiction-scoped rules

Control pack rules can be scoped to specific jurisdictions. Pass the user's jurisdiction in the evaluate request — only matching rules (plus global rules) will fire:

result = dg.evaluate(
    project_id="your-project-id",
    prompt="What odds can I get?",
    response="Current odds for the Melbourne Cup are...",
    model_id="gpt-4o",
    jurisdiction="AU-VIC",  # only VIC + global rules fire
)

Rules without a jurisdictions field are global — they fire for all requests regardless of jurisdiction. Rules with jurisdictions: ["AU-VIC", "AU-NSW"] only fire when the request's jurisdiction matches.

Supported jurisdiction codes include:

• Australia: AU, AU-ACT, AU-NSW, AU-NT, AU-QLD, AU-SA, AU-TAS, AU-VIC, AU-WA
• United States: US, US-AL, US-AK, US-AZ, US-AR, US-CA, US-CO, US-CT, US-DE, US-FL, US-GA, US-HI, US-ID, US-IL, US-IN, US-IA, US-KS, US-KY, US-LA, US-ME, US-MD, US-MA, US-MI, US-MN, US-MS, US-MO, US-MT, US-NE, US-NV, US-NH, US-NJ, US-NM, US-NY, US-NC, US-ND, US-OH, US-OK, US-OR, US-PA, US-RI, US-SC, US-SD, US-TN, US-TX, US-UT, US-VT, US-VA, US-WA, US-WV, US-WI, US-WY, US-DC
• United Kingdom: GB, GB-ENG, GB-SCT, GB-WLS, GB-NIR
• Europe: EU, DE, FR, IE, NL, ES, IT, SE
• Asia-Pacific: NZ, SG, JP, IN, HK
• Other: CA, BR, ZA, AE, SA

Custom codes are also supported — use any string your team agrees on.

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
)

Experiments work in remote and local-with-audit modes. In local-with-audit mode, the experiment_id is included in the verdict metadata sent to DriftGard, so experiment comparisons appear on the dashboard without transmitting prompt/response content. In pure local mode, no data reaches the server so experiments won't appear on the dashboard.

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
• Local evaluation mode (beta) — evaluate via WASM, no data leaves your environment
• Three modes: remote, local, local-with-audit
• Control pack sync with background refresh and stale-pack fallback
• 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)

    # Local mode options (beta)
    mode="remote",                               # "remote" | "local" | "local-with-audit" (default "remote")
    project_id="your-project-id",               # required for local/local-with-audit modes
    refresh_interval_seconds=60,                 # control pack refresh interval (default 60)
    on_control_pack_refresh=lambda e: print(e),  # callback on refresh success/failure
)

# For local modes, call init() to fetch the control pack
dg.init()

Live Approval timeout

If you have Live Approval enabled (Settings → HITL → Live Approval), the evaluate request may be held for up to 60 seconds while waiting for a human reviewer. Set your timeout accordingly:

dg = Driftgard(
    api_key=os.environ["DG_API_KEY"],
    timeout=90,  # 90 seconds — allows 60s live review + 30s buffer
)

Normal evaluations still complete in <100ms. The timeout only matters if no response arrives at all (dead connection).

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
# "local"             — local WASM evaluation
# "local_stale"       — local evaluation with stale control pack
# "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)

Framework Integrations

LangChain

Response Guardrail

Use DriftGardGuardrail as a step in your LangChain chain. It evaluates the LLM output against your control pack and either passes it through, returns a sanitized version, or raises on block.

from driftgard.integrations.langchain import DriftGardGuardrail

guardrail = DriftGardGuardrail(
    api_key="your-api-key",
    project_id="your-project-id",
    model_id="langchain",       # optional — for tracking
    on_block="raise",           # "raise" (ValueError) or "fallback" (return fallback message)
    fail_open=True,             # allow through if DriftGard unreachable
)

# Use in a LangChain chain via pipe:
chain = prompt | llm | guardrail | output_parser

# Or standalone:
try:
    safe = guardrail.invoke("AI response text here")
    print(safe)  # original text, or sanitized version if redaction applied
except ValueError as e:
    print("Blocked:", e)

The guardrail handles three outcomes:

• Allowed: returns the original text unchanged
• Sanitized: returns sanitized_response (PII/patterns redacted with [REDACTED])
• Blocked: raises ValueError (or returns fallback message if on_block="fallback")

Tool Guard

Wrap LangChain agent tools with DriftGard policy validation:

from driftgard.integrations.langchain import DriftGardToolGuard

tool_guard = DriftGardToolGuard(
    api_key="your-api-key",
    project_id="your-project-id",
    agent_role="payments_agent",
)

# Wrap a LangChain StructuredTool:
guarded_tool = tool_guard.wrap(transfer_money_tool)
# Validates tool name + params against your CP's tool_rules before execution
# Blocked → raises ValueError | Allowed → tool executes normally

CrewAI

Response Guardrail

Evaluate agent text output via DriftGard:

from driftgard.integrations.crewai import DriftGardCrewGuardrail

guardrail = DriftGardCrewGuardrail(
    api_key="your-api-key",
    project_id="your-project-id",
    on_block="raise",  # or "fallback"
)

# Evaluate any agent output:
safe_output = guardrail.evaluate(
    prompt="What's my coverage?",
    response=agent_response_text,
)
# Blocked → raises ValueError with fallback message
# Sanitized → returns redacted version
# Allowed → returns original response

Tool Guard

Use the driftgard_tool_guard decorator to wrap CrewAI tool functions with DriftGard policy checks:

from driftgard.integrations.crewai import driftgard_tool_guard

@driftgard_tool_guard(
    api_key="your-api-key",
    project_id="your-project-id",
    agent_role="payments_agent",   # optional — for identity rules
    agent_id="agent_pay_001",      # optional — for identity rules
    fail_open=True,                # allow through if DriftGard unreachable
)
def transfer_money(amount: float, to_account: str):
    """Transfer money to an account."""
    return {"status": "success", "amount": amount}

# Usage — raises ValueError if blocked by policy
try:
    result = transfer_money(amount=500, to_account="account_123")
except ValueError as e:
    print("Blocked:", e)

The decorator:

• Extracts the function name as the tool name
• Maps positional/keyword args to parameters
• Evaluates against your control pack's tool_rules before execution
• Raises ValueError with the fallback message if blocked

Strands (AWS Agent Core)

Full integration — both response guardrail and tool call validation in one hook:

from driftgard.integrations.strands import DriftgardHook

agent = Agent(
    model="us.anthropic.claude-sonnet-4-20250514",
    tools=[transfer_money, send_email],
    hooks=[DriftgardHook(
        api_key="your-api-key",
        project_id="your-project-id",
        agent_role="payments_agent",
    )],
)

# Handles both:
# - BeforeToolCallEvent → validates tool name + params against tool_rules
# - AfterModelInferenceEvent → evaluates response text against policy_rules

Requirements

• Python 3.8+
• requests library
• Node.js 18+ (for local evaluation mode only)
• API key from DriftGard (Settings → API Keys)
• Compliance or Enterprise tier for API evaluation

License

MIT