Build AgentGate — a secure gateway that lets LLM agents call internal tools (databases, APIs, shell commands) with per-tool policy enforcement, rate limiting, and a full audit trail of every action taken. TypeScript + Express.

AI Agent Tool Gateway

What you'll build: AgentGate — a gateway that sits between an LLM agent (Claude, GPT-4, etc.) and your internal tools. Every tool call is authenticated, policy-checked, rate-limited, and audited. Agents can read databases but not write them. Agents can call APIs but not with admin credentials. The prompt is visible in the audit log. TypeScript + Express.

The Story

Automata Labs gives enterprise customers AI agents that can query their CRM, run SQL reports, and send Slack messages. The problem: agents make mistakes. An agent that can call DELETE FROM orders is a liability. An agent that can send Slack to #all-company without approval is a PR disaster.

AgentGate enforces:

Every tool call carries a signed JWT from the agent runtime
Rego policy decides which tools an agent can call based on its agent_tier and tenant_id
Rate limiting per tenant per tool — run_sql max 10/min, send_slack max 5/min
Full audit trail: agent ID, tenant, tool name, input (redacted), outcome, latency
Dangerous tools (delete_records, admin_api) require a second human-in-the-loop approval claim

Architecture

LLM Agent (Claude / GPT-4)
         │
         │  POST /tools/{name}
         │  Authorization: Bearer <agent-JWT>
         │  Body: { "input": {...} }
         ▼
┌──────────────────────────────────────┐
│           AgentGate (Express)        │
│                                      │
│  CoreSDK JWT Auth                    │  ← validates agent identity
│       │                              │
│  Rego Policy: canCallTool?           │  ← checks agent_tier + tool
│       │                              │
│  Rate limiter (per tenant+tool)      │
│       │                              │
│  Tool executor                       │
│       │                              │
│  Audit: input+output+latency         │
└──────────────────────────────────────┘
         │  gRPC :50051
         ▼
┌─────────────────┐
│  CoreSDK Sidecar│  ← JWT + policy evaluation
└─────────────────┘

Prerequisites

npm install @coresdk/sdk express zod

Quickstart

git clone https://github.com/coresdk-dev/examples
cd examples/typescript/agentgate
CORESDK_USE_MOCK=true npx ts-node main.ts

Output:

AgentGate — Tool Gateway Test Suite
=====================================

▸ Authentication
  [PASS]  Missing token → 401
  [PASS]  Invalid token → 401
  [PASS]  Valid agent token → tool call proceeds

▸ Policy Enforcement
  [PASS]  free-tier agent: run_sql allowed (read-only)
  [PASS]  free-tier agent: delete_records → 403 (policy denied)
  [PASS]  pro-tier agent: admin_api allowed with approval claim
  [PASS]  pro-tier agent: admin_api without approval → 403

▸ Rate Limiting
  [PASS]  11th run_sql in 1 min → 429 Too Many Requests
  [PASS]  Rate limit resets after window

▸ Audit Trail
  [PASS]  Tool call recorded with agent_id, tenant, input hash
  [PASS]  Failed calls recorded with denial reason

10/10 passed

Code Walkthrough

Step 1 — SDK setup and Express middleware

import { SDK } from '@coresdk/sdk'
import express, { Request, Response, NextFunction } from 'express'

const sdk = SDK.fromEnv()
const app = express()
app.use(express.json())

// Auth middleware: validate agent JWT, attach claims
async function requireAgentAuth(req: Request, res: Response, next: NextFunction) {
  const token = req.headers.authorization?.replace('Bearer ', '')
  if (!token) {
    return res.status(401).json({
      type: 'https://agentgate.example/errors/unauthorized',
      title: 'Unauthorized',
      status: 401,
      detail: 'Agent JWT required in Authorization header.',
    })
  }

  try {
    const decision = await sdk.authorize(token, req.path, req.method)
    if (!decision.allowed) {
      return res.status(401).json({
        type: 'https://agentgate.example/errors/unauthorized',
        title: 'Unauthorized',
        status: 401,
        detail: decision.reason ?? 'Token rejected by sidecar.',
      })
    }
    res.locals.claims = decision.claims
    next()
  } catch (err: any) {
    // fail_mode=closed: propagate error
    return res.status(401).json({ status: 401, title: 'Unauthorized', detail: err.message })
  }
}

Step 2 — Rego policy for tool access

Policy file policies/agent_tools.rego:

package agentgate

import future.keywords.if

# Default: deny
default allow := false

# Free tier: read-only tools only
allow if {
    input.agent_tier == "free"
    input.tool_name in {"run_sql_read", "get_crm_contact", "fetch_report"}
}

# Pro tier: write tools allowed
allow if {
    input.agent_tier == "pro"
    input.tool_name in {"run_sql_read", "run_sql_write", "get_crm_contact",
                        "update_crm_contact", "fetch_report", "send_slack"}
}

# Dangerous tools: require explicit human approval claim
allow if {
    input.agent_tier == "pro"
    input.tool_name in {"delete_records", "admin_api"}
    input.human_approved == true
}

Evaluate in the gateway:

async function checkToolPolicy(
  toolName: string,
  claims: any
): Promise<{ allowed: boolean; reason?: string }> {
  const input = {
    agent_tier: claims.agentTier ?? 'free',
    tool_name: toolName,
    tenant_id: claims.tenantId,
    human_approved: claims.humanApproved ?? false,
  }

  const allowed = await sdk.evaluatePolicy('agentgate.allow', input)
  return {
    allowed,
    reason: allowed ? undefined : `Policy denied '${toolName}' for tier '${input.agent_tier}'`,
  }
}

Step 3 — Rate limiting per tenant + tool

// Sliding window counter: { "acme-corp:run_sql": [timestamp, ...] }
const rateLimitWindows = new Map<string, number[]>()

const RATE_LIMITS: Record<string, { max: number; windowSec: number }> = {
  run_sql_read:       { max: 10,  windowSec: 60 },
  run_sql_write:      { max: 5,   windowSec: 60 },
  send_slack:         { max: 5,   windowSec: 60 },
  get_crm_contact:    { max: 50,  windowSec: 60 },
  update_crm_contact: { max: 20,  windowSec: 60 },
  admin_api:          { max: 2,   windowSec: 60 },
}

function checkRateLimit(tenantId: string, toolName: string): boolean {
  const limit = RATE_LIMITS[toolName]
  if (!limit) return true

  const key = `${tenantId}:${toolName}`
  const now = Date.now()
  const windowMs = limit.windowSec * 1000
  const calls = (rateLimitWindows.get(key) ?? []).filter(t => now - t < windowMs)

  if (calls.length >= limit.max) return false

  calls.push(now)
  rateLimitWindows.set(key, calls)
  return true
}

Step 4 — Tool executor

const TOOLS: Record<string, (input: unknown) => Promise<unknown>> = {
  run_sql_read: async (input: any) => {
    // In production: execute against read replica with row-level security
    return { rows: [{ id: 1, name: 'Acme Corp', arr: 120000 }], count: 1 }
  },

  get_crm_contact: async (input: any) => {
    return { id: input.contact_id, name: 'Alice Smith', email: 'alice@acme.com' }
  },

  send_slack: async (input: any) => {
    // In production: validate channel is allowed for this tenant
    return { ok: true, message_ts: '1234567890.123456' }
  },
}

// Main tool endpoint
app.post('/tools/:toolName', requireAgentAuth, async (req, res) => {
  const { toolName } = req.params
  const claims = res.locals.claims
  const tenantId = claims.tenantId ?? 'unknown'
  const t0 = Date.now()

  // 1. Policy check
  const { allowed, reason } = await checkToolPolicy(toolName, claims)
  if (!allowed) {
    await audit({ tenantId, toolName, claims, outcome: 'denied', reason, latencyMs: Date.now() - t0 })
    return res.status(403).json({
      type: 'https://agentgate.example/errors/forbidden',
      title: 'Forbidden',
      status: 403,
      detail: reason,
    })
  }

  // 2. Rate limit
  if (!checkRateLimit(tenantId, toolName)) {
    await audit({ tenantId, toolName, claims, outcome: 'rate_limited', latencyMs: Date.now() - t0 })
    return res.status(429).json({
      type: 'https://agentgate.example/errors/rate-limited',
      title: 'Too Many Requests',
      status: 429,
      detail: `Rate limit exceeded for tool '${toolName}'.`,
      retry_after: 60,
    })
  }

  // 3. Execute
  const tool = TOOLS[toolName]
  if (!tool) {
    return res.status(404).json({ status: 404, title: 'Tool Not Found' })
  }

  try {
    const result = await tool(req.body.input)
    const latencyMs = Date.now() - t0
    await audit({ tenantId, toolName, claims, outcome: 'success', latencyMs })
    return res.json({ tool: toolName, result, latency_ms: latencyMs })
  } catch (err: any) {
    const latencyMs = Date.now() - t0
    await audit({ tenantId, toolName, claims, outcome: 'error', reason: err.message, latencyMs })
    return res.status(500).json({ status: 500, title: 'Tool Error', detail: err.message })
  }
})

Step 5 — Audit trail (never skip this)

interface AuditEntry {
  timestamp: string
  agentId: string
  tenantId: string
  tool: string
  outcome: 'success' | 'denied' | 'rate_limited' | 'error'
  reason?: string
  latencyMs: number
  inputHash: string  // SHA-256 of input — never log raw input (may contain PII)
}

const auditLog: AuditEntry[] = []

async function audit(params: {
  tenantId: string; toolName: string; claims: any
  outcome: AuditEntry['outcome']; reason?: string; latencyMs: number
}) {
  auditLog.push({
    timestamp: new Date().toISOString(),
    agentId: params.claims.sub ?? 'unknown',
    tenantId: params.tenantId,
    tool: params.toolName,
    outcome: params.outcome,
    reason: params.reason,
    latencyMs: params.latencyMs,
    inputHash: 'sha256:...',  // hash of req.body in production
  })
}

Calling AgentGate from Claude

import anthropic
import httpx

client = anthropic.Anthropic()
gateway_url = "http://localhost:3000"
agent_token = "Bearer <agent-jwt>"

tools = [
    {
        "name": "run_sql_read",
        "description": "Execute a read-only SQL query against the CRM database",
        "input_schema": {
            "type": "object",
            "properties": {"query": {"type": "string"}},
            "required": ["query"],
        },
    },
]

def call_tool(name: str, input: dict) -> str:
    r = httpx.post(
        f"{gateway_url}/tools/{name}",
        json={"input": input},
        headers={"Authorization": agent_token},
    )
    return r.json()

# Agentic loop
messages = [{"role": "user", "content": "How many enterprise customers do we have?"}]
while True:
    response = client.messages.create(
        model="claude-opus-4-6", max_tokens=1024, tools=tools, messages=messages
    )
    if response.stop_reason == "end_turn":
        print(response.content[0].text)
        break
    for block in response.content:
        if block.type == "tool_use":
            result = call_tool(block.name, block.input)
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": [
                {"type": "tool_result", "tool_use_id": block.id, "content": str(result)}
            ]})
            break

Environment Variables

Variable	Default	Description
`CORESDK_SIDECAR_ADDR`	`localhost:50051`	Sidecar address
`CORESDK_FAIL_MODE`	`open`	`closed` recommended for production
`CORESDK_POLICY_BUNDLE`	—	Path to `agent_tools.rego` bundle
`PORT`	`3000`	Gateway listen port

Full Source

→ examples/typescript/agentgate/

cd examples/typescript/agentgate
CORESDK_USE_MOCK=true npx ts-node main.ts

AI Agent Tool Gateway

On this page