Python SDK

Version: 0.20.0

Python SDK for cq — the shared agent knowledge commons.

Lets any Python application query, propose, confirm, and flag knowledge units against a remote cq API, or store locally when no remote is configured.

Installation

uv add cq-sdk

Or with pip:

pip install cq-sdk

Quick Start

from cq import Client, FlagReason

cq = Client()  # Auto-discovers config; falls back to local-only.

# Query.
results = cq.query(domains=["api", "stripe"], language="python")

# Propose.
ku = cq.propose(
    summary="Stripe 402 means card_declined",
    detail="Check error.code, not error.type.",
    action="Handle card_declined explicitly.",
    domains=["api", "stripe"],
)

# Confirm / flag.
cq.confirm(ku.id)
cq.flag(ku.id, reason=FlagReason.STALE)

# Get the canonical agent prompts.
from cq import prompts

skill_prompt = prompts.skill()
reflect_prompt = prompts.reflect()

Configuration

The client reads configuration from environment variables:

Variable	Description	Default
CQ_ADDR	Remote cq API address	None (local-only)
CQ_API_KEY	API key for the remote API	None
CQ_LOCAL_DATABASE_URL	Local store connection URL (e.g. sqlite:///abs/path/local.db)	None (falls back to CQ_LOCAL_DB_PATH)
CQ_LOCAL_DB_PATH	Local SQLite file path	$XDG_DATA_HOME/cq/local.db
XDG_DATA_HOME	Base directory for the default database path (XDG spec)	~/.local/share

Or pass directly:

cq = Client(
    addr="http://localhost:3000",
    local_db_path=Path("~/.local/share/cq/local.db").expanduser(),
)

Store protocol

The local store is pluggable. The SDK defines a Store runtime-checkable Protocol that the Client depends on; the default SqliteStore satisfies it, and you can supply any implementation.

Selecting a store

The client resolves the local store in this precedence order:

1. store= argument — inject any cq.Store directly.
2. CQ_LOCAL_DATABASE_URL — a connection-string URL resolved by cq.create_store. Accepted schemes: sqlite:///abs/path, sqlite:path, and postgresql:// (requires the cq-sdk[postgres] extra).
3. local_db_path= argument / CQ_LOCAL_DB_PATH env var — path to a SQLite file.
4. XDG default — $XDG_DATA_HOME/cq/local.db (typically ~/.local/share/cq/local.db).

Interface

The cq.Store Protocol requires eight methods. Implementations must be safe for use across asyncio.to_thread executor threads.

Method	Signature	Semantics
get	(unit_id: str) -> KnowledgeUnit | None	Retrieve by ID, or None if absent.
all	() -> list[KnowledgeUnit]	Return every unit in the store.
insert	(unit: KnowledgeUnit) -> None	Insert a unit. Raise DuplicateUnitError on an existing ID; raise ValueError if domains are empty after normalization.
update	(unit: KnowledgeUnit) -> None	Replace an existing unit. Raise KeyError when the ID is absent; raise ValueError if domains are empty.
delete	(unit_id: str) -> None	Remove by ID. Raise KeyError when absent.
query	(params: QueryParams) -> StoreQueryResult	Return units matching the query, ranked most-relevant first.
stats	(*, recent_limit: int = 5) -> StoreStats	Return aggregated store statistics.
close	() -> None	Release resources. Must be safe to call more than once.

Built-in implementations

• SqliteStore — the default. Opens a SQLite file with FTS5 full-text search, WAL journaling, and domain-tag indexing.
• PostgresStore — requires cq-sdk[postgres]. Connects to a shared PostgreSQL instance for multi-agent knowledge sharing. Domain-tag matching only (no full-text). Install with uv add cq-sdk[postgres] or pip install cq-sdk[postgres].
• InMemoryStore — map-backed, no persistence. Useful for tests and as a worked example for custom stores (domain-tag matching only, no full-text).

from cq import Client, InMemoryStore

client = Client(store=InMemoryStore())

PostgreSQL via CQ_LOCAL_DATABASE_URL:

import os
os.environ["CQ_LOCAL_DATABASE_URL"] = "postgres://user:pass@localhost:5432/cq"

client = Client()  # Resolves to PostgresStore automatically.

Or inject directly:

from cq import Client
from cq.stores.postgres import PostgresStore

store = PostgresStore("postgres://user:pass@localhost:5432/cq")
client = Client(store=store)

Bring your own

Implement the cq.Store protocol and inject it via store=. Reuse the shared ranker cq.rank_candidates from your query implementation so ranking stays consistent across backends. Verify the implementation against the conformance suite in tests/conformance.py:

from tests.conformance import run_store_conformance

run_store_conformance(lambda: MyCustomStore())

Knowledge tiers

Every knowledge unit has a tier: local (on-disk SQLite, never leaves the machine), private (stored on the remote API at CQ_ADDR, visible to every client pointing at the same remote), or public (open commons; not yet available).

With a remote configured, cq.propose(...) sends the unit to the remote and returns it tagged private; with no remote, or if the remote is unreachable, it writes the unit locally as local.

See the top-level README for the full description.

Dev Setup

uv sync --group dev

Testing

make test

Linting

make lint

License

Apache License 2.0