Usage & Integration

Audience: customer developers — working code samples for talking to a running vault.svc from outside the kis.ai monorepo. Each example shows the same handful of operations in a different language.

For the full wire-protocol reference, see the HTTP API. For shell-side commands, see the CLI reference.

What we’ll cover in each language

The four operations every integration needs at minimum:

1. Health probe — sanity-check the URL + your TLS material before doing anything sensitive.
2. JWT mint — exchange your mTLS cert for a service JWT.
3. Secret read — fetch a scoped secret.
4. Cert + public-key retrieval — fetch JWT signing material so you can verify tokens from other services.

Adapt from there. Other endpoints follow the same shape (mTLS + JSON body + JSON response).

Common prereqs

You need three files from your ops team, plus the Vault URL:

	Purpose
issuer.crt	The CA bundle that signed vault.svc’s leaf cert. Verifies the server.
client.crt	Your service’s client cert. Identifies your service to vault.svc.
client.key	The private key for client.crt.
VAULT_URL	e.g., https://vault.us-west.example.com:7999

All four are typically delivered through your cluster’s deploy pipeline; never check them into source control.

curl (the universal example)

If you can’t run curl against your vault.svc successfully, no language SDK is going to help — debug at this layer first.

export VAULT_URL=https://vault.us-west.example.com:7999
export VAULT_CA=/etc/myservice/vault/issuer.crt
export VAULT_CERT=/etc/myservice/vault/client.crt
export VAULT_KEY=/etc/myservice/vault/client.key

CURL="curl -sf --cacert $VAULT_CA --cert $VAULT_CERT --key $VAULT_KEY"

# 1. Health
$CURL $VAULT_URL/ready          || { echo "vault not ready"; exit 1; }

# 2. JWT mint
TOKEN=$($CURL $VAULT_URL/certificate/auth | jq -r .token)
echo "JWT: ${TOKEN:0:40}..."

# 3. Secret read — tenant-scoped db password
SCOPE='{"datacenter":"us-west","cluster":"prod-cluster-01","customer":"acme","product":"forge","environment":"prod","tenant":"acme-prod"}'
PASSWORD=$($CURL -X PATCH -H 'Content-Type: application/json' \
    -d "$SCOPE" \
    "$VAULT_URL/secret/db_password" | jq -r .secret)

# 4. JWT verifier cert for the "users" realm of the IAM service
$CURL -X PATCH -H 'Content-Type: application/json' \
    -d "$SCOPE" \
    "$VAULT_URL/jwt/internal/users" | jq .

Python

"""
Minimal Vault client for a Python service.

requires: requests, pyjwt (for JWT verification on the receiver side)
"""
import os
import requests
from pathlib import Path

VAULT_URL  = os.environ["VAULT_URL"]
VAULT_CA   = os.environ["VAULT_CA"]
VAULT_CERT = os.environ["VAULT_CERT"]
VAULT_KEY  = os.environ["VAULT_KEY"]

# A reusable session keeps the TLS connection open across calls.
session = requests.Session()
session.verify = VAULT_CA
session.cert = (VAULT_CERT, VAULT_KEY)

# Your service's scope, populated once at startup.
SCOPE = {
    "datacenter":  "us-west",
    "cluster":     "prod-cluster-01",
    "customer":    "acme",
    "product":     "forge",
    "environment": "prod",
    "tenant":      "acme-prod",
}

def ready() -> bool:
    """Liveness probe."""
    r = session.get(f"{VAULT_URL}/ready")
    return r.ok

def mint_jwt() -> str:
    """Exchange mTLS cert for a short-lived service JWT."""
    r = session.get(f"{VAULT_URL}/certificate/auth")
    r.raise_for_status()
    return r.json()["token"]

def read_secret(name: str) -> str:
    """Read a scoped secret. PATCH is the read verb (carries the scope body)."""
    r = session.patch(f"{VAULT_URL}/secret/{name}", json=SCOPE)
    r.raise_for_status()
    return r.json()["secret"]

def secret_exists(name: str) -> bool:
    """Cheaper than read; doesn't surface the value."""
    r = session.patch(f"{VAULT_URL}/checksecret/{name}", json=SCOPE)
    if r.status_code == 404:
        return False
    r.raise_for_status()
    return r.json().get("secretexists", False)

def write_secret(name: str, value: str, *, indefinite: bool = False) -> None:
    body = dict(SCOPE, secret=value, indefinite=indefinite)
    r = session.post(f"{VAULT_URL}/secret/{name}", json=body)
    r.raise_for_status()

def fetch_jwt_verifier(service: str, realm: str) -> dict:
    """Fetch a peer service's JWT signing cert + public key, for token verification."""
    body = dict(SCOPE, service=service)
    r = session.patch(f"{VAULT_URL}/jwt/internal/{realm}", json=body)
    r.raise_for_status()
    return r.json()  # {"cert": "...PEM...", "publickey": "...PEM..."}

if __name__ == "__main__":
    assert ready(), "vault unreachable"

    db_password = read_secret("db_password")
    print(f"db password length: {len(db_password)}")

    jwt = mint_jwt()
    print(f"service JWT: {jwt[:40]}...")

    janus_keys = fetch_jwt_verifier("janus", "users")
    print(f"janus JWT verifier pubkey:\n{janus_keys['publickey'][:80]}...")

Receiving + validating a JWT (Python)

Once you’ve fetched a peer service’s JWT verifier public key, validate incoming JWTs with PyJWT:

import jwt as pyjwt

# Fetched once at startup; refresh periodically (e.g., every 6h).
verifier = fetch_jwt_verifier("janus", "users")
public_key_pem = verifier["publickey"]

def authenticate_request(authorization_header: str) -> dict:
    """Returns the JWT claims if valid, raises otherwise."""
    if not authorization_header.startswith("Bearer "):
        raise ValueError("missing Bearer token")
    token = authorization_header.removeprefix("Bearer ")
    return pyjwt.decode(
        token,
        public_key_pem,
        algorithms=["RS256"],
        audience="myservice",   # IMPORTANT: validate audience
        issuer="kisai-janus",
    )

Go (from outside the kis.ai monorepo)

If you’re an in-tree kis.ai service, use lib-vault.VaultClient instead. This example is for standalone Go services.

// Package vaultclient is a minimal Vault client for a Go service
// that doesn't depend on kis.ai internal packages.
package vaultclient

import (
  "bytes"
  "crypto/tls"
  "crypto/x509"
  "encoding/json"
  "errors"
  "fmt"
  "io"
  "net/http"
  "os"
  "time"
)

type Client struct {
  url        string
  httpClient *http.Client
  scope      map[string]any
}

func New(url, caPath, certPath, keyPath string, scope map[string]any) (*Client, error) {
  caPEM, err := os.ReadFile(caPath)
  if err != nil {
    return nil, fmt.Errorf("read CA: %w", err)
  }
  caPool := x509.NewCertPool()
  if !caPool.AppendCertsFromPEM(caPEM) {
    return nil, errors.New("no certs found in CA file")
  }

  cert, err := tls.LoadX509KeyPair(certPath, keyPath)
  if err != nil {
    return nil, fmt.Errorf("load client cert: %w", err)
  }

  tr := &http.Transport{
    TLSClientConfig: &tls.Config{
      RootCAs:      caPool,
      Certificates: []tls.Certificate{cert},
      MinVersion:   tls.VersionTLS12,
    },
    IdleConnTimeout: 90 * time.Second,
  }

  return &Client{
    url:        url,
    httpClient: &http.Client{Transport: tr, Timeout: 20 * time.Second},
    scope:      scope,
  }, nil
}

func (c *Client) Ready() error {
  r, err := c.httpClient.Get(c.url + "/ready")
  if err != nil {
    return err
  }
  defer r.Body.Close()
  if r.StatusCode != http.StatusOK {
    return fmt.Errorf("vault not ready: %s", r.Status)
  }
  return nil
}

func (c *Client) MintJWT() (string, error) {
  r, err := c.httpClient.Get(c.url + "/certificate/auth")
  if err != nil {
    return "", err
  }
  defer r.Body.Close()
  if r.StatusCode != http.StatusOK {
    return "", fmt.Errorf("mint jwt: %s", r.Status)
  }
  var resp struct {
    Token string `json:"token"`
  }
  if err := json.NewDecoder(r.Body).Decode(&resp); err != nil {
    return "", err
  }
  return resp.Token, nil
}

func (c *Client) ReadSecret(name string) (string, error) {
  body, _ := json.Marshal(c.scope)
  req, _ := http.NewRequest(http.MethodPatch, c.url+"/secret/"+name, bytes.NewReader(body))
  req.Header.Set("Content-Type", "application/json")
  r, err := c.httpClient.Do(req)
  if err != nil {
    return "", err
  }
  defer r.Body.Close()
  if r.StatusCode != http.StatusOK {
    errBody, _ := io.ReadAll(r.Body)
    return "", fmt.Errorf("read secret %s: %s — %s", name, r.Status, errBody)
  }
  var resp struct {
    Secret string `json:"secret"`
  }
  if err := json.NewDecoder(r.Body).Decode(&resp); err != nil {
    return "", err
  }
  return resp.Secret, nil
}

func (c *Client) WriteSecret(name, value string, indefinite bool) error {
  body := make(map[string]any, len(c.scope)+2)
  for k, v := range c.scope {
    body[k] = v
  }
  body["secret"] = value
  body["indefinite"] = indefinite

  b, _ := json.Marshal(body)
  r, err := c.httpClient.Post(c.url+"/secret/"+name, "application/json", bytes.NewReader(b))
  if err != nil {
    return err
  }
  defer r.Body.Close()
  if r.StatusCode != http.StatusOK {
    errBody, _ := io.ReadAll(r.Body)
    return fmt.Errorf("write secret %s: %s — %s", name, r.Status, errBody)
  }
  return nil
}

Usage:

package main

import (
  "log"
  "os"

  "yourorg.example.com/vaultclient"
)

func main() {
  vc, err := vaultclient.New(
    os.Getenv("VAULT_URL"),
    os.Getenv("VAULT_CA"),
    os.Getenv("VAULT_CERT"),
    os.Getenv("VAULT_KEY"),
    map[string]any{
      "datacenter":  "us-west",
      "cluster":     "prod-cluster-01",
      "customer":    "acme",
      "product":     "forge",
      "environment": "prod",
      "tenant":      "acme-prod",
    },
  )
  if err != nil {
    log.Fatal(err)
  }
  if err := vc.Ready(); err != nil {
    log.Fatal(err)
  }

  pw, err := vc.ReadSecret("db_password")
  if err != nil {
    log.Fatal(err)
  }
  log.Printf("got db password (length %d)", len(pw))
}

Node.js

// Minimal Vault client for a Node.js service.
//
// requires: native https + fs; for production, prefer axios or undici
// with a custom https.Agent for mTLS.

const fs = require('fs');
const https = require('https');
const { URL } = require('url');

const VAULT_URL  = new URL(process.env.VAULT_URL);
const VAULT_CA   = fs.readFileSync(process.env.VAULT_CA);
const VAULT_CERT = fs.readFileSync(process.env.VAULT_CERT);
const VAULT_KEY  = fs.readFileSync(process.env.VAULT_KEY);

const SCOPE = {
  datacenter:  'us-west',
  cluster:     'prod-cluster-01',
  customer:    'acme',
  product:     'forge',
  environment: 'prod',
  tenant:      'acme-prod',
};

const tlsAgent = new https.Agent({
  ca:   VAULT_CA,
  cert: VAULT_CERT,
  key:  VAULT_KEY,
  keepAlive: true,
});

function request(method, path, body) {
  return new Promise((resolve, reject) => {
    const opts = {
      method,
      hostname: VAULT_URL.hostname,
      port:     VAULT_URL.port,
      path:     path,
      agent:    tlsAgent,
      headers:  body ? { 'Content-Type': 'application/json' } : {},
    };
    const req = https.request(opts, (res) => {
      let data = '';
      res.on('data', (chunk) => (data += chunk));
      res.on('end', () => {
        if (res.statusCode >= 200 && res.statusCode < 300) {
          resolve(data ? JSON.parse(data) : {});
        } else {
          reject(new Error(`${method} ${path} ${res.statusCode}: ${data}`));
        }
      });
    });
    req.on('error', reject);
    if (body) req.write(JSON.stringify(body));
    req.end();
  });
}

async function ready()                      { return request('GET',   '/ready'); }
async function mintJWT()                    { return request('GET',   '/certificate/auth'); }
async function readSecret(name)             { return request('PATCH', `/secret/${name}`, SCOPE); }
async function writeSecret(name, value, indefinite = false) {
  return request('POST', `/secret/${name}`, { ...SCOPE, secret: value, indefinite });
}

(async () => {
  await ready();

  const password = (await readSecret('db_password')).secret;
  console.log(`got db password (length ${password.length})`);

  const { token } = await mintJWT();
  console.log(`service JWT: ${token.slice(0, 40)}...`);
})().catch((e) => { console.error(e); process.exit(1); });

Patterns

Refresh the JWT periodically

JWTs from /certificate/auth are short-lived (default 1h). Cache the token and refresh before expiry. The exact lifetime is in the JWT’s exp claim:

import time
import jwt as pyjwt

class JWTCache:
    def __init__(self, mint_fn, refresh_before_seconds=300):
        self.mint_fn = mint_fn
        self.refresh_before = refresh_before_seconds
        self._token = None
        self._exp = 0

    def get(self) -> str:
        now = time.time()
        if self._token and now < self._exp - self.refresh_before:
            return self._token
        self._token = self.mint_fn()
        claims = pyjwt.decode(self._token, options={"verify_signature": False})
        self._exp = claims["exp"]
        return self._token

Read secrets on hot paths — cache or batch

PATCH /secret/:name is a synchronous HTTPS round-trip. Don’t call it per-request on a hot path. Cache at process startup, refresh on config change, or batch reads at the edges.

A reasonable pattern: a RefreshableSecret wrapper that re-reads every N minutes and exposes the latest value via a getter:

import threading
import time

class RefreshableSecret:
    def __init__(self, vault_client, name, refresh_seconds=300):
        self.vc = vault_client
        self.name = name
        self.refresh_seconds = refresh_seconds
        self._value = None
        self._lock = threading.Lock()
        self._stop = threading.Event()
        # Initial read at construction:
        self._value = self.vc.read_secret(name)
        threading.Thread(target=self._refresher, daemon=True).start()

    def _refresher(self):
        while not self._stop.wait(self.refresh_seconds):
            try:
                v = self.vc.read_secret(self.name)
                with self._lock:
                    self._value = v
            except Exception:
                # Stale value remains current; log + alarm in real code.
                pass

    def value(self) -> str:
        with self._lock:
            return self._value

Handle vault.svc outages gracefully

vault.svc should be available, but design as if it might not be at any given moment. Two strategies:

On startup, refuse to serve traffic until the bootstrap reads succeed. Once you have cached values, keep serving with them even if subsequent refresh attempts fail (loud alarms — but don’t take the service down for a vault.svc blip).

2. Reactive refresh. Don’t try to refresh on a request that’s waiting; refresh asynchronously and let the request continue with the cached value.

Validate JWT audience on the receiver side

When you receive a JWT from a peer service, validate the aud claim against your own service identity:

claims = pyjwt.decode(
    token,
    public_key_pem,
    algorithms=["RS256"],
    audience="myservice",          # ← REQUIRED — otherwise any JWT works against you
    issuer="kisai-janus",
)

Skipping audience validation is a known platform-wide hole that the spec’s phase-3 closes. Don’t write new code without it.

Don’t log secrets

Even at debug level. vault.svc’s own audit log is the system of record for “who read what”; your application logs should never duplicate the value. A practical safeguard:

def read_secret(self, name: str) -> str:
    value = self._fetch(name)
    self._log.debug("read secret name=%s length=%d", name, len(value))
    # NOT: self._log.debug("read secret %s = %s", name, value)
    return value

Where to go next

• Need the exact request/response shape for an endpoint not covered above? HTTP API.
• Want to use the shell instead of code? CLI reference.