Python SDK

The Klock Python SDK provides native access to the coordination kernel. It is built on top of PyO3, so your Python code calls compiled Rust directly.

Installation

pip install klock

Basic Usage

The core primitive is the KlockClient, which embeds a local coordination engine and manages leases.

from klock import KlockClient

# Initialize the client with an in-memory kernel
klock = KlockClient()

# 1. Register an agent with a priority (older/higher priority first)
klock.register_agent("refactor-bot-a", 100)
klock.register_agent("refactor-bot-b", 200)

# 2. Acquire a lease for an intent
# Signature: acquire_lease(agent_id, session_id, resource_type, resource_path, predicate, ttl_ms)
result = klock.acquire_lease("refactor-bot-a", "session_1", "FILE", "/src/auth.ts", "MUTATES", 60000)

if result['success']:
    print(f"Lease acquired: {result['lease_id']}")
    
    # 3. DO YOUR WORK HERE
    
    # 4. Release when finished
    klock.release_lease(result['lease_id'])
else:
    print(f"Blocked. Reason: {result['reason']} (Wait Time: {result['wait_time']}ms)")

Supported Predicates (KLIS Protocol)

When you acquire a lease, you specify the predicate based on the KLIS matrix:

• "PROVIDES" — Agent creates or introduces a resource.
• "CONSUMES" — Agent reads or depends on a resource.
• "MUTATES" — Agent modifies an existing resource.
• "DELETES" — Agent removes a resource.
• "DEPENDS_ON" — Agent declares dependency on a resource.
• "RENAMES" — Agent changes resource identity or location.

Wait-Die Automation

You rarely deal with acquire_lease failures manually unless you are building a custom integration.

We highly recommend using the pre-built integrations, like @klock_protected for LangChain, which handles the Wait-Die retry loops and lease release automatically.

Centralized Server

If you want a shared coordination service, run klock-cli and call its HTTP API. The Python SDK currently exposes the embedded client surface only.