Python API Overview

watercooler/
├── __init__.py          # Public exports (AdvisoryLock, read, write, thread_path)
├── commands.py          # Management commands (list_threads, search, reindex)
├── commands_graph.py    # Graph-canonical thread commands (say, ack, handoff)
├── models.py            # Model registry and resolution
├── config_schema.py     # Configuration classes
└── fs.py                # File system operations

pip install watercooler

from watercooler import (
    AdvisoryLock,      # File locking for thread safety
    read,              # Read thread files
    write,             # Write thread files
    thread_path,       # Get thread file paths
    __version__,       # Package version
)

from watercooler.commands_graph import (
    init_thread,    # Initialize new thread
    say,            # Add entry with auto-ball-flip
    ack,            # Acknowledge without ball flip
    handoff,        # Explicit handoff to counterpart
    set_status,     # Update thread status
    set_ball,       # Update ball owner
)

from watercooler.commands import (
    list_threads,   # List all threads
    list_entries,   # List entries for a thread
    search,         # Search thread content
    reindex,        # Generate markdown index
    web_export,     # Export HTML index
    unlock,         # Clear advisory locks
)

from watercooler.models import (
    # Embedding models
    resolve_embedding_model,
    ensure_model_available,
    get_model_dimension,
    # LLM models
    resolve_llm_gguf_model,
    ensure_llm_model_available,
    get_response_field,
)

from watercooler.config_schema import (
    WatercoolerConfig,
    McpConfig,
    GraphConfig,
    MemoryConfig,
)

from pathlib import Path
from watercooler.commands_graph import init_thread, say

# Initialize a new thread
threads_dir = Path("~/threads").expanduser()
topic = "implement-user-auth"

thread_path = init_thread(
    topic,
    threads_dir=threads_dir,
    title="Implement User Authentication",
    status="OPEN",
    ball="codex"
)

# Add an entry
say(
    topic,
    threads_dir=threads_dir,
    agent="Claude Code",
    role="implementer",
    title="Starting OAuth implementation",
    body="Planning to use OAuth2 with JWT tokens...",
    entry_id="entry-001"
)

from watercooler import AdvisoryLock
from pathlib import Path

lock_path = Path(".locks/my-thread.lock")

with AdvisoryLock(lock_path, timeout=2, ttl=10):
    # Perform thread operations
    pass