KyberShield API Reference

Overview

KyberShield sits between your AI agent and your database. Every query passes through the gateway, which evaluates it for threats, applies your policies, signs the verdict with HMAC-SHA256, and returns results — or blocks the query.

AI Agent → KyberShield Gateway → Your Database
             ↓
   SQL injection detection
   Policy engine evaluation
   Risk scoring (0-100)
   HMAC-SHA256 verdict signing (ML-DSA-87 on roadmap)
             ↓
   ✓ Allowed → execute & return rows
   🚫 Blocked → return reason + risk score

Authentication

All API requests require an agent API key. Generate API keys in the KyberShield portal under the Agents section.

Recommended: environment variable

export KYBERSHIELD_AGENT_KEY=ks_your_api_key_here

Both SDKs read this automatically — no need to pass the key in code.

Alternative: HTTP header

X-Agent-Key: ks_your_api_key_here

Quickstart

Install the SDK, set your key, and run your first protected query in under 30 seconds.

from kybershield import KyberShield

ks = KyberShield()  # reads KYBERSHIELD_AGENT_KEY from env

result = ks.query("SELECT name, email FROM users WHERE active = true")
print(result.rows)          # [{"name": "Alice", "email": "alice@co.com"}, ...]
print(result.risk_score)    # 5

import KyberShield from "kybershield";

const ks = new KyberShield();  // reads KYBERSHIELD_AGENT_KEY from env

const result = await ks.query("SELECT name, email FROM users WHERE active = true");
console.log(result.rows);            // [{name: "Alice", ...}, ...]
console.log(result.riskScore);       // 5

curl -X POST https://app.kybershield.ai/api/gateway/query \
  -H "Content-Type: application/json" \
  -H "X-Agent-Key: $KYBERSHIELD_AGENT_KEY" \
  -d '{"query": "SELECT 1 AS ping"}'

Python SDK

Zero-config initialization

from kybershield import KyberShield

# Reads KYBERSHIELD_AGENT_KEY from environment automatically
ks = KyberShield()

# Or pass explicitly
ks = KyberShield(agent_key="ks_your_api_key_here")

Configuration

Methods

QueryResult fields

Error handling

from kybershield import KyberShield, BlockedError, KyberShieldError

ks = KyberShield()

try:
    result = ks.query("DROP TABLE users")
except BlockedError as e:
    print(e.reason)      # "Destructive queries blocked"
    print(e.risk_score)  # 95
except KyberShieldError as e:
    print(e)  # Connection errors, timeouts, etc.

Context manager

with KyberShield() as ks:
    result = ks.query("SELECT 1 AS ping")
    print(result.rows)

Self-hosted gateway

ks = KyberShield(
    agent_key="ks_your_key",
    gateway_url="https://your-gateway.internal.com"
)

Node.js / TypeScript SDK

Zero-config initialization

import KyberShield from "kybershield";

// Reads KYBERSHIELD_AGENT_KEY from environment automatically
const ks = new KyberShield();

// Or pass explicitly
const ks = new KyberShield({ agentKey: "ks_your_api_key_here" });

Configuration

Methods

TypeScript types

interface QueryResult {
  rows: Record<string, unknown>[];
  rowCount: number;
  queryType: string;
  riskScore: number;
  verdict: string;
  signature: string;          // HMAC-SHA256 hex; ML-DSA-87 on roadmap
  executionTimeMs: number;
  raw: Record<string, unknown>;
}

Error handling

import KyberShield, { BlockedError, KyberShieldError } from "kybershield";

const ks = new KyberShield();

try {
  await ks.query("DROP TABLE users");
} catch (err) {
  if (err instanceof BlockedError) {
    console.log(err.reason);    // "Destructive queries blocked"
    console.log(err.riskScore); // 95
  } else if (err instanceof KyberShieldError) {
    console.log(err.message);
  }
}

Self-hosted gateway

const ks = new KyberShield({
  agentKey: "ks_your_key",
  gatewayUrl: "https://your-gateway.internal.com",
});

Wire-Protocol Proxy ZERO CODE

The simplest way to put KyberShield in front of any database: swap your connection string. No SDK install, no code changes, works with every framework and ORM that speaks Postgres — Django, SQLAlchemy, Prisma, Drizzle, Rails, Laravel, Hibernate, psql, BI tools, or any AI agent framework.

Before / after

# Before
DATABASE_URL=postgres://app_user:p4ss@your-db.internal:5432/app

# After (point at the proxy, password is now your KyberShield API key)
DATABASE_URL=postgres://ks_your_api_key_here@localhost:5433/app

Run it

docker run --rm -p 5433:5433 \
  -e KS_UPSTREAM_DATABASE_URL=postgres://app_user:p4ss@your-db.internal:5432/app \
  -e KS_GATEWAY_URL=https://app.kybershield.ai \
  kybershield/proxy:latest

Configuration

Operating modes

• Sidecar — one proxy per app instance, talking to localhost. Recommended for self-hosted.
• Centralized — one proxy fleet behind an internal load balancer, all apps point at it.
• Hosted (coming soon) — gateway.kybershield.ai:5432, no infra to run.

Docker Sidecar ZERO CODE

The same wire-protocol proxy, packaged as a sidecar container. Drop it next to your app in Docker Compose, Kubernetes, or ECS — your app talks to localhost:5433, the sidecar handles security and forwards to the real DB.

Docker Compose

services:
  app:
    image: your-app:latest
    environment:
      DATABASE_URL: postgres://${KS_API_KEY}@kybershield:5433/app
    depends_on: [kybershield]

  kybershield:
    image: kybershield/proxy:latest
    environment:
      KS_UPSTREAM_DATABASE_URL: postgres://app_user:p4ss@your-db:5432/app
      KS_GATEWAY_URL: https://app.kybershield.ai
    ports: ["5433:5433"]

Kubernetes (sidecar in the same pod)

spec:
  containers:
    - name: app
      image: your-app:latest
      env:
        - name: DATABASE_URL
          value: postgres://$(KS_API_KEY)@127.0.0.1:5433/app
    - name: kybershield-proxy
      image: kybershield/proxy:latest
      env:
        - name: KS_UPSTREAM_DATABASE_URL
          valueFrom:
            secretKeyRef: { name: db, key: url }
        - name: KS_GATEWAY_URL
          value: https://app.kybershield.ai

n8n Connector NO CODE

For teams building AI workflows in n8n, drop a KyberShield node between any AI step and the rest of your workflow. No code, drag-and-drop.

Install

Operations

Typical workflow

Webhook (user message)
   ↓
KyberShield: Check Prompt        ← two outputs: Allowed / Blocked
   ↓ (Allowed)
OpenAI Chat
   ↓
KyberShield: Check Response      ← redacts PII before downstream
   ↓
Slack: Post message

Zapier Connector NO CODE

Add a KyberShield action to any Zap. Useful for non-engineering teams running AI automations across email, Slack, CRMs, and chat.

Install

Search the Zapier App Directory for KyberShield and connect with your agent API key.

Actions

Typical Zap

1. Trigger: Webhook (incoming user message)
2. Action: KyberShield — Check Prompt for Injection
3. Filter: only continue if verdict = "allow"
4. Action: OpenAI — Send to ChatGPT
5. Action: KyberShield — Redact PII & Secrets from Response
6. Action: Slack — Post the redacted message

POST /api/gateway/query

Submit a SQL query for security evaluation and (optional) execution against your connected database. This is the raw HTTP endpoint that both SDKs wrap.

{
  "query": "SELECT name, email FROM users WHERE active = true"
}

{
  "verdict": "allowed",
  "queryId": "q_abc123",
  "riskScore": 5,
  "data": [
    {"name": "Alice", "email": "alice@company.com"},
    {"name": "Bob", "email": "bob@company.com"}
  ],
  "rowsReturned": 2,
  "signature": "6e20ffa5a5211229845207d53796d5244ddcb7872122bd951ceae02a05b103cb"
}

{
  "verdict": "blocked",
  "queryId": "q_def456",
  "riskScore": 95,
  "reason": "Destructive queries blocked by gateway settings",
  "threats": ["destructive_operation"]
}

{
  "verdict": "blocked",
  "reason": "Rate limit exceeded: 60 queries/minute",
  "riskScore": 60
}

GET /api/gateway/health

Check the health status of the gateway.

{
  "status": "healthy",
  "version": "1.0.0",
  "uptime": 86400
}

OpenAI Function Calling

Use KyberShield as a tool in OpenAI's function-calling interface. Every SQL query your GPT agent generates gets protected automatically.

import openai, json
from kybershield import KyberShield

ks = KyberShield()  # reads KYBERSHIELD_AGENT_KEY from env

tools = [{
    "type": "function",
    "function": {
        "name": "secure_query",
        "description": "Execute a SQL query securely via KyberShield",
        "parameters": {
            "type": "object",
            "properties": {
                "sql": {"type": "string", "description": "The SQL query to execute"}
            },
            "required": ["sql"]
        }
    }
}]

def handle_tool_call(tool_call):
    args = json.loads(tool_call.function.arguments)
    result = ks.query(args["sql"])
    return json.dumps({"rows": result.rows, "risk_score": result.risk_score})

response = openai.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Show me all active users"}],
    tools=tools
)

if response.choices[0].message.tool_calls:
    for tc in response.choices[0].message.tool_calls:
        output = handle_tool_call(tc)
        print(output)

LangChain Integration

KyberShield includes a built-in LangChain tool. Install with the langchain extras and use it in one line.

from kybershield.langchain import create_query_tool

tool = create_query_tool()  # reads KYBERSHIELD_AGENT_KEY from env

# Use directly in your LangChain agent
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")
# agent = create_openai_tools_agent(llm, [tool], prompt)
# executor = AgentExecutor(agent=agent, tools=[tool])

from langchain.tools import tool
from kybershield import KyberShield

ks = KyberShield()

@tool
def secure_query(sql: str) -> dict:
    """Execute a SQL query securely through KyberShield gateway."""
    result = ks.query(sql)
    return {"rows": result.rows, "risk_score": result.risk_score}

tools = [secure_query]

import { DynamicTool } from "langchain/tools";
import KyberShield from "kybershield";

const ks = new KyberShield();  // reads KYBERSHIELD_AGENT_KEY from env

const secureQuery = new DynamicTool({
  name: "secure_query",
  description: "Execute a SQL query securely through KyberShield",
  func: async (sql) => {
    const result = await ks.query(sql);
    return JSON.stringify(result.rows);
  },
});

Vercel AI SDK

Use KyberShield with the Vercel AI SDK's tool() helper for Next.js and Edge functions.

import { tool } from "ai";
import { z } from "zod";
import KyberShield from "kybershield";

const ks = new KyberShield();  // reads KYBERSHIELD_AGENT_KEY from env

const secureQuery = tool({
  description: "Execute a SQL query securely through KyberShield gateway",
  parameters: z.object({
    sql: z.string().describe("The SQL query to execute"),
  }),
  execute: async ({ sql }) => {
    const result = await ks.query(sql);
    return { rows: result.rows, riskScore: result.riskScore };
  },
});

OpenClaw Integration

KyberShield provides an official ClawHub plugin that registers a kybershield_query tool and bundles a skill for your OpenClaw agent.

# Set your agent key
export KYBERSHIELD_AGENT_KEY=ks_your_agent_key_here

# Restart the gateway
openclaw gateway restart

{
  "plugins": {
    "entries": {
      "kybershield": {
        "enabled": true,
        "config": {
          "agentKey": "ks_your_agent_key_here"
        }
      }
    }
  }
}

# Set your agent key
export KYBERSHIELD_AGENT_KEY=ks_your_agent_key_here

# Optional: install the Python SDK for faster queries
pip install kybershield

How it works

1. Your OpenClaw agent generates a SQL query as part of a task
2. Plugin: The before_tool_call hook intercepts SQL tool calls (sql_query, db_query, etc.) and routes them through the gateway — the original tool call is replaced with the KyberShield-routed result
3. If the gateway allows the query, the routed result (rows, risk score, verdict) is returned to the agent
4. If the gateway blocks the query, the agent receives the block reason and risk score
5. If the gateway is unreachable, the query is blocked (fail-closed enforcement)
6. The agent can also call kybershield_query directly for secure SQL access

Environment variables

Claude Managed Agents

Wire KyberShield into Anthropic's Claude Managed Agents with one call. The integration uses MCP (Model Context Protocol), Anthropic's officially supported extension point — no custom runtime, no proxy required.

What you get: four guardrail tools the agent calls automatically — kybershield_check_prompt, kybershield_check_tool_call, kybershield_check_response, kybershield_log_event. Every tool invocation, prompt, and response is screened by the gateway and recorded in the PQ-signed verdict ledger.

from anthropic import Anthropic
from kybershield_claude import secured_agent

client = Anthropic()  # reads ANTHROPIC_API_KEY

agent = secured_agent(
    client,
    model="claude-sonnet-4-5",
    instructions="You are a helpful assistant for ACME Corp.",
    policy="msp-default",          # optional policy bundle id
)

# agent.agent is the standard Anthropic Agent object
session = client.beta.managed_agents.sessions.create(agent_id=agent.agent.id)

{
  "model": "claude-sonnet-4-5",
  "instructions": "You are a helpful assistant.\n\n[KyberShield guardrails — REQUIRED]\nBefore any external tool call, call kybershield_check_tool_call ...",
  "mcp_servers": [
    {
      "name": "kybershield",
      "url": "https://mcp.kybershield.ai/mcp",
      "transport": "http",
      "headers": {
        "x-kybershield-agent-key": "ks_live_...",
        "x-kybershield-policy": "msp-default"
      }
    }
  ]
}

# stdio mode (Claude Desktop, local agent containers)
KYBERSHIELD_AGENT_KEY=ks_live_... kybershield-mcp --transport stdio

# HTTP mode (hosted endpoint your Managed Agents call by URL)
KYBERSHIELD_AGENT_KEY=ks_live_... \
KYBERSHIELD_GATEWAY_URL=https://gateway.acme-internal.example \
kybershield-mcp --transport http --port 8787

Tools exposed

Verdicts: allow (proceed), redact (use the sanitized version returned in redacted / arguments), block (do not proceed; explain to the user). Every call is recorded in the PQ-signed verdict ledger and viewable in the portal.

Environment variables

Packages: kybershield-claude (PyPI), @kybershield/claude-mcp (npm). Source: sdks/claude-python/ and sdks/claude-mcp/.

Webhooks

Receive real-time notifications when the gateway blocks a query. Configure your webhook URL in the KyberShield portal settings.

Event: query.blocked

Fired every time a query is blocked by the gateway.

Headers

Payload

{
  "eventType": "query.blocked",
  "timestamp": "2025-01-15T10:30:00.000Z",
  "agentId": "agent_abc123",
  "agentName": "Data Analyst Bot",
  "queryId": "q_def456",
  "queryType": "DELETE",
  "riskScore": 95,
  "reason": "Destructive queries blocked by gateway settings",
  "threats": ["destructive_operation"],
  "riskFactors": ["destructive_query", "bulk_modification"],
  "query": "DELETE FROM users WHERE 1=1",
  "tables": ["users"]
}

Verifying signatures

import hmac, hashlib

def verify_signature(payload: bytes, signature: str, secret: str) -> bool:
    expected = "sha256=" + hmac.new(
        secret.encode(), payload, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, signature)

Risk Scoring

Every query gets a risk score from 0 to 100 based on multiple factors:

Verdict Signing

KyberShield signs every gateway verdict today using HMAC-SHA256 over the canonical JSON payload (agent key hash, query type, verdict, risk score, timestamp). The signature is returned in every response and persisted to the audit ledger so verdicts are tamper-evident.

// Returned in every gateway response today
{
  "signature": "6e20ffa5a5211229845207d53796d5244ddcb7872122bd951ceae02a05b103cb",
  "verdict": "allow"
}

Policy Engine

Define fine-grained rules to control what each agent can do. Policies are evaluated by priority order on every query.

Policy actions

Policy conditions