Runtime / Agent Integration

Use Lexega to check SQL queries before execution. Pass the SQL string, get a decision. Agents that speak the Model Context Protocol can use the built-in MCP server; everything else calls the CLI.

This guide covers runtime integration for agents and applications. For CI/CD integration (PR reviews, semantic diff), see Integration Options.

Performance: Single-statement analysis typically completes in < 20ms. Multiple semicolon-delimited statements are analyzed together in one call (~75ms for 10 simple statements).

Setup

1. Install the binary

# Download (Linux x86_64 example)
curl -L -o lexega-sql https://github.com/Lexega/releases/releases/latest/download/lexega-sql-linux-x86_64
chmod +x lexega-sql

# Move to PATH (or keep in project directory)
sudo mv lexega-sql /usr/local/bin/

2. Set your license key

export LEXEGA_LICENSE_KEY="your-license-key-here"

Or in your application's environment configuration.

3. Create a policy file

Save as policy.yaml:

schema_version: 1
policy_id: runtime-agent-policy
policy_version: "1.0.0"
policies: []

severity_actions:
  - critical: block
    high: warn

default_action: allow

This blocks any critical signal (like DELETE without WHERE) and warns on high. For more control, add explicit policies:

schema_version: 1
policy_id: runtime-agent-policy
policy_version: "1.0.0"

policies:
  - rule_id: DML-WRITE-UNBOUNDED       # DELETE/UPDATE without WHERE
    action: block

  - rule_id: GRT-ALL-PRIV              # GRANT ALL PRIVILEGES
    action: block

severity_actions:
  - critical: block
    high: warn

default_action: allow

MCP Server (AI Agents)

If your agent speaks the Model Context Protocol — Claude Code, Cursor, or anything built on an MCP-capable framework — run Lexega as an MCP server instead of shelling out per query. The server loads your policy and rules once and stays warm, so each check skips process startup.

Add to your project's .mcp.json (Claude Code) or the equivalent MCP configuration:

{
  "mcpServers": {
    "lexega": {
      "command": "lexega-sql",
      "args": ["mcp", "--policy", "policy.yaml", "--env", "prod"]
    }
  }
}

The agent gets one tool, check_sql. It submits SQL (a single statement or a semicolon-delimited batch), Lexega analyzes it and evaluates your policy, and the agent receives the decision:

{
  "allowed": false,
  "blocked_reason": "Blocked by severity_actions.critical (signal severity = critical)",
  "matched_rules": [
    {
      "rule_id": "_severity_critical",
      "action": "block",
      "severity": "Critical",
      "rule_description": "unbounded write operation detected...",
      "policy_description": "fallback: severity_actions.critical = block",
      "exception_used": null
    }
  ],
  "exceptions_used": [],
  "statements_analyzed": 1,
  "statements_skipped": 0,
  "decision_id": "e458603a656094e8...",
  "decision_path": ".lexega/decisions/decision_f4f8f8c2668ec23b.json"
}

Four properties matter for governance:

• The agent cannot weaken the policy. Policy, environment, and exceptions are fixed when the server starts — the tool only accepts SQL (and an optional per-check dialect). Edit the policy file and restart the server to change them.
• Every check leaves a record. Each call writes a decision artifact to --decision-out (default: .lexega/decisions, the same directory the dashboard reads). Agent-gated SQL gets the same audit trail as CI runs, labeled as runtime decisions. Repeated checks of identical SQL update one artifact rather than accumulating copies.
• Unparsed SQL is visible. statements_skipped counts statements the parser couldn't fully analyze, same as the parse-failure handling below — instruct your agent (or wrap the tool) to treat a non-zero count conservatively.
• dbt models check directly. dbt/Jinja templates are rendered before analysis, exactly as analyze renders them, so an agent working in a dbt project can submit a model source and get the compiled SQL checked. A render failure fails the check closed — it never falls back to analyzing the unrendered template. Pass the render context the same way as the CLI: --var, --var-file, --dbt-profile, --dbt-project, --load-macros. For a server exposed to untrusted callers, --no-render refuses template input instead.

The server accepts the same analysis configuration as analyze: --dialect (default for checks), --custom-rules, --no-builtin, --exceptions, --catalog, the dbt/Jinja render flags above, plus --team/--job-type for policy scoping and --repo/--run-id to override the identity stamped on artifacts. Signals below high severity are filtered before policy evaluation — the same default as analyze, so the CLI and MCP surfaces reach the same decision for the same SQL and policy; change the floor with --min-severity. See lexega-sql mcp --help.

Checking a single model in isolation resolves ref() and source() to placeholders rather than real relations — the same best-effort reading analyze <file> gives. The full cross-model picture comes from analyze -r or a CI run over the project.

Paths like policy.yaml and .lexega/decisions resolve against the working directory your MCP client launches the server in — for Claude Code and Cursor that's the project root. Use absolute paths if your client launches servers elsewhere.

If your application doesn't speak MCP, call the CLI directly:

Python Example

import subprocess
import json
import os

def check_sql(
    sql: str,
    policy_path: str = "policy.yaml",
    env: str = "prod",
  dialect: str = "snowflake",  # e.g. "snowflake", "postgresql", "bigquery", "mysql", "mssql", "databricks"
) -> dict:
    """
    Check SQL against policy. Returns decision dict.
    Raises if lexega-sql not found or license invalid.
    """
    result = subprocess.run(
        [
            "lexega-sql", "analyze", 
            "--stdin",
            "-q",
            "--dialect", dialect,
            "--policy", policy_path,
            "--env", env,
            "--mode", "runtime",
            "--decision-out", "-"
        ],
        input=sql,
        capture_output=True,
        text=True,
        env={**os.environ}  # Inherit LEXEGA_LICENSE_KEY
    )
    
    if result.returncode not in (0, 2):
        # Something went wrong (not found, license error, etc.)
        raise RuntimeError(f"lexega-sql failed: {result.stderr}")
    
    decision = json.loads(result.stdout)
    return {
        "allowed": result.returncode == 0,
        "outcome": decision.get("outcome"),
        "matched_rules": decision.get("analysis", {}).get("matched_rules", []),
    }


# Usage
sql = "DELETE FROM users"  # No WHERE clause
result = check_sql(sql)

if not result["allowed"]:
    print(f"Blocked: {result['matched_rules']}")
    # Don't execute
else:
    # Safe to execute
    execute(sql)

Node.js Example

const { spawn } = require('child_process');

function checkSql(sql, policyPath = 'policy.yaml', env = 'prod', dialect = 'snowflake') {
  return new Promise((resolve, reject) => {
    const proc = spawn('lexega-sql', [
      'analyze', '--stdin', '-q',
      '--dialect', dialect,
      '--policy', policyPath,
      '--env', env,
      '--mode', 'runtime',
      '--decision-out', '-'
    ], {
      env: process.env  // Inherit LEXEGA_LICENSE_KEY
    });

    let stdout = '';
    let stderr = '';

    proc.stdout.on('data', (data) => { stdout += data; });
    proc.stderr.on('data', (data) => { stderr += data; });

    proc.stdin.write(sql);
    proc.stdin.end();

    proc.on('close', (code) => {
      if (code !== 0 && code !== 2) {
        reject(new Error(`lexega-sql failed: ${stderr}`));
        return;
      }
      
      const decision = JSON.parse(stdout);
      resolve({
        allowed: code === 0,
        outcome: decision.outcome,
        matchedRules: decision.analysis?.matched_rules || [],
      });
    });
  });
}

// Usage
async function handleQuery(sql) {
  const result = await checkSql(sql);
  
  if (!result.allowed) {
    console.log('Blocked:', result.matchedRules);
    return;
  }
  
  await execute(sql);
}

Reference

Batching

Pass multiple semicolon-delimited statements in a single call—all are analyzed together:

sql = """
SELECT * FROM users WHERE id = 1;
INSERT INTO audit_log (action) VALUES ('query');
UPDATE users SET last_seen = NOW() WHERE id = 1;
"""
result = check_sql(sql)  # One call, all statements analyzed

No separate batch API needed. If any statement triggers a block, the entire batch is blocked.

Flags

Exit Codes

Decision JSON

The decision JSON contains the full analysis. Key fields for agent integration:

{
  "decision_schema_version": 1,
  "decision_id": "e458603a656094e856aac0003fb04bbf...",
  "timestamp": "2026-02-04T19:34:40.353Z",
  "inputs": {
    "sql_sha256": "c3f483a8bb391787b4515f9afd8f3ede...",
    "policy_id": "baseline",
    "env_context": { "env": "prod" }
  },
  "analysis": {
    "risk_summary": {
      "critical_count": 1,
      "tables_written": 1
    },
    "matched_rules": [
      {
        "rule_id": "_severity_critical",
        "action": "block",
        "rule_description": "unbounded write operation detected..."
      }
    ]
  },
  "outcome": {
    "allowed": false,
    "blocked_reason": "Blocked by severity_actions.critical (signal severity = critical)"
  }
}

Most agents only need outcome.allowed - the exit code (0=allowed, 2=blocked) is the simplest check.

Handling Parse Failures

If the SQL contains syntax the parser doesn't recognize (e.g., dialect-specific features), check analysis.risk_summary.statements_skipped:

decision = json.loads(result.stdout)
skipped = decision.get("analysis", {}).get("risk_summary", {}).get("statements_skipped", 0)

if skipped > 0:
    # Parser couldn't fully analyze this SQL
    # Conservative option: block unknown syntax
    return {"allowed": False, "reason": "Unrecognized SQL syntax"}

This prevents fail-open behavior where unknown SQL slips through unanalyzed.

Read-Only Agent Example

Block all write operations (INSERT/UPDATE/DELETE/MERGE/TRUNCATE) in production - only allow SELECT:

1. Create agent_rules.yaml:

rules:
  - id: AGENT-READONLY
    risk_level: critical
    message: "Write operations not allowed in production"
    triggers:
      kind:
        in: [insert, update, delete, merge, multi_insert, truncate]

2. Create policy.yaml:

schema_version: 1
policy_id: agent-readonly
policy_version: "1.0.0"
policies: []
severity_actions:
  - critical: block
    scope:
      envs: [prod]
default_action: allow

3. Run with both files:

lexega-sql analyze --stdin -q \
  --dialect snowflake \
  --custom-rules agent_rules.yaml \
  --policy policy.yaml \
  --env prod \
  --mode runtime \
  --decision-out -

For PostgreSQL/BigQuery/MySQL/MSSQL/Databricks/Redshift agents, change --dialect snowflake to --dialect postgresql, --dialect bigquery, --dialect mysql, --dialect mssql, --dialect databricks, or --dialect redshift.

Result:

• SELECT * FROM users → allowed
• INSERT INTO users ... → blocked
• UPDATE users SET ... → blocked
• DELETE FROM users ... → blocked
• MERGE INTO users ... → blocked
• TRUNCATE TABLE users → blocked

More Policy Examples

# Strict policy: block critical, warn high
severity_actions:
  - critical: block
    high: warn

# Or be explicit about specific rules:
policies:
  - rule_id: DML-WRITE-UNBOUNDED   # DELETE/UPDATE/MERGE without WHERE
    action: block

  - rule_id: GRT-ALL-PRIV          # GRANT ALL PRIVILEGES
    action: block

  - rule_id: MASK-DROP             # DROP MASKING POLICY
    action: block

  - rule_id: RAP-DROP              # DROP ROW ACCESS POLICY
    action: block

See Policy Reference for full syntax.