SDKs

Python SDK

Python 3.8+, no dependencies — everything is hmac, hashlib and json, so the SDK never brings a transitive advisory into your service.

The endpoint

import os
from tether import Tether

tools = {
    "get_order_status": {
        "description": "Look up one order by its number and return its fulfilment state.",
        "domain": "orders",
        "schema": {
            "type": "object",
            "properties": {"order_id": {"type": "string", "description": "The order number."}},
            "required": ["order_id"],
        },
        "handler": lambda args, ctx: db.orders.status(ctx.tenant["scope_id"], args["order_id"]),
    },
    "cancel_order": {
        "description": "Cancel an order that has not yet been dispatched.",
        "domain": "orders",
        "write": True,
        "risk": "elevated",
        "schema": {
            "type": "object",
            "properties": {"order_id": {"type": "string", "description": "The order to cancel."}},
            "required": ["order_id"],
        },
        "handler": lambda args, ctx: db.orders.cancel(args["order_id"]),
    },
}

tether = Tether(secret=os.environ["TETHER_SECRET"], tools=tools)

A tool is a dict carrying handler, or a bare callable (arguments, ctx). Use the first form: build_catalog reads the same dict, so the schema you register cannot drift from the function serving it.

"write": True is the only thing that makes a write a write. There is no inference from the name.

Wiring it in

tether.handle(headers, raw_body) returns a Response with .status, .body and .headers. It takes no framework, so it fits whichever one you already run.

FastAPI

@app.post("/tether")
async def tool_endpoint(request: Request):
    result = tether.handle(request.headers, await request.body())
    return JSONResponse(result.body, status_code=result.status, headers=result.headers)

Flask

@app.post("/tether")
def tool_endpoint():
    result = tether.handle(request.headers, request.get_data())
    return result.body, result.status, result.headers

Django

@csrf_exempt
def tool_endpoint(request):
    result = tether.handle(request.headers, request.body)
    response = JsonResponse(result.body, status=result.status)
    for name, value in result.headers.items():
        response[name] = value
    return response

Standalone — tether.wsgi_app() is a complete WSGI application for gunicorn or wsgiref.

The context argument

def handler(args, ctx):
    ctx.tenant["organization_id"]  # who this turn belongs to
    ctx.tenant["scope_id"]         # the site, ledger or region, when the tenant has several
    ctx.tenant["user_id"]
    ctx.tool                       # the tool being called
    ctx.is_write                   # as you declared it
    ctx.mode                       # "execute" on a confirmed write, None otherwise
    ctx.request_id                 # also the idempotency key
    ctx.customer                   # {"id": ...} of the verified end customer, or None

Scope every query by ctx.tenant, never by anything in args. The tenant came from the credential and cannot be influenced by the model; an argument is whatever the model decided to send.

Enforce ownership with the verified customer. ctx.customer is the end customer Tether verified from the token your server signed (Verified customers), or None for an anonymous turn. Refuse a booking, order or account it does not own, whatever the arguments say: a customer_id in the arguments is something the model wrote. Declare "requiresCustomer": True on a tool that acts for one customer and Tether neither calls nor stages it without one. Your host mints the token with mint_customer_token(key, tenant, customer_id, ttl_seconds=300).

Registering the catalog

import json, os, urllib.request
from tether import build_catalog

request = urllib.request.Request(
    os.environ["TETHER_URL"] + "/v1/tools",
    data=json.dumps(build_catalog(tools)).encode(),
    headers={"X-Api-Key": os.environ["TETHER_SECRET_KEY"], "Content-Type": "application/json"},
    method="PUT",
)
print(json.load(urllib.request.urlopen(request)))

build_catalog raises CatalogError with the engine's own message when a definition is incomplete, so a broken catalog fails before it reaches the network.

Lifecycle events

def on_event(event, ctx):
    if event["event"] == "action.pending":
        notify_approvers(event["action_id"])
    elif event["event"] == "conversation.handoff":
        open_ticket(event["session_id"], ctx.customer)

tether = Tether(secret=..., tools=tools, on_event=on_event)

Omit on_event and the event is acknowledged and dropped. EVENT_TYPES lists every event, including action.verified and action.unverified for a write's verified outcome, action.precondition_failed for a precondition that no longer held at confirmation, action.provider_refused for a confirmed write an imported API refused, and conversation.handoff for a conversation handed to a person.

A store that survives more than one process

MemoryStore is per-process. Behind a load balancer or across a deploy, a replayed request reaches a process that never saw the original and runs your function again — for a write, twice.

The store contract is three operations, not two, because a check followed by a write cannot stop two overlapping requests from both running the effect. A store with only get and set is refused when you construct Tether.

claim(key) must be atomic against every caller that could share the key. It returns a Claim whose state is Claim.FRESH (you own the key and must run the effect), Claim.IN_FLIGHT (another request is running it right now) or Claim.DONE (it already ran; the claim carries the result and your function is not called).
complete(key, value) records the result and releases the claim.
release(key) gives up a claim whose effect did not happen, so a retry may proceed.

In Redis the claim is one SET ... NX:

import json
from tether import Claim


class RedisStore:
    CLAIMED = "claimed"

    def __init__(self, client, lease=300, retention=86400):
        self._client, self._lease, self._retention = client, lease, retention

    def _key(self, key):
        return f"tether:{key}"

    def _is_claim(self, held):
        return held in (self.CLAIMED, self.CLAIMED.encode())

    def claim(self, key):
        if self._client.set(self._key(key), self.CLAIMED, nx=True, ex=self._lease):
            return Claim(Claim.FRESH)

        held = self._client.get(self._key(key))
        if held is None or self._is_claim(held):
            return Claim(Claim.IN_FLIGHT)
        return Claim(Claim.DONE, json.loads(held))

    def complete(self, key, value):
        self._client.set(self._key(key), json.dumps(value), ex=self._retention)

    def release(self, key):
        if self._is_claim(self._client.get(self._key(key))):
            self._client.delete(self._key(key))


tether = Tether(secret=..., tools=tools, store=RedisStore(redis_client))

The claim lease is how long a crashed process can hold a key before another request may take it, so keep it longer than your slowest write.

What it returns

Situation	Status	`error`
Signature does not match	`401`	`invalid_signature`
Timestamp outside tolerance	`401`	`stale_timestamp`
Write without execute mode	`400`	`execute_mode_required`
No handler for that tool	`400`	`unknown_tool`
Body is not JSON	`400`	`invalid_body`
Your handler raised	`500`	`tool_failed`, with the message
Replayed `request_id`	`200`	the first result, `X-Tether-Replayed: true`

A failed call is not cached. The retry reaches your handler.

SDKs — what all three share, and where they differ
Retries, timeouts, idempotency
Choosing risk levels — what to put in risk
Testing with mock mode — exercise it with no provider key