wenchang/reference.md View as Markdown (for AI agents; also llms.txt)

Tool and error reference#

Version 0.1.1. Everything here is taken from the source: the tool layer (wenchang.tools, wenchang.schemas, wenchang.mcp), the core store (wenchang.core), and storage (wenchang.storage). Paths, scopes, metadata, versions, and limits are explained in Concepts. How to wire the tools in is under Integrate, and how to run storage is in Storage and operations.

Tools#

Seven tools. With tool_prefix set, each is registered as {tool_prefix}_{name}. All results are JSON objects; all failures are rendered as described under errors. In every tool, scope, area, and name are strings, and expected_version is a string (the version from a prior result).

Common file object, returned by read_file, write_file, append_line, and replace_fact:

Tool schemas#

wenchang.schemas (in the base package) describes each tool for frameworks other than MCP:

• tool_schemas(product=None) -> Mapping[str, ToolSchema]: one read-only entry per tool, in wenchang.tools.TOOL_NAMES order.
• ToolSchema: name, description, input, output, and error, the last three JSON Schema draft 2020-12 dicts.
• error_schema() -> dict: the schema shared by every render_error payload.

wenchang.tools.tool_descriptions(product=None) -> Mapping[str, str] returns each tool's description by name, in TOOL_NAMES order, without binding a session. With product, each first line names it. It is the same text as ToolSchema.description and MemoryTools.descriptions().

Tool schemas explains each field and how to register the tools with them.

The memory prompt resource#

The server exposes the assembled memory prompt as an MCP resource:

Clients that drop server instructions can read the resource instead. Over MCP, a successful get_memory_index call also returns the same prompt as a second text content block.

wenchang.mcp.mcp_tool_descriptions(slots=None, product=None) -> Mapping[str, str] returns the tool descriptions the MCP server registers: the core descriptions plus the guidance preamble (see Where the guidance goes). With slots, get_memory_index starts with slots.purpose. wenchang.mcp.MAX_TOOL_DESCRIPTION_CHARS is 2000; build_server and register_memory_tools raise ValueError at startup if a description exceeds it. register_memory_tools takes an optional slots keyword for this. The prompt is markdown with ## headings and no H1; see Customizing the prompts for its sections.

Errors#

Every failure renders through render_error. Over MCP it is the payload of an error result (isError true); without MCP you call render_error yourself. The payload has this shape:

{
  "error": "VersionConflictError",
  "category": "recoverable",
  "message": "...",
  "scope": "...",
  "area": "...",
  "name": "...",
  "content": "...",
  "version": "..."
}

error is the class name, category is recoverable, permanent, transient, or internal, and message combines the detail with fixed guidance for the category. Extra fields appear only when relevant. When the location parses, the payload carries scope, area, and name, and the message names it as scope user, area preferences, name csv. Errors never include a storage path.

InvalidArgumentError#

wenchang.errors.InvalidArgumentError(argument, detail) names the rejected argument. argument must be a non-empty str (TypeError or ValueError otherwise). The message is Argument {argument} is invalid: {detail} (Argument {argument} is invalid when detail is blank), ended with a period unless it already ends in ., !, or ?, followed by the recoverable guidance. It renders as error, category (recoverable), message, and argument, with no location fields:

{
  "error": "InvalidArgumentError",
  "category": "recoverable",
  "message": "Argument scope is invalid: scope 'project' is not available in this session; available scopes: organization, user. ...",
  "argument": "scope"
}

The tools raise it with argument set to one of their own parameter names: scope, area, name, content, description, aliases, line, old_string, new_string, expected_version, or cursor. You can raise it from your own dispatch code too, as the direct example does for a missing or unknown argument.

Notes:

• The tool layer builds each path from the caller's own grants, so over MCP an ungranted scope is reported as InvalidArgumentError (naming the available scopes); not_granted is raised by the core path check.
• MCP validates arguments before wenchang sees them. A wrong-type or missing argument returns MCP's own error text, not a rendered wenchang error.
• MCP pre-parses JSON-like strings: the string "null" arrives as None for a nullable parameter. For example list_prefix(area="null") lists the whole scope.

Configuration#

Identity types#

wenchang.identity.Identity(grants) is what a resolver returns: a mapping from scope name to ScopeGrant(entity_id, role). Scope names and entity IDs must be valid path segments (non-empty, no /, no backslash, no control characters, not . or ..). The role must be non-empty. identity.scope_map is scope name to entity ID. identity.entity_id(scope) and identity.role(scope) raise KeyError if the scope is not granted.

IdentityResolver[C] is a protocol with one method, resolve(credentials: C) -> Identity | ResolutionFailure. It must not raise, and the detail of a ResolutionFailure is shown to the agent. resolve must be a plain synchronous method: it is never awaited. If it is detectably async (an async def, an async generator function, an async staticmethod, a functools.partial of an async function, or a callable object with async def __call__), register_memory_tools, build_server, and bind_tools raise TypeError("resolver.resolve must be a plain method, not async"). The first two raise at registration, before anything is registered; bind_tools raises before the resolver runs. A sync def that returns a coroutine is not detected, and every call fails with ResolverFailureError. It runs on every tool call over MCP, and once per bind_tools call without MCP. SandboxResolver(identity) returns one fixed identity for any credentials, for sandboxes and tests.

wenchang.scope.ScopePolicy(write_roles) maps a scope to the frozenset of roles allowed to write it. A scope absent from the mapping is unrestricted. policy.permitted_roles(scope) returns the set or None.

Store options#

wenchang.core.MemoryStore(storage, *, max_file_bytes=16384, clock=..., list_page_size=100, index_max_bytes=65536, scope_priority=(), max_files_per_entity=500, index_cache=None). The three size options must be positive. max_files_per_entity caps the non-system/ files per entity and must be positive or None (no cap); see Files per entity. index_cache is an optional IndexCache; see The index cache. scope_priority entries must be distinct valid path segments, and the scopes it lists are kept first when the startup index is capped; unlisted scopes rank after them, and system/ areas always come first. clock returns timezone-aware datetimes and stamps last_updated; pass a fixed one in tests. A write over max_file_bytes raises OversizeWriteError.

from datetime import UTC, datetime

from wenchang.core import MemoryStore
from wenchang.storage.memory import InMemoryStorage
from wenchang.transport import InProcessClient

store = MemoryStore(
    InMemoryStorage(),
    max_file_bytes=8 * 1024,  # default 16384: larger writes raise OversizeWriteError
    index_max_bytes=32 * 1024,  # default 65536: the startup index is capped at this size
    list_page_size=50,  # default 100: entries per list_prefix page
    max_files_per_entity=200,  # default 500 (None: no cap): at the cap, creates fail
    scope_priority=("user", "project", "organization"),  # which scopes the capped index keeps first
    clock=lambda: datetime.now(UTC),  # must return timezone-aware datetimes
)
client = InProcessClient(store)
print(store.index_max_bytes, store.scope_priority)

Storage backends#

wenchang.storage.Storage is the only way the rest of wenchang touches bytes. It is a protocol over objects with a string key, bytes, a flat string metadata map, and an opaque version token:

Backends may raise BackendUnavailableError for timeouts and outages. Three ship with the library:

• GcsStorage(bucket) in wenchang.storage.gcs takes a google.cloud.storage.Bucket. One bucket is one storage instance. Metadata is GCS custom object metadata, the version is the object's generation, and conditional writes use generation preconditions. The constructor builds the bucket client's HTTP session (including any mTLS channel setup), so mTLS configuration errors raise from GcsStorage(bucket) rather than from the first call. It makes no network call.
• InMemoryStorage() in wenchang.storage.memory is a fake for tests and trials. It holds a lock, so it is safe to share across threads.
• FilesystemStorage(root) in wenchang.storage.filesystem (not re-exported from wenchang.storage) keeps objects in a local directory, for tests and local development rather than production. root is a Path, created if missing. POSIX and a local filesystem only. Each key segment is percent-encoded, with uppercase escaped, and directories mirror the key's segments. A key a/b/c.md is stored as a/b/c.md%obj, the exact body, plus a/b/c.md%meta, a JSON sidecar holding the version and metadata. Versions are random hex strings. It is safe to share across threads, and across instances and processes on one root, through a lock on root/.lock. Accepted limits: a crash between replacing the body and the sidecar can pair a new body with old metadata; a hand-edited body is visible on read but keeps its old version; nothing calls fsync; and stray .tmp- files are not cleaned up. An empty key, or a key segment that encodes to more than 200 bytes or is not valid UTF-8 (such as a lone surrogate), raises InvalidStorageKeyError.

Object keys, the stored metadata keys, timeouts, and retries are on Storage and operations.

Operator methods#

Not agent tools, and not on TransportClient or MemoryTools. See Erasing an entity.

• MemoryStore.purge(scope, entity_id, *, max_passes=PURGE_MAX_PASSES) -> int: delete every object under {scope}/{entity_id}/; returns the confirmed delete count. wenchang.core.PURGE_MAX_PASSES is 5. Raises wenchang.core.PurgeIncompleteError if objects remain after the last pass.
• MemoryStore.export(scope, entity_id) -> tuple[ExportedObject, ...]: every object under that prefix, raw, in key order. ExportedObject has key, data, metadata, and version.
• MemoryStore.invalidate_index(scope, entity_id) -> None: force the next index that includes the entity to list storage again. Does nothing without an index cache.

Index cache#

wenchang.index_cache.IndexCache(get, set, ttl=None, key_prefix="wenchang:index:"), where get(key) -> str | None and set(key, value, ttl) -> None. InMemoryIndexCache() provides both for one process. See The index cache.

Transport#

The tool layer calls a wenchang.transport.TransportClient, a protocol that mirrors MemoryStore's seven operations. InProcessClient(store) calls a MemoryStore directly and is the only client shipped. The MCP adapter calls one client from concurrent worker threads. What is safe to share across threads is under Threads.

Telemetry#

An on_call observer receives a wenchang.observer.ToolCallRecord (re-exported by wenchang.mcp) after each memory tool call. How to use it on each path: new server, existing server, without MCP.

arguments and result are read-only mappings (MappingProxyType) built from deep copies, so an observer can't change what the tool or client sees.

on_call must be a plain (non-async) function. An exception it raises is logged at ERROR and swallowed.

Conformance suites#

pip install "wenchang[testing] @ git+https://github.com/msiebert/wenchang@v0.1.1" adds pytest and two mixins in wenchang.testing:

• ResolverConformance[C] checks an IdentityResolver. Supply fixtures resolver, valid_credentials, invalid_credentials (call pytest.skip if your resolver accepts everything), policy, expected_scopes, and known_roles.
• TransportConformance checks a TransportClient. Supply fixtures client (fresh and empty per test), source, scope_map (at least two writable scopes), max_file_bytes (at least 64), index_max_bytes (at least 1024), scope_priority, and list_page_size. Within one test the client's writes must stamp strictly increasing last_updated values. An optional max_files_per_entity fixture (default None) returns the client's store max_files_per_entity (at least 32 and at least list_page_size + 2) to run the file-cap case, which checks that a create at the cap raises FileQuotaExceededError; with None that case is skipped.

Name your subclasses with a Test prefix. This example passes against the shipped InProcessClient:

from collections.abc import Mapping
from datetime import UTC, datetime, timedelta

import pytest

from wenchang.core import MemoryStore
from wenchang.identity import Identity, IdentityResolver, ResolutionFailure, ScopeGrant
from wenchang.scope import ScopePolicy
from wenchang.storage.memory import InMemoryStorage
from wenchang.testing import ResolverConformance, TransportConformance
from wenchang.transport import InProcessClient, TransportClient

POLICY = ScopePolicy({"organization": frozenset({"admin", "owner"})})


class MyResolver:
    def resolve(self, credentials: str) -> Identity | ResolutionFailure:
        if credentials != "tok-tianlong":
            return ResolutionFailure("unknown token")
        return Identity(
            grants={
                "user": ScopeGrant("u-tianlong", "owner"),
                "organization": ScopeGrant("o-hanlin-press", "admin"),
            }
        )


class TestMyResolver(ResolverConformance[str]):
    @pytest.fixture
    def resolver(self) -> IdentityResolver[str]:
        return MyResolver()

    @pytest.fixture
    def valid_credentials(self) -> str:
        return "tok-tianlong"

    @pytest.fixture
    def invalid_credentials(self) -> str:
        return "tok-nobody"

    @pytest.fixture
    def policy(self) -> ScopePolicy:
        return POLICY

    @pytest.fixture
    def expected_scopes(self) -> frozenset[str]:
        return frozenset({"user", "organization"})

    @pytest.fixture
    def known_roles(self) -> frozenset[str]:
        return frozenset({"owner", "admin", "member"})


class _Ticking:
    """A clock whose every call is strictly later than the last."""

    def __init__(self) -> None:
        self._n = 0

    def __call__(self) -> datetime:
        self._n += 1
        return datetime(2024, 1, 1, tzinfo=UTC) + timedelta(microseconds=self._n)


class TestMyClient(TransportConformance):
    @pytest.fixture
    def client(self) -> TransportClient:
        return InProcessClient(
            MemoryStore(
                InMemoryStorage(),
                clock=_Ticking(),
                max_file_bytes=256,
                index_max_bytes=4096,
                scope_priority=("user", "organization"),
                list_page_size=2,
            )
        )

    @pytest.fixture
    def source(self) -> str:
        return "conformance"

    @pytest.fixture
    def scope_map(self) -> Mapping[str, str]:
        return {"user": "u-tianlong", "organization": "o-hanlin-press"}

    @pytest.fixture
    def max_file_bytes(self) -> int:
        return 256

    @pytest.fixture
    def index_max_bytes(self) -> int:
        return 4096

    @pytest.fixture
    def scope_priority(self) -> tuple[str, ...]:
        return ("user", "organization")

    @pytest.fixture
    def list_page_size(self) -> int:
        return 2

Known limitations are on Status and compatibility.