Custom Agent

Gate tool calls from any Python or TypeScript agent through LatchGate using the SDK’s execute() method.

Prerequisites

• LatchGate running (latchgate up or production deployment)

Install

pip install latchgate

Direct SDK usage

Call client.execute() for each protected action:

import asyncio
from latchgate import LatchGateClient, LatchGateApprovalRequired

async def agent_loop():
    async with LatchGateClient(agent_id="my-agent") as client:
        # Every call goes through auth => policy => WASM sandbox => receipt.
        result = await client.execute("http_fetch", {
            "url": "https://api.example.com/data",
        })
        print(result.output)
        print(f"receipt: {result.receipt_id}")

        # High-risk actions trigger the approval flow automatically.
        try:
            await client.execute("github_create_issue", {
                "owner": "myorg",
                "repo": "myrepo",
                "title": "Automated issue",
                "body": "Created by agent with LatchGate audit trail.",
            })
        except LatchGateApprovalRequired as exc:
            print(f"Waiting for operator approval: {exc.approval_id}")

asyncio.run(agent_loop())

This works with any orchestration code — a simple loop, a state machine, a LangGraph pipeline, or a framework LatchGate doesn’t have a dedicated integration for.

Handling approvals

When policy requires human approval, execute() raises LatchGateApprovalRequired. You can poll for the approval status:

from latchgate import LatchGateApprovalRequired

try:
    result = await client.execute("send_message", {
        "to": "ceo@example.com",
        "subject": "Contract signed",
        "body": "The deal is closed.",
    })
except LatchGateApprovalRequired as exc:
    print(f"Approval required: {exc.approval_id}")
    # Notify operator, then poll:
    while True:
        status = await client.get_approval_status(exc.approval_id)
        if not status.is_pending:
            break
        await asyncio.sleep(5)
    print(f"Approval resolved: {status.decision}")

Meanwhile, an operator approves or denies from the CLI:

latchgate approvals list
latchgate approvals approve <approval_id>

Verifying receipts

Every action produces a signed receipt. Retrieve it for audit:

result = await client.execute("http_fetch", {"url": "https://example.com"})
receipt = await client.get_receipt(result.receipt_id)
print(f"Verified: {receipt.is_fully_successful}")

UDS transport (production)

client = LatchGateClient(
    socket="/run/latchgate/gate.sock",
    agent_id="my-agent",
)

LATCHGATE_URL environment variable

Set LATCHGATE_URL in the environment and omit base_url:

# Reads LATCHGATE_URL automatically:
client = LatchGateClient(agent_id="my-agent")

TypeScript

The TypeScript SDK provides the same execute() approach:

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.receiptId);

See the SDK documentation for the full TypeScript API.