SDKs

LatchGate provides client SDKs for Python and TypeScript. Both handle DPoP key generation, lease management, and proof construction automatically.

Python

Requires: Python 3.10+

pip install latchgate

Basic usage

import asyncio
from latchgate import LatchGateClient

async def main():
    async with LatchGateClient(agent_id="my-agent") as client:
        result = await client.execute("http_fetch", {
            "url": "https://httpbin.org/get",
        })
        print(result.output)
        print(result.receipt_id)

        receipt = await client.get_receipt(result.receipt_id)
        print(receipt.is_fully_successful)

asyncio.run(main())

Lazy-connect

When agent_id is set at construction time, the SDK automatically connects on the first execute() call — no explicit connect() needed. The SDK generates a DPoP key pair, obtains a sender-bound lease, and renews it transparently when it nears expiry.

# Lazy-connect: just set agent_id and start calling execute().
async with LatchGateClient(agent_id="my-agent") as client:
    result = await client.execute("http_fetch", {"url": "https://example.com"})

Without agent_id, you must call connect() explicitly:

async with LatchGateClient(base_url="http://localhost:3000") as client:
    await client.connect(agent_id="my-agent")
    result = await client.execute("http_fetch", {"url": "https://example.com"})

LATCHGATE_URL environment variable

Both SDKs resolve the gate URL in this order:

1. base_url constructor argument (TCP)
2. LATCHGATE_URL environment variable (TCP)
3. socket constructor argument (UDS, default: /run/latchgate/gate.sock)

This means you can set LATCHGATE_URL=http://localhost:3000 once and omit base_url from all client constructors:

# With LATCHGATE_URL set in the environment:
async with LatchGateClient(agent_id="my-agent") as client:
    result = await client.execute("http_fetch", {"url": "https://example.com"})

Approval flow

When policy requires human approval, execute() raises LatchGateApprovalRequired:

from latchgate import LatchGateApprovalRequired, LatchGateClient

async def main():
    async with LatchGateClient(agent_id="my-agent") as client:
        try:
            result = await client.execute("http_post", {
                "to": "user@example.com",
                "subject": "Weekly report",
                "body": "Please find the report attached.",
            })
        except LatchGateApprovalRequired as exc:
            print(f"approval required: {exc.approval_id}")
            status = await client.get_approval_status(exc.approval_id)

asyncio.run(main())

UDS transport (production)

LatchGateClient(socket="/run/latchgate/gate.sock")

Egress profile

Retrieve the egress profile for an action — useful for orchestrator UIs displaying which external domains an action may contact:

egress = await client.get_action_egress("http_fetch")
print(egress.profile)          # "none" or "proxy_allowlist"
print(egress.allowed_domains)  # ["httpbin.org", "api.github.com", ...]
print(egress.has_egress)       # True if profile != "none"

Error handling

The SDK raises typed exceptions for each failure class. Match on the specific type to decide whether to retry, reconnect, or surface the error:

from latchgate import (
    LatchGateDenied,           # action denied by policy — do not retry as-is
    LatchGateApprovalRequired, # needs human approval — poll approval_id
    LatchGateBudgetExhausted,  # lease budget used up — reconnect for fresh budget
    LatchGateAuthError,        # expired/invalid lease — reconnect
    LatchGateLeaseExpired,     # lease TTL elapsed — reconnect (subclass of AuthError)
    LatchGateReplayDetected,   # DPoP jti reused — possible replay; do NOT retry the proof
    LatchGateUnavailable,      # OPA/Redis down — retry with backoff
    LatchGateTransportError,   # socket error — retry
    LatchGateNotConnected,     # connect() not called and no agent_id set
)

try:
    result = await client.execute("http_fetch", {"url": "https://example.com"})
except LatchGateApprovalRequired as exc:
    print(f"approval required: {exc.approval_id}")
except LatchGateDenied as exc:
    print(f"denied: {exc.action_id} — {exc.reason}")
except LatchGateBudgetExhausted:
    await client.connect()  # fresh lease with new budget
except LatchGateReplayDetected:
    raise  # never retry with the same proof — surface to security
except LatchGateLeaseExpired:
    await client.connect()  # lease TTL elapsed — obtain a new one
except LatchGateAuthError:
    await client.connect()  # other auth failure — reconnect
except LatchGateUnavailable:
    pass  # transient — retry with backoff
except LatchGateTransportError:
    pass  # socket unreachable or connection reset — retry with backoff
except LatchGateNotConnected:
    pass  # programming error — call connect() first or set agent_id

LatchGateLeaseExpired and LatchGateReplayDetected both subclass LatchGateAuthError, so a single except LatchGateAuthError catches them when finer handling is not needed. Order matters: list the specific subclasses before the base so they are not shadowed.

All exceptions inherit from LatchGateError.

TypeScript

Requires: Node 18+

npm install latchgate

Basic usage

import { LatchGateClient } from "latchgate";

const client = new LatchGateClient({ agentId: "my-agent" });
const result = await client.execute("http_fetch", {
  url: "https://httpbin.org/get",
});

console.log(result.output);
console.log(result.receiptId);

Lazy-connect

Same behavior as the Python SDK. When agentId is set in the constructor, execute() auto-connects on the first call:

// No explicit connect() needed:
const client = new LatchGateClient({ agentId: "my-agent" });
const result = await client.execute("http_fetch", {
  url: "https://example.com",
});

LATCHGATE_URL environment variable

The TypeScript SDK reads process.env["LATCHGATE_URL"] with the same fallback order as Python: explicit baseUrl => LATCHGATE_URL env var => UDS socket.

Approval flow

import { LatchGateClient, LatchGateApprovalRequired } from "latchgate";

try {
  const result = await client.execute("http_post", {
    to: "user@example.com",
    subject: "Report",
    body: "See attached.",
  });
} catch (err) {
  if (err instanceof LatchGateApprovalRequired) {
    console.log(`approval required: ${err.approvalId}`);
  }
}

Egress profile

const egress = await client.getActionEgress("http_fetch");
console.log(egress.profile); // "none" or "proxy_allowlist"
console.log(egress.allowedDomains); // ["httpbin.org", "api.github.com", ...]
console.log(egress.hasEgress); // true if profile !== "none"

Error handling

import {
  LatchGateDenied, // action denied by policy
  LatchGateApprovalRequired, // needs human approval — poll approvalId
  LatchGateBudgetExhausted, // lease budget used up — reconnect
  LatchGateAuthError, // expired/invalid lease — reconnect
  LatchGateLeaseExpired, // lease TTL elapsed — reconnect (subclass of AuthError)
  LatchGateReplayDetected, // DPoP jti reused — possible replay; do NOT retry the proof
  LatchGateUnavailable, // OPA/Redis down — retry with backoff
  LatchGateTransportError, // socket unreachable or connection reset — retry
  LatchGateNotConnected, // connect() not called and no agentId set
} from "latchgate";

try {
  const result = await client.execute("http_fetch", {
    url: "https://example.com",
  });
} catch (err) {
  if (err instanceof LatchGateApprovalRequired) {
    console.log(`approval required: ${err.approvalId}`);
  } else if (err instanceof LatchGateDenied) {
    console.error(err.actionId, err.reason);
  } else if (err instanceof LatchGateBudgetExhausted) {
    await client.connect(); // fresh lease
  } else if (err instanceof LatchGateReplayDetected) {
    throw err; // never retry with the same proof — surface to security
  } else if (err instanceof LatchGateLeaseExpired) {
    await client.connect(); // lease TTL elapsed — obtain a new one
  } else if (err instanceof LatchGateAuthError) {
    await client.connect(); // other auth failure — reconnect
  } else if (err instanceof LatchGateUnavailable) {
    // transient — retry with backoff
  } else if (err instanceof LatchGateTransportError) {
    // socket unreachable or connection reset — retry with backoff
  }
}

LatchGateLeaseExpired and LatchGateReplayDetected both extend LatchGateAuthError, so a single instanceof LatchGateAuthError branch catches them when finer handling is not needed. Check the specific subclasses before the base so they are not shadowed.

All exceptions inherit from LatchGateError.

What the SDKs handle

• Ephemeral DPoP key pair generation (P-256)
• Lease acquisition, auto-renewal, and lazy-connect
• DPoP proof construction for every request (method + URI + body binding)
• Request signing and canonical hashing
• Approval status polling
• Receipt retrieval
• Egress profile retrieval per action

What the SDKs do not handle

• Operator approval/deny (use the CLI or Approval API directly)
• Policy configuration
• Action manifest authoring

Examples

Runnable examples in each SDK:

Example	Python	TypeScript	What it shows
Hello	sdk/python/examples/hello.py	sdk/typescript/examples/hello.ts	Lease => execute => receipt
Approval flow	sdk/python/examples/approval_flow.py	sdk/typescript/examples/approval_flow.ts	Human-in-the-loop approval handling
Smoke test	sdk/python/examples/smoke_test.py	sdk/typescript/examples/smoke_test.ts	CI-oriented pass/fail (make test-smoke)

# Python
cd sdk/python && uv run examples/hello.py

# TypeScript
cd sdk/typescript && npx tsx examples/hello.ts

Running SDK tests

make test-sdk              # both SDKs
make test-sdk-python       # Python only
make test-sdk-typescript   # TypeScript only

Cross-language golden vectors

Both SDKs validate against shared JCS (RFC 8785) test vectors in definitions/test_vectors/jcs/golden.json. These ensure that canonical hashing produces identical SHA-256 digests across Rust, Python, and TypeScript. If an SDK’s JSON serialization or hashing diverges, the golden vector tests fail.