Quickstart#

pip install "wenchang[mcp] @ git+https://github.com/msiebert/wenchang@v0.1.1"

gcloud storage buckets create gs://hanlin-inkwell-memory --project my-gcp-project --location us-central1 --uniform-bucket-level-access
gcloud auth application-default login
gcloud storage buckets add-iam-policy-binding gs://hanlin-inkwell-memory --member user:tianlong@example.com --role roles/storage.objectUser

import os

from google.cloud import storage
from mcp.server.mcpserver import Context

from wenchang.core import MemoryStore
from wenchang.identity import Identity, SandboxResolver, ScopeGrant
from wenchang.mcp import build_server
from wenchang.prompts import PromptSlots
from wenchang.scope import ScopePolicy
from wenchang.storage.gcs import GcsStorage
from wenchang.transport import InProcessClient

slots = PromptSlots(
    purpose=(
        "You have persistent memory of your work helping an author write a novel: reference it "
        "and save to it with these tools whenever you work on a manuscript."
    ),
    scope_guidance=(
        "There are three scopes. `user` is private to the author you are talking with. "
        "`project` is the manuscript, shared by everyone working on it; any of them may write it. "
        "`organization` is Hanlin Press, shared by every author and editor there; only "
        "editors-in-chief may write it. Writes to `user` and `project` need no ask. If a new fact "
        "contradicts one stored in a shared scope, show the stored fact and ask which to keep."
    ),
    seed_areas=(
        "- user: preferences, workflows\n"
        "- project: characters, plot, timeline, style\n"
        "- organization: conventions\n\n"
        "Every scope also has a `system/` area."
    ),
)

# Each person sets their own WENCHANG_USER. Everyone on a manuscript shares WENCHANG_PROJECT.
identity = Identity(
    grants={
        "user": ScopeGrant(os.environ["WENCHANG_USER"], "owner"),
        "project": ScopeGrant(os.environ["WENCHANG_PROJECT"], "member"),
        "organization": ScopeGrant(os.environ["WENCHANG_ORG"], "member"),
    }
)

# Application Default Credentials; the bucket must already exist.
bucket = storage.Client(project="my-gcp-project").bucket(os.environ["WENCHANG_BUCKET"])
store = MemoryStore(GcsStorage(bucket), scope_priority=("user", "project", "organization"))


def credentials_from_context(ctx: Context) -> None:
    return None  # SandboxResolver ignores credentials.


server = build_server(
    slots=slots,
    client=InProcessClient(store),
    resolver=SandboxResolver(identity),
    # Only editors-in-chief (admins) write organization.
    policy=ScopePolicy({"organization": frozenset({"admin", "owner"})}),
    credentials_from_context=credentials_from_context,
    source="claude-code",
    product="Inkwell",
)

if __name__ == "__main__":
    server.run("stdio")

WENCHANG_USER=u-tianlong WENCHANG_PROJECT=p-ninth-examination WENCHANG_ORG=o-hanlin-press WENCHANG_BUCKET=hanlin-inkwell-memory python -c "import server; print(server.server)"

{
  "mcpServers": {
    "memory": {
      "command": "/path/to/venv/bin/python",
      "args": ["/path/to/server.py"],
      "env": {
        "WENCHANG_BUCKET": "hanlin-inkwell-memory",
        "WENCHANG_ORG": "o-hanlin-press",
        "WENCHANG_PROJECT": "p-ninth-examination",
        "WENCHANG_USER": "u-tianlong"
      }
    }
  }
}

claude mcp add memory -e WENCHANG_BUCKET=hanlin-inkwell-memory -e WENCHANG_ORG=o-hanlin-press -e WENCHANG_PROJECT=p-ninth-examination -e WENCHANG_USER=u-tianlong -- /path/to/venv/bin/python /path/to/server.py

gcloud storage ls --recursive gs://hanlin-inkwell-memory/project/p-ninth-examination/

import os
from datetime import UTC, datetime

from google.cloud import storage

from wenchang.core import MemoryStore
from wenchang.file_format import FileMetadata
from wenchang.paths import build_path
from wenchang.storage.gcs import GcsStorage
from wenchang.transport import InProcessClient

bucket = storage.Client(project="my-gcp-project").bucket(os.environ["WENCHANG_BUCKET"])
client = InProcessClient(MemoryStore(GcsStorage(bucket)))

# Goes through the store directly, so it may write system/.
path = build_path("organization", "o-hanlin-press", "system", "house-style")
client.write_file(
    path,
    "- [system] Dialogue uses double quotation marks.\n"
    "- [system] Spelling follows British English.\n",
    FileMetadata(
        description="Hanlin Press house style",
        aliases=("style guide", "house style"),
        sources=frozenset(),
        last_updated=datetime.now(UTC),
    ),
    None,  # create; pass the current version to replace
    source="seed-job",
)

import tempfile
from datetime import UTC, datetime
from pathlib import Path

from wenchang.core import MemoryStore
from wenchang.file_format import FileMetadata
from wenchang.paths import build_path
from wenchang.storage.filesystem import FilesystemStorage

with tempfile.TemporaryDirectory() as tmp:
    root = Path(tmp) / "memory"  # created if missing; use a persistent path in practice
    store = MemoryStore(FilesystemStorage(root))

    path = build_path("user", "u-tianlong", "preferences", "prose-style")
    store.write_file(
        path,
        "- [stated] Writes in close third person, past tense.\n",
        FileMetadata(
            description="Prose style preferences",
            aliases=("narration",),
            sources=frozenset(),
            last_updated=datetime.now(UTC),
        ),
        None,  # create
        source="local-dev",
    )

    memory = store.read_file(path)
    assert memory.content == "- [stated] Writes in close third person, past tense.\n"

    # Each file is a plain body object beside a metadata sidecar.
    [body] = root.rglob("*%obj")
    assert body.read_text() == memory.content
    print(body.relative_to(root))

docker run -d --rm --name wenchang-fake-gcs -p 4443:4443 fsouza/fake-gcs-server -scheme http -port 4443

from google.auth.credentials import AnonymousCredentials
from google.cloud import storage

client = storage.Client(
    project="test",
    credentials=AnonymousCredentials(),
    client_options={"api_endpoint": "http://localhost:4443"},
)
bucket = client.bucket("wenchang-dev")
if not bucket.exists():
    bucket = client.create_bucket("wenchang-dev")