Skip to main content
The Commands API provides functions for creating, updating, and managing Watercooler threads.

​
Graph-Canonical Commands

These are the primary implementations that write to the graph first, then project to markdown.

​
init_thread

Initialize a new thread with graph-canonical approach. 
def init_thread(
    topic: str,
    *,
    threads_dir: Path,
    title: Optional[str] = None,
    status: str = "OPEN",
    ball: str = "codex",
) -> Path
Parameters:
  • topic (str): Thread topic identifier (e.g., “implement-auth”)
  • threads_dir (Path): Directory containing threads
  • title (Optional[str]): Title override (defaults to formatted topic)
  • status (str): Initial status (default: “OPEN”)
  • ball (str): Initial ball owner (default: “codex”)
Returns: Path to the created thread file Example: 
from pathlib import Path
from watercooler.commands_graph import init_thread

threads_dir = Path("~/threads").expanduser()
thread_path = init_thread(
    "fix-login-bug",
    threads_dir=threads_dir,
    title="Fix Login Bug",
    status="OPEN",
    ball="claude"
)

​
say

Add a team note with automatic ball flip to counterpart. 
def say(
    topic: str,
    *,
    threads_dir: Path,
    agent: str | None = None,
    role: str | None = None,
    title: str,
    entry_type: str = "Note",
    body: str,
    status: str | None = None,
    ball: str | None = None,
    registry: dict | None = None,
    user_tag: str | None = None,
    entry_id: str | None = None,
    code_branch: Optional[str] = None,
) -> Path
Parameters:
  • topic (str): Thread topic identifier
  • threads_dir (Path): Directory containing threads
  • agent (str | None): Agent name (defaults to “Team”)
  • role (str | None): Agent role (defaults from registry)
  • title (str): Entry title (required)
  • entry_type (str): Entry type (default: “Note”)
  • body (str): Entry body text
  • status (str | None): Optional status update
  • ball (str | None): Optional ball update (auto-flips if not provided)
  • registry (dict | None): Optional agent registry
  • user_tag (str | None): Optional user tag
  • entry_id (str | None): Entry ID (required for graph-canonical)
  • code_branch (Optional[str]): Code branch reference
Returns: Path to updated thread file Example: 
from watercooler.commands_graph import say

say(
    "implement-oauth",
    threads_dir=threads_dir,
    agent="Claude Code",
    role="implementer",
    title="Completed OAuth setup",
    body="Implemented OAuth2 flow with JWT tokens.",
    entry_id="entry-123"
)

​
ack

Acknowledge an entry without flipping the ball. 
def ack(
    topic: str,
    *,
    threads_dir: Path,
    agent: str | None = None,
    role: str | None = None,
    title: str | None = None,
    entry_type: str = "Note",
    body: str | None = None,
    status: str | None = None,
    ball: str | None = None,
    registry: dict | None = None,
    user_tag: str | None = None,
    entry_id: str | None = None,
    code_branch: Optional[str] = None,
) -> Path
Parameters:
  • topic (str): Thread topic identifier
  • threads_dir (Path): Directory containing threads
  • agent (str | None): Agent name (defaults to “Team”)
  • role (str | None): Agent role (defaults from registry)
  • title (str | None): Entry title (defaults to “Ack”)
  • entry_type (str): Entry type (default: “Note”)
  • body (str | None): Entry body text (defaults to “ack”)
  • status (str | None): Optional status update
  • ball (str | None): Optional ball update (preserves current ball if not specified)
  • registry (dict | None): Optional agent registry
  • user_tag (str | None): Optional user tag
  • entry_id (str | None): Entry ID (required for graph-canonical)
  • code_branch (Optional[str]): Code branch reference
Returns: Path to updated thread file Example: 
from watercooler.commands_graph import ack

ack(
    "implement-oauth",
    threads_dir=threads_dir,
    agent="Codex",
    title="Reviewed",
    body="Looks good, proceeding with integration",
    entry_id="entry-124"
)

​
handoff

Explicitly flip the ball to the counterpart agent. 
def handoff(
    topic: str,
    *,
    threads_dir: Path,
    agent: str | None = None,
    role: str = "pm",
    note: str | None = None,
    registry: dict | None = None,
    user_tag: str | None = None,
    entry_id: str | None = None,
    code_branch: Optional[str] = None,
) -> Path
Parameters:
  • topic (str): Thread topic identifier
  • threads_dir (Path): Directory containing threads
  • agent (str | None): Agent performing handoff (defaults to “Team”)
  • role (str): Agent role (default: “pm”)
  • note (str | None): Optional custom handoff message
  • registry (dict | None): Optional agent registry
  • user_tag (str | None): Optional user tag
  • entry_id (str | None): Entry ID (required for graph-canonical)
  • code_branch (Optional[str]): Code branch reference
Returns: Path to updated thread file Example: 
from watercooler.commands_graph import handoff

handoff(
    "implement-oauth",
    threads_dir=threads_dir,
    agent="Claude Code",
    note="Ready for review and testing",
    entry_id="entry-125"
)

​
set_status

Update thread status. 
def set_status(
    topic: str,
    *,
    threads_dir: Path,
    status: str,
) -> Path
Parameters:
  • topic (str): Thread topic identifier
  • threads_dir (Path): Directory containing threads
  • status (str): New status value (e.g., “OPEN”, “CLOSED”, “BLOCKED”)
Returns: Path to updated thread file Example: 
from watercooler.commands_graph import set_status

set_status(
    "implement-oauth",
    threads_dir=threads_dir,
    status="CLOSED"
)

​
set_ball

Update ball owner. 
def set_ball(
    topic: str,
    *,
    threads_dir: Path,
    ball: str,
) -> Path
Parameters:
  • topic (str): Thread topic identifier
  • threads_dir (Path): Directory containing threads
  • ball (str): New ball owner
Returns: Path to updated thread file Example: 
from watercooler.commands_graph import set_ball

set_ball(
    "implement-oauth",
    threads_dir=threads_dir,
    ball="codex"
)

​
Management Commands

These commands provide read-only access and utility operations.

​
list_threads

List all threads from the graph. 
def list_threads(
    *,
    threads_dir: Path,
    open_only: bool | None = None
) -> list[tuple[str, str, str, str, Path, bool]]
Parameters:
  • threads_dir (Path): Directory containing threads
  • open_only (bool | None): Filter to open threads only
Returns: List of tuples: (title, status, ball, updated_iso, path, is_new) Example: 
from watercooler.commands import list_threads

threads = list_threads(threads_dir=threads_dir, open_only=True)
for title, status, ball, updated, path, is_new in threads:
    print(f"{title}: {status} (ball: {ball})")

​
list_entries

List parsed entries for a thread topic. 
def list_entries(
    topic: str,
    threads_dir: Path
) -> list[dict[str, str]]
Parameters:
  • topic (str): Thread topic identifier
  • threads_dir (Path): Directory containing threads
Returns: List of dicts with keys: entry_id, title, body, timestamp Example: 
from watercooler.commands import list_entries

entries = list_entries("implement-oauth", threads_dir)
for entry in entries:
    print(f"{entry['timestamp']}: {entry['title']}")
Case-insensitive search across graph entries. 
def search(
    *,
    threads_dir: Path,
    query: str
) -> list[tuple[Path, int, str]]
Parameters:
  • threads_dir (Path): Directory containing threads
  • query (str): Search query
Returns: List of tuples: (path, line_no, line) Example: 
from watercooler.commands import search

results = search(threads_dir=threads_dir, query="authentication")
for path, line_no, line in results:
    print(f"{path.name}:{line_no} - {line}")

​
reindex

Generate a markdown index of threads. 
def reindex(
    *,
    threads_dir: Path,
    out_file: Path | None = None,
    open_only: bool | None = True
) -> Path
Parameters:
  • threads_dir (Path): Directory containing threads
  • out_file (Path | None): Output file path (defaults to threads_dir/index.md)
  • open_only (bool | None): Include only open threads
Returns: Path to generated index file Example: 
from watercooler.commands import reindex

index_path = reindex(threads_dir=threads_dir, open_only=True)
print(f"Index generated at: {index_path}")

​
web_export

Export threads as static HTML. 
def web_export(
    *,
    threads_dir: Path,
    out_file: Path | None = None,
    open_only: bool | None = True
) -> Path
Parameters:
  • threads_dir (Path): Directory containing threads
  • out_file (Path | None): Output file path (defaults to threads_dir/index.html)
  • open_only (bool | None): Include only open threads
Returns: Path to generated HTML file Example: 
from watercooler.commands import web_export

html_path = web_export(threads_dir=threads_dir)
print(f"HTML index generated at: {html_path}")

​
unlock

Clear advisory lock for a topic (debugging tool). 
def unlock(
    topic: str,
    *,
    threads_dir: Path,
    force: bool = False
) -> None
Parameters:
  • topic (str): Thread topic identifier
  • threads_dir (Path): Directory containing threads
  • force (bool): Remove lock even if it appears active
Example: 
from watercooler.commands import unlock

unlock("stuck-thread", threads_dir=threads_dir, force=True)

​
Type Signatures

All commands use type hints for better IDE support: 
from pathlib import Path
from typing import Optional

# Return types are explicit
thread_path: Path = init_thread(...)
entries: list[dict[str, str]] = list_entries(...)
results: list[tuple[Path, int, str]] = search(...)

Build docs developers (and LLMs) love

Get started for freeTalk to us