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 — see the platform integration guide. 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:

1. Fail closed during cold start, fail open during steady state. 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.
• Building a service inside the kis.ai monorepo? You should use lib-vault.VaultClient instead of writing your own HTTP client — see the platform integration guide.